91 lines
3.2 KiB
Python
91 lines
3.2 KiB
Python
import requests
|
||
import os
|
||
from datetime import datetime
|
||
|
||
|
||
class DouyuYubaMonitor:
|
||
def __init__(self, hash_id):
|
||
self.hash_id = hash_id
|
||
self.api_url = "https://yuba.douyu.com/wgapi/yubanc/api/feed/getUserFeedList"
|
||
self.record_file = f"last_processed_{hash_id}.txt"
|
||
self.headers = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
"Referer": f"https://yuba.douyu.com/member/{hash_id}/main/news",
|
||
}
|
||
|
||
def get_last_id(self):
|
||
if os.path.exists(self.record_file):
|
||
with open(self.record_file, "r") as f:
|
||
return f.read().strip()
|
||
return None
|
||
|
||
def save_last_id(self, feed_id):
|
||
with open(self.record_file, "w") as f:
|
||
f.write(str(feed_id))
|
||
|
||
def fetch_and_process(self):
|
||
params = {
|
||
"filter_type": 1,
|
||
"hash_id": self.hash_id,
|
||
"limit": 10,
|
||
"offset": 0
|
||
}
|
||
|
||
try:
|
||
response = requests.get(self.api_url, headers=self.headers, params=params, timeout=10)
|
||
data = response.json()
|
||
|
||
if data.get("error") != 0:
|
||
print(f"API Error: {data.get('msg')}")
|
||
return
|
||
|
||
feed_list = data.get("data", {}).get("feed_list", [])
|
||
|
||
# 查找第一条【非置顶】动态
|
||
target_feed = None
|
||
for feed in feed_list:
|
||
if feed.get("home_feed_top") == 1:
|
||
continue
|
||
target_feed = feed
|
||
break
|
||
|
||
if not target_feed:
|
||
print("未发现有效动态（可能全是置顶）。")
|
||
return
|
||
|
||
feed_id = str(target_feed.get("feed_id"))
|
||
last_id = self.get_last_id()
|
||
|
||
# 标记检查
|
||
if feed_id == last_id:
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] 监控中，暂无最新非置顶消息。")
|
||
return
|
||
|
||
# 提取参数拼接精准链接
|
||
# 根据你提供的样本：origin 固定为 9, scode 可以从 publisher 逻辑或固定获取（通常 scode 是动态的，这里采用 feed 中的核心参数）
|
||
group_id = target_feed.get("group", {}).get("group_id", "0")
|
||
|
||
# 斗鱼 Web 端scode通常在分享链接中生成，这里优先匹配你给出的格式
|
||
# 如果接口没直接给 scode，可以根据业务需求固定或留空，这里保留你给出的示例参数
|
||
|
||
full_url = f"https://yuba.douyu.com/feed/{feed_id}"
|
||
|
||
# 输出结果
|
||
print("\n" + "★" * 50)
|
||
print(f"【提取到最新动态】")
|
||
print(f"发布时间: {datetime.fromtimestamp(int(target_feed.get('ctime'))).strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"正文内容: {target_feed.get('text', '')[:200]}")
|
||
print(f"精准链接: {full_url}")
|
||
print("★" * 50 + "\n")
|
||
|
||
# 保存标记
|
||
self.save_last_id(feed_id)
|
||
|
||
except Exception as e:
|
||
print(f"运行异常: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 使用你提供的 hash_id
|
||
monitor = DouyuYubaMonitor("PDAP2zEk3nwx")
|
||
monitor.fetch_and_process() |