mikan_watcher.py
11.5 KB| 1 | # -*- coding: utf-8 -*- |
| 2 | """蜜柑计划（mikanime.tv）番剧更新监测（通用版） |
| 3 | |
| 4 | 用法： |
| 5 | python3 mikan_watcher.py --check # 巡检预设番剧，发现新条目推送 QQ |
| 6 | python3 mikan_watcher.py --search "番剧名" # 即时搜索某番剧最新带中文字幕磁力并推送 |
| 7 | python3 mikan_watcher.py --list # 列出当前预设番剧 |
| 8 | python3 mikan_watcher.py --add "番剧名" # 添加预设番剧 |
| 9 | python3 mikan_watcher.py --remove "番剧名" # 移除预设番剧 |
| 10 | |
| 11 | 机制：蜜柑官方 RSS 接口 https://mikanime.tv/RSS/Search?searchstr=<关键词> |
| 12 | 返回最新条目（标题/guid/链接/pubDate/torrent）；episode 页面含完整 magnet。 |
| 13 | 带中文字幕过滤：标题含 简体/简繁/双语/中文/繁简 等关键词优先。 |
| 14 | """ |
| 15 | import argparse |
| 16 | import json |
| 17 | import os |
| 18 | import re |
| 19 | import subprocess |
| 20 | import sys |
| 21 | import time |
| 22 | import urllib.parse |
| 23 | |
| 24 | BASE = os.path.dirname(os.path.abspath(__file__)) |
| 25 | CONFIG_FILE = os.path.join(BASE, "mikan_watch_config.json") |
| 26 | STATE_FILE = os.path.join(BASE, "mikan_watch_state.json") |
| 27 | UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" |
| 28 | RSS_URL = "https://mikanime.tv/RSS/Search?searchstr={}" |
| 29 | |
| 30 | # 中文字幕关键词（优先命中）与排除词 |
| 31 | SUB_CN_KEYWORDS = ["简体", "简繁", "繁简", "双语", "中文", "字幕"] |
| 32 | SUB_EXCLUDE = ["生肉", "raw", "RAW", "无字幕", "英字", "繁日"] |
| 33 | |
| 34 | # 可选推送：设置以下环境变量后会把结果推送到 OneBot 兼容网关；未设置则只输出到终端 |
| 35 | QQ_USER_ID = os.environ.get("QQ_USER_ID", "") |
| 36 | ONE_BOT_URL = os.environ.get("ONE_BOT_URL", "http://127.0.0.1:3000/send_private_msg") |
| 37 | ONE_BOT_TOKEN = os.environ.get("ONE_BOT_TOKEN", "") |
| 38 | |
| 39 | |
| 40 | def load_json(path, default): |
| 41 | if os.path.exists(path): |
| 42 | try: |
| 43 | with open(path, encoding="utf-8") as f: |
| 44 | return json.load(f) |
| 45 | except Exception: |
| 46 | pass |
| 47 | return default |
| 48 | |
| 49 | |
| 50 | def save_json(path, data): |
| 51 | with open(path, "w", encoding="utf-8") as f: |
| 52 | json.dump(data, f, ensure_ascii=False, indent=2) |
| 53 | |
| 54 | |
| 55 | def get_config(): |
| 56 | return load_json(CONFIG_FILE, {"shows": []}) |
| 57 | |
| 58 | |
| 59 | def get_state(): |
| 60 | return load_json(STATE_FILE, {"seen": {}}) |
| 61 | |
| 62 | |
| 63 | def http_get(url): |
| 64 | try: |
| 65 | r = subprocess.run( |
| 66 | ["curl", "-s", "--max-time", "20", "-A", UA, url], |
| 67 | capture_output=True, text=True, timeout=25, |
| 68 | ) |
| 69 | return r.stdout |
| 70 | except Exception: |
| 71 | return "" |
| 72 | |
| 73 | |
| 74 | def parse_rss(xml): |
| 75 | """解析蜜柑 RSS XML → 条目列表 [{title,guid,link,pub_date}] 按 pubDate 倒序""" |
| 76 | items = re.findall(r"<item>(.*?)</item>", xml, flags=re.S) |
| 77 | out = [] |
| 78 | for it in items: |
| 79 | def g(tag): |
| 80 | m = re.search(r"<%s[^>]*>(.*?)</%s>" % (tag, tag), it, flags=re.S) |
| 81 | return m.group(1).strip() if m else "" |
| 82 | title = g("title") |
| 83 | guid = g("guid") |
| 84 | link = g("link") |
| 85 | pub = g("pubDate") |
| 86 | if title and link: |
| 87 | out.append({"title": title, "guid": guid, "link": link, "pub_date": pub}) |
| 88 | out.sort(key=lambda x: x["pub_date"], reverse=True) |
| 89 | return out |
| 90 | |
| 91 | |
| 92 | def extract_episode(title): |
| 93 | """从标题提取集数（排除分辨率干扰）；合集取最大集；提取失败返回 None""" |
| 94 | t = re.sub(r"\b(1080|720|1920|2160|4060)[pPiI]?\b", "", title) |
| 95 | eps = [] |
| 96 | m = re.findall(r"-\s*(\d{1,2})(?:v\d)?(?=\s|\(|\[|$)", t) |
| 97 | eps += [int(x) for x in m] |
| 98 | m = re.findall(r"\[(\d{1,2})(?:v\d)?\]", t) |
| 99 | eps += [int(x) for x in m] |
| 100 | m = re.findall(r"(?:EP|Ep\.?|第)\s*(\d{1,2})\s*(?:话|話)?", t) |
| 101 | eps += [int(x) for x in m] |
| 102 | m = re.findall(r"(\d{1,2})\s*-\s*(\d{1,2})", t) |
| 103 | if m: |
| 104 | eps += [int(m[0][1])] |
| 105 | if not eps: |
| 106 | return None |
| 107 | return max(eps) |
| 108 | |
| 109 | |
| 110 | def cn_subtitle_rank(title): |
| 111 | """中文字幕匹配度：命中关键词越多越靠前；命中排除词直接淘汰""" |
| 112 | low = title |
| 113 | for ex in SUB_EXCLUDE: |
| 114 | if ex in low: |
| 115 | return -1 |
| 116 | score = 0 |
| 117 | for k in SUB_CN_KEYWORDS: |
| 118 | if k in low: |
| 119 | score += 1 |
| 120 | return score |
| 121 | |
| 122 | |
| 123 | def quality_score(title): |
| 124 | """综合质量分：字幕分优先，再叠分辨率/内封/大小写标点等细节，用于选最优磁力""" |
| 125 | s = cn_subtitle_rank(title) |
| 126 | if s < 0: |
| 127 | return -1 |
| 128 | # 分辨率：1080p 最好，720 次之 |
| 129 | if "1080" in title: |
| 130 | s += 2 |
| 131 | elif "720" in title: |
| 132 | s += 1 |
| 133 | # 简中优先（简体/简繁/双语），纯繁中略降 |
| 134 | if "简体" in title or "简繁" in title or "双语" in title: |
| 135 | s += 1 |
| 136 | elif "CHT" in title or "繁体" in title or "繁日" in title: |
| 137 | s -= 0.5 |
| 138 | # 内封/WebRip 加分 |
| 139 | if "内封" in title or "内嵌" in title: |
| 140 | s += 0.5 |
| 141 | # 招募翻译、SP 这种凑数的降分 |
| 142 | if "招募" in title or "SP" in title or "总集篇" in title: |
| 143 | s -= 1 |
| 144 | return s |
| 145 | |
| 146 | |
| 147 | def pick_best(items, top_n=3): |
| 148 | """选出带中文字幕的最新条目（过滤后按综合质量分排序，取 top_n）""" |
| 149 | scored = [it for it in items if cn_subtitle_rank(it["title"]) >= 0] |
| 150 | scored.sort(key=lambda it: (quality_score(it["title"]), it["pub_date"]), reverse=True) |
| 151 | return scored[:top_n] |
| 152 | |
| 153 | |
| 154 | def fetch_magnet(episode_url): |
| 155 | """从 episode 页面抓完整 magnet 链接""" |
| 156 | html = http_get(episode_url) |
| 157 | m = re.search(r'magnet:\?[^"\'<\\]*', html) |
| 158 | if m: |
| 159 | return m.group(0).replace("&", "&") |
| 160 | # 兜底：episode 页 id 就是 btih，构造无 tracker 的 magnet |
| 161 | m2 = re.search(r"/Home/Episode/([0-9a-f]{40})", episode_url) |
| 162 | if m2: |
| 163 | return "magnet:?xt=urn:btih:%s" % m2.group(1) |
| 164 | return None |
| 165 | |
| 166 | |
| 167 | def send_qq(text): |
| 168 | """通过 OneBot HTTP 发私聊；未配置 ONE_BOT_TOKEN 时只提示不发送""" |
| 169 | if not ONE_BOT_TOKEN or not QQ_USER_ID: |
| 170 | return "(通知未配置：ONE_BOT_TOKEN / QQ_USER_ID 未设置，仅打印)" |
| 171 | payload = {"user_id": str(QQ_USER_ID), "message": text} |
| 172 | try: |
| 173 | r = subprocess.run( |
| 174 | ["curl", "-s", "--max-time", "15", "-X", "POST", ONE_BOT_URL, |
| 175 | "-H", "Authorization: Bearer %s" % ONE_BOT_TOKEN, |
| 176 | "-H", "Content-Type: application/json", |
| 177 | "-d", json.dumps(payload, ensure_ascii=False)], |
| 178 | capture_output=True, text=True, timeout=20, |
| 179 | ) |
| 180 | return r.stdout |
| 181 | except Exception as exc: |
| 182 | return "ERR: %s" % exc |
| 183 | |
| 184 | |
| 185 | def search_show(keyword): |
| 186 | """搜索番剧 → 返回「最新一集」的带中文字幕条目（最多 2 条）""" |
| 187 | xml = http_get(RSS_URL.format(urllib.parse.quote(keyword))) |
| 188 | if not xml: |
| 189 | return [] |
| 190 | items = parse_rss(xml) |
| 191 | scored = [] |
| 192 | for it in items: |
| 193 | if cn_subtitle_rank(it["title"]) < 0: |
| 194 | continue |
| 195 | ep = extract_episode(it["title"]) |
| 196 | if ep is None: |
| 197 | continue |
| 198 | scored.append((ep, it)) |
| 199 | if not scored: |
| 200 | return [] |
| 201 | max_ep = max(ep for ep, _ in scored) |
| 202 | cands = [it for ep, it in scored if ep == max_ep] |
| 203 | cands.sort(key=lambda it: (quality_score(it["title"]), it["pub_date"]), reverse=True) |
| 204 | return cands[:2] |
| 205 | |
| 206 | |
| 207 | def check_show(keyword, state, notify=True): |
| 208 | """巡检单个番剧：按集数判断最新集。 |
| 209 | |
| 210 | 机制：提取 RSS 所有条目的集数 → 找当前已发布的最大集数 max_ep； |
| 211 | 与 state 记录的已确认集数对比，若出现更高集（任一字幕组）→ 视为新更新。 |
| 212 | 返回 (新条目列表, 当前最大集数)。 |
| 213 | """ |
| 214 | xml = http_get(RSS_URL.format(urllib.parse.quote(keyword))) |
| 215 | if not xml: |
| 216 | return [], None |
| 217 | items = parse_rss(xml) |
| 218 | scored = [] |
| 219 | for it in items: |
| 220 | if cn_subtitle_rank(it["title"]) < 0: |
| 221 | continue # 生肉/无字幕 淘汰 |
| 222 | ep = extract_episode(it["title"]) |
| 223 | if ep is None: |
| 224 | continue |
| 225 | scored.append((ep, it)) |
| 226 | if not scored: |
| 227 | return [], None |
| 228 | max_ep = max(ep for ep, _ in scored) |
| 229 | # state: {"seen": {kw: {"ep": 已确认集数, "guids": [已推 guid]}}} |
| 230 | rec = state.setdefault("seen", {}).setdefault(keyword, {"ep": 0, "guids": []}) |
| 231 | if max_ep <= rec["ep"]: |
| 232 | return [], max_ep # 没有更高集 |
| 233 | # 出现更高集：收集该集所有带字幕版本（按综合质量分排序），去重 guid，最多推 2 条 |
| 234 | cands = [it for ep, it in scored if ep == max_ep] |
| 235 | cands.sort(key=lambda it: (quality_score(it["title"]), it["pub_date"]), reverse=True) |
| 236 | new = [] |
| 237 | for it in cands: |
| 238 | if it["guid"] not in rec["guids"] and len(new) < 2: |
| 239 | new.append(it) |
| 240 | rec["guids"].append(it["guid"]) |
| 241 | rec["ep"] = max_ep |
| 242 | # 只保留最近 30 个 guid 防膨胀 |
| 243 | rec["guids"] = rec["guids"][-30:] |
| 244 | return new, max_ep |
| 245 | |
| 246 | |
| 247 | def format_report(keyword, entries): |
| 248 | """拼 QQ 推送文本""" |
| 249 | if not entries: |
| 250 | return "（%s 暂无新更新）" % keyword |
| 251 | lines = ["【%s】更新啦！" % keyword] |
| 252 | for e in entries: |
| 253 | lines.append("・%s" % e["title"]) |
| 254 | lines.append(" 磁力：%s" % e["magnet"]) |
| 255 | return "\n".join(lines) |
| 256 | |
| 257 | |
| 258 | def main(): |
| 259 | p = argparse.ArgumentParser(description="蜜柑计划番剧监测") |
| 260 | p.add_argument("--check", action="store_true", help="巡检预设番剧") |
| 261 | p.add_argument("--search", metavar="番剧名", help="即时搜索") |
| 262 | p.add_argument("--list", action="store_true", help="列出预设番剧") |
| 263 | p.add_argument("--add", metavar="番剧名", help="添加预设番剧") |
| 264 | p.add_argument("--remove", metavar="番剧名", help="移除预设番剧") |
| 265 | p.add_argument("--dry-run", action="store_true", help="只打印不推送 QQ") |
| 266 | a = p.parse_args() |
| 267 | |
| 268 | cfg = get_config() |
| 269 | state = get_state() |
| 270 | |
| 271 | if a.add: |
| 272 | name = a.add.strip() |
| 273 | if name not in cfg["shows"]: |
| 274 | cfg["shows"].append(name) |
| 275 | save_json(CONFIG_FILE, cfg) |
| 276 | print("已添加预设番剧：%s（当前 %d 部）" % (name, len(cfg["shows"]))) |
| 277 | return |
| 278 | |
| 279 | if a.remove: |
| 280 | name = a.remove.strip() |
| 281 | if name in cfg["shows"]: |
| 282 | cfg["shows"].remove(name) |
| 283 | save_json(CONFIG_FILE, cfg) |
| 284 | print("已移除：%s（当前 %d 部）" % (name, len(cfg["shows"]))) |
| 285 | return |
| 286 | |
| 287 | if a.list: |
| 288 | if not cfg["shows"]: |
| 289 | print("（暂无预设番剧，用 --add 添加）") |
| 290 | for s in cfg["shows"]: |
| 291 | print("・%s" % s) |
| 292 | return |
| 293 | |
| 294 | if a.search: |
| 295 | kw = a.search.strip() |
| 296 | entries = search_show(kw) |
| 297 | if not entries: |
| 298 | msg = "「%s」暂时没找到带中文字幕的更新，换个别称试试？" % kw |
| 299 | print(msg) |
| 300 | if not a.dry_run: |
| 301 | send_qq(msg) |
| 302 | return |
| 303 | parts = [] |
| 304 | for e in entries: |
| 305 | m = fetch_magnet(e["link"]) |
| 306 | e["magnet"] = m or "（未取到磁力链接）" |
| 307 | parts.append("・%s\n 磁力：%s" % (e["title"], e["magnet"])) |
| 308 | msg = "【%s】最新带中文字幕：\n%s" % (kw, "\n".join(parts)) |
| 309 | print(msg) |
| 310 | if not a.dry_run: |
| 311 | send_qq(msg) |
| 312 | return |
| 313 | |
| 314 | if a.check: |
| 315 | if not cfg["shows"]: |
| 316 | print("(no shows configured)") |
| 317 | return |
| 318 | for kw in cfg["shows"]: |
| 319 | new, max_ep = check_show(kw, state) |
| 320 | if new: |
| 321 | for e in new: |
| 322 | e["magnet"] = fetch_magnet(e["link"]) or "（未取到磁力链接）" |
| 323 | msg = format_report(kw, new) |
| 324 | print(msg) |
| 325 | if not a.dry_run: |
| 326 | send_qq(msg) |
| 327 | else: |
| 328 | print("(%s: no new, latest ep=%s)" % (kw, max_ep)) |
| 329 | save_json(STATE_FILE, state) |
| 330 | return |
| 331 | |
| 332 | p.print_help() |
| 333 | |
| 334 | |
| 335 | if __name__ == "__main__": |
| 336 | main() |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
