#!/usr/bin/env python3 """用 MoviePilot 的实现生成差分测试的期望输出(只读调用上游,不复制源码)。 用法(必须用 MP 自己的 venv 解释器,工作目录无要求): MP_ROOT=/opt/moviepilot-plus/upstream/MoviePilot \ $MP_ROOT/.venv/bin/python scripts/mp-sync/gen-differential.py meta ... gen-differential.py filter # 过滤规则 ... gen-differential.py rss # RSS / 站点索引解析 ... gen-differential.py all meta:读 packages/domain/test/differential/meta-samples.json,分别以 accelerator=None(Python 参考实现)与 accelerator=rust(moviepilot_rust 快路径) 调用 MetaInfo / MetaInfoPath,写 meta-mp.json。两条路径不一致的样本记在 rustMismatch。 生成后执行 `deno task fmt`(biome 统一 JSON 格式),再跑差分测试;新出现的差异需人工分类后 登记进对应 *-deviations.json / deviations 字段(kind 见 docs/porting/mp-rust-logic-parity.md)。 """ from __future__ import annotations import glob import hashlib import json import os import subprocess import sys import tempfile import time from pathlib import Path from types import SimpleNamespace from typing import Any, Optional REPO = Path(__file__).resolve().parents[2] MP_ROOT = Path(os.environ.setdefault("MP_ROOT", "/opt/moviepilot-plus/upstream/MoviePilot")) DOMAIN_DIFF = REPO / "packages/domain/test/differential" ADAPTERS_DIFF = REPO / "packages/adapters/test/differential" def _bootstrap_mp() -> str: """把 MP 放进 sys.path,隔离 CONFIG_DIR,返回上游提交号。""" os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="mp-diff-cfg-")) if str(MP_ROOT) not in sys.path: sys.path.insert(0, str(MP_ROOT)) os.chdir(MP_ROOT) try: return subprocess.check_output(["git", "-C", str(MP_ROOT), "rev-parse", "--short", "HEAD"], text=True).strip() except Exception: # noqa: BLE001 return "unknown" META_FIELDS = ( "name", "cn_name", "en_name", "year", "type", "begin_season", "end_season", "total_season", "begin_episode", "end_episode", "total_episode", "resource_pix", "resource_type", "resource_effect", "video_encode", "audio_encode", "resource_team", "edition", "part", "customization", "web_source", "fps", "video_bit", "media_source", "media_id", ) def _meta_to_dict(meta) -> dict: out = {} for key in META_FIELDS: value = getattr(meta, key, None) if key == "type": value = value.value if value is not None else None elif key == "media_source" and value is not None: value = getattr(value, "value", str(value)) if value == "": value = None out[key] = value out["class"] = type(meta).__name__ return out def gen_meta() -> None: commit = _bootstrap_mp() from pathlib import Path as P from app.adapters.system import rust from app.domain.meta.runtime import configure_recognition_runtime from app.domain.metainfo import MetaInfo, MetaInfoPath, clear_rust_parse_options_cache from app.runtime.config import settings samples = json.loads((DOMAIN_DIFF / "meta-samples.json").read_text("utf-8"))["samples"] def run(accelerator) -> dict: configure_recognition_runtime( media_extensions_provider=lambda: (*settings.RMT_MEDIAEXT, *settings.RMT_SUBEXT, *settings.RMT_AUDIOEXT), audio_extensions_provider=lambda: settings.RMT_AUDIOEXT, accelerator=accelerator, ) clear_rust_parse_options_cache() res = {} for s in samples: try: if s["kind"] == "path": meta = MetaInfoPath(P(s["input"])) else: meta = MetaInfo(title=s["input"], subtitle=s.get("subtitle")) res[s["id"]] = _meta_to_dict(meta) except Exception as err: # noqa: BLE001 res[s["id"]] = {"error": f"{type(err).__name__}: {err}"} return res python_out = run(None) rust_out = run(rust) if rust.is_available() else {} mismatch = {} for sid, py in python_out.items(): rs = rust_out.get(sid) if rs is None: continue diff = {k: {"python": py.get(k), "rust": rs.get(k)} for k in py if py.get(k) != rs.get(k)} if diff: mismatch[sid] = diff out = { "generatedFrom": {"repo": "MoviePilot", "commit": commit, "entry": "app.domain.metainfo.MetaInfo/MetaInfoPath", "rust": rust.status() if rust.is_available() else None}, "python": python_out, "rustMismatch": mismatch, } (DOMAIN_DIFF / "meta-mp.json").write_text(json.dumps(out, ensure_ascii=False, indent=1) + "\n", "utf-8") print(f"meta: {len(python_out)} samples, rust≠python: {len(mismatch)}") def gen_zhconv() -> None: """繁→简:moviepilot_rust.zhconv_fast 无 Python 回退(app/foundation/text.py 直接调用原生扩展)。""" commit = _bootstrap_mp() import moviepilot_rust samples = json.loads((DOMAIN_DIFF / "zhconv-samples.json").read_text("utf-8"))["samples"] out = { "generatedFrom": {"repo": "MoviePilot", "commit": commit, "entry": "moviepilot_rust.zhconv_fast(text, 'zh-hans')"}, "hans": {s: moviepilot_rust.zhconv_fast(s, "zh-hans") for s in samples}, } (DOMAIN_DIFF / "zhconv-mp.json").write_text(json.dumps(out, ensure_ascii=False, indent=1) + "\n", "utf-8") print(f"zhconv: {len(samples)} samples") # ---------------------------------------------------------------- filter:过滤规则 # 用 MP 的 Python 回退实现(FilterModule,rust_accel 关闭)对 TORRENTS x RULES 求匹配与优先级, # 并用 Rust filter_torrents_fast 交叉核对。无需数据库:规则组/规则集直接注入。 # 已知差异登记(deviations)不由本脚本生成,重新生成时从已有 filter-cases.json 保留。 GiB = 1024 ** 3 MiB = 1024 ** 2 # ---------------------------------------------------------------- 样本 # size 为站点报告的总大小(字节);labels 为站点标签 TORRENTS: list[dict[str, Any]] = [ {"id": "t01", "title": "Dune.Part.Two.2024.2160p.UHD.BluRay.Remux.DV.HDR.HEVC.TrueHD.Atmos.7.1-FraMeSToR", "description": "", "size": 70 * GiB, "seeders": 20}, {"id": "t02", "title": "Dune.Part.Two.2024.2160p.BluRay.x265.10bit.DV.HDR-CHD", "description": "沙丘2 中字 特效", "size": 25 * GiB, "seeders": 12}, {"id": "t03", "title": "Dune Part Two 2024 2160p UHD Blu-ray HEVC DV TrueHD 7.1-BeyondHD", "description": "", "size": 90 * GiB, "seeders": 4}, {"id": "t04", "title": "Dune.Part.Two.2024.1080p.WEB-DL.H264.DDP5.1-HHWEB", "description": "沙丘2 | 国语 中字", "size": 8 * GiB, "seeders": 50}, {"id": "t05", "title": "Dune.Part.Two.2024.1080p.BluRay.x264.DTS-HD.MA.5.1-WiKi", "description": "简繁英字幕", "size": 15 * GiB, "seeders": 9}, {"id": "t06", "title": "Dune.Part.Two.2024.720p.WEBRip.x264-GRP", "description": "", "size": 2 * GiB, "seeders": 1}, {"id": "t07", "title": "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos.DV.HDR.H.265-FLUX", "description": "", "size": 20 * GiB, "seeders": 30}, {"id": "t08", "title": "Movie.2022.2160p.HDR.HEVC-NOSRC", "description": "", "size": 18 * GiB, "seeders": 2}, {"id": "t09", "title": "The.Bear.S03.2160p.DSNP.WEB-DL.DDP5.1.HDR.H.265-NTb", "description": "熊家餐馆 第三季 全10集 中字", "size": 60 * GiB, "seeders": 15}, {"id": "t10", "title": "The.Bear.S03E01.1080p.WEB-DL.H264-NTb", "description": "", "size": int(2.5 * GiB), "seeders": 8}, {"id": "t11", "title": "Movie.2023.1080p.BluRay.REMUX.AVC.DTS-HD.MA.5.1-EPSiLON", "description": "", "size": 30 * GiB, "seeders": 6}, {"id": "t12", "title": "Movie 2023 1080p Blu-ray AVC DTS-HD MA 5.1-AAA", "description": "", "size": 40 * GiB, "seeders": 3}, {"id": "t13", "title": "Movie.2023.COMPLETE.BLURAY-GRP", "description": "", "size": 45 * GiB, "seeders": 2}, {"id": "t14", "title": "Movie.2023.1080i.HDTV.H264-GRP", "description": "", "size": 10 * GiB, "seeders": 1}, {"id": "t15", "title": "[Nekomoe kissaten] Frieren - 01 [WebRip 1080p HEVC-10bit AAC ASSx2].mkv", "description": "简繁日内封", "size": 1 * GiB, "seeders": 40}, {"id": "t16", "title": "The.Wandering.Earth.II.2023.2160p.WEB-DL.H265.HDR.DDP5.1-HDSWEB", "description": "流浪地球2 国语配音 中文字幕", "size": 12 * GiB, "seeders": 25}, {"id": "t17", "title": "Oppenheimer.2023.IMAX.2160p.BluRay.x265.10bit.HDR.DTS-HD.MA.5.1-SWTYBLZ", "description": "奥本海默 CHS/ENG", "size": 30 * GiB, "seeders": 7}, {"id": "t18", "title": "Movie.2022.3840x2160.WEB-DL.HDR10.HEVC-GRP", "description": "", "size": 15 * GiB, "seeders": 5}, {"id": "t19", "title": "Movie.2022.1920x1080.WEB-DL.H264-GRP", "description": "", "size": 6 * GiB, "seeders": 5}, {"id": "t20", "title": "Movie.2022.4K.WEB-DL.H265-GRP", "description": "", "size": 14 * GiB, "seeders": 5}, {"id": "t21", "title": "Movie.2022.1080p.WEB-DL.x264 [GB]", "description": "", "size": 4 * GiB, "seeders": 5}, {"id": "t22", "title": "Movie.2022.1080p.WEB-DL.H264.AAC 39.23 GB", "description": "", "size": 4 * GiB, "seeders": 5}, {"id": "t23", "title": "Some.Show.S01.1080p.BluRay.x264-GRP", "description": "全12集 简体中字", "size": 36 * GiB, "seeders": 3}, {"id": "t24", "title": "Movie.2021.DVDRip.x264-GRP", "description": "", "size": int(1.4 * GiB), "seeders": 2}, {"id": "t25", "title": "Movie 2021 1080p 3D BluRay Half-SBS x264-GRP", "description": "", "size": 8 * GiB, "seeders": 2}, {"id": "t26", "title": "Movie.2021.2160p.WEB-DL.60fps.HEVC-GRP", "description": "60帧 国语", "size": 10 * GiB, "seeders": 4}, {"id": "t27", "title": "Movie.2020.2160p.Remux.HEVC.DoVi-GRP", "description": "", "size": 50 * GiB, "seeders": 3}, {"id": "t28", "title": "Movie.2020.1080p.NF.WEB-DL.DDP5.1.H.264-CHDWEB", "description": "官方 中字", "labels": ["官方"], "size": 5 * GiB, "seeders": 10, "downloadvolumefactor": 0}, ] # 自定义规则(对应 MP「自定义规则」页:include/exclude 为单个正则字符串,size_range 单位 MB) CUSTOM_RULES: dict[str, dict[str, Any]] = { "SIZEMOVIE": {"size_range": "2000-30000"}, "SIZESMALL": {"size_range": "<8000"}, "SIZEBIG": {"size_range": ">20000"}, "SEED5": {"seeders": "5"}, } # verdict: full=可完整表达 / partial=近似表达(测试按近似翻译比对,差异登记 not_expressible) / none=不可表达(只生成 MP 结果) RULES: list[dict[str, Any]] = [ {"id": "R01", "rule": "4K", "verdict": "full"}, {"id": "R02", "rule": "1080P", "verdict": "full"}, {"id": "R03", "rule": "720P", "verdict": "full"}, {"id": "R04", "rule": "4K | 1080P", "verdict": "full"}, {"id": "R05", "rule": "4K > 1080P > 720P", "verdict": "full"}, {"id": "R06", "rule": "BLU", "verdict": "full"}, {"id": "R07", "rule": "!BLU", "verdict": "full"}, {"id": "R08", "rule": "REMUX", "verdict": "full"}, {"id": "R09", "rule": "!REMUX", "verdict": "full"}, {"id": "R10", "rule": "WEBDL", "verdict": "full"}, {"id": "R11", "rule": "!BLU & !REMUX", "verdict": "full"}, {"id": "R12", "rule": "4K & !REMUX > 1080P & WEBDL", "verdict": "full"}, {"id": "R13", "rule": "CNSUB", "verdict": "full"}, {"id": "R14", "rule": "CNSUB & 4K & !BLU", "verdict": "full"}, {"id": "R15", "rule": "CNSUB & 1080P > 1080P", "verdict": "full"}, {"id": "R16", "rule": "SIZEMOVIE", "verdict": "full"}, {"id": "R17", "rule": "SIZESMALL", "verdict": "full"}, {"id": "R18", "rule": "4K & SIZEMOVIE", "verdict": "full"}, {"id": "R19", "rule": "!BLU & !REMUX & !WEBDL", "verdict": "full"}, {"id": "R20", "rule": "BLURAY & !REMUX & !BLU", "verdict": "full"}, {"id": "R21", "rule": "4K & !BLU & !REMUX > 1080P & !BLU & !REMUX", "verdict": "full"}, {"id": "R22", "rule": "(4K | 1080P) & WEBDL", "verdict": "full"}, {"id": "R23", "rule": "CNSUB", "verdict": "partial", "mediainfo": {"original_language": "zh"}, "note": "tmdb.original_language=zh 时 CNSUB 对任何种子直接放行"}, {"id": "R24", "rule": "4K & !DOLBY", "verdict": "partial", "note": "硬条件无 HDR 维度,近似为 4K"}, {"id": "R25", "rule": "SPECSUB & CNSUB & !BLU", "verdict": "partial", "note": "无特效字幕维度,近似为 CNSUB & !BLU"}, {"id": "R26", "rule": "SIZEBIG", "verdict": "full"}, # ---- 以下不可表达,只记录 MP 行为供报告/将来补迁 {"id": "N01", "rule": "H265", "verdict": "none", "note": "编码只有软偏好 soft.codecs"}, {"id": "N02", "rule": "!DOLBY & !REMUX", "verdict": "none", "note": "HDR 只有软偏好"}, {"id": "N03", "rule": "HDR", "verdict": "none"}, {"id": "N04", "rule": "SPECSUB", "verdict": "none"}, {"id": "N05", "rule": "CNVOI", "verdict": "none", "note": "无音轨语言字段"}, {"id": "N06", "rule": "60FPS", "verdict": "none"}, {"id": "N07", "rule": "!3D", "verdict": "none"}, {"id": "N08", "rule": "ATMOS", "verdict": "none"}, {"id": "N09", "rule": "GZ", "verdict": "none", "note": "站点标签"}, {"id": "N10", "rule": "FREE", "verdict": "none", "note": "站点促销"}, {"id": "N11", "rule": "SEED5", "verdict": "none", "note": "做种数"}, {"id": "N12", "rule": "CNSUB | 4K", "verdict": "none", "note": "跨字段 OR"}, {"id": "N13", "rule": "!(4K & REMUX)", "verdict": "none", "note": "跨字段 NOT(AND) = OR"}, {"id": "N14", "rule": "UHD", "verdict": "none", "note": "ParsedRelease.uhd 有,但硬条件没有"}, {"id": "N15", "rule": "4K & H265 > 4K", "verdict": "none", "note": "level1 需编码硬条件"}, {"id": "N16", "rule": "NOSUCHRULE | 4K", "verdict": "none", "note": "未知规则名视为不匹配"}, ] def _filter_mp_root(mp_root: Optional[str]) -> Path: root = mp_root or os.environ.get("MP_ROOT") or "/opt/moviepilot-plus/upstream/MoviePilot" return Path(root).resolve() def _filter_mp_commit(root: Path) -> str: try: return subprocess.check_output(["git", "-C", str(root), "rev-parse", "--short", "HEAD"], text=True).strip() except Exception: # noqa: BLE001 return "unknown" def gen_filter(out_path: str, mp_root: Optional[str] = None) -> dict: root = _filter_mp_root(mp_root) if str(root) not in sys.path: sys.path.insert(0, str(root)) old_cwd = os.getcwd() os.chdir(root) try: from unittest.mock import patch from app.adapters.system import rust as rust_accel from app.application.rules import BUILTIN_RULE_SET from app.domain.context import TorrentInfo from app.modules.filter import FilterModule finally: os.chdir(old_cwd) rule_set = {**BUILTIN_RULE_SET, **CUSTOM_RULES} class _RuleHelper: def __init__(self, groups): self._groups = groups def get_rule_group_by_media(self, media=None, group_names=None): # noqa: ARG002 return self._groups def mk(t: dict) -> Any: return TorrentInfo( site_name="S", title=t["title"], description=t.get("description") or "", size=t["size"], seeders=t.get("seeders") or 0, labels=t.get("labels") or [], downloadvolumefactor=t.get("downloadvolumefactor", 1.0), ) def run(rule: dict, python: bool) -> list[Optional[int]]: module = FilterModule() module.rulehelper = _RuleHelper([SimpleNamespace(name=rule["id"], rule_string=rule["rule"])]) module.rule_set = rule_set mi = SimpleNamespace(**rule["mediainfo"]) if rule.get("mediainfo") else None out: list[Optional[int]] = [] for t in TORRENTS: # 逐条过滤,和订阅匹配链路(单种子列表)一致 ti = mk(t) if python: with patch("app.modules.filter.rust_accel.is_enabled", return_value=False): res = module.filter_torrents(rule_groups=[rule["id"]], torrent_list=[ti], mediainfo=mi) else: res = module.filter_torrents(rule_groups=[rule["id"]], torrent_list=[ti], mediainfo=mi) out.append(res[0].pri_order if res else None) return out rust_ok = rust_accel.is_enabled() cases = [] rust_mismatch = [] for rule in RULES: py = run(rule, python=True) rs = run(rule, python=False) if rust_ok else py for t, p, r in zip(TORRENTS, py, rs): cid = f"{rule['id']}:{t['id']}" if p != r: rust_mismatch.append({"caseId": cid, "python": p, "rust": r}) cases.append({ "caseId": cid, "ruleId": rule["id"], "torrentId": t["id"], "mp": {"matched": p is not None, "priOrder": p, "level": (101 - p) if p is not None else None}, }) out = Path(out_path) deviations: dict = {} if out.exists(): try: deviations = json.loads(out.read_text("utf-8")).get("deviations", {}) except Exception: # noqa: BLE001 deviations = {} doc = { "generatedFrom": { "repo": "MoviePilot", "commit": _filter_mp_commit(root), "engine": "python-fallback (app.modules.filter FilterModule, rust_accel disabled)", "rustCrossCheck": {"available": rust_ok, "mismatches": rust_mismatch}, "generator": "scripts/mp-sync/gen-differential.py filter", "semantics": "level = 1-based index of first matching '>' tier; priOrder = 101 - level (MP pri_order)", }, "customRules": CUSTOM_RULES, "torrents": TORRENTS, "rules": RULES, "cases": cases, "deviations": deviations, } out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(doc, ensure_ascii=False, indent=1) + "\n", "utf-8") return doc def gen_filter_cmd() -> None: d = gen_filter(str(DOMAIN_DIFF / "filter-cases.json"), str(MP_ROOT)) print(f"filter: cases={len(d['cases'])} rust≠python={len(d['generatedFrom']['rustCrossCheck']['mismatches'])}") # ---------------------------------------------------------------- rss / indexer:RSS 与站点列表页解析 # MoviePilot RSS / 站点索引解析 差分期望生成器。 # # 用 MoviePilot(MP_ROOT,须用其 .venv 的 python 运行)的 **Python 回退路径** 解析本项目的 fixture, # 把输入、MP 输出(同时附带 Rust 快路径输出作参考)写成 JSON,供 # `packages/adapters/test/differential/{rss,indexer}.test.ts` 与本项目解析器逐字段比对。 # # - RSS:`app/application/rss.py RssHelper._parse_impl`(注入假 HTTP/解析端口,解析端口返回 None 强制走 # lxml 回退;`parse_rss_items_fast` 另行直接调用记录到 `mp_rust`)。 # - 站点索引:`app/modules/indexer/spider/__init__.py SiteSpider.parse`(把 `rust_accel.parse_indexer_torrents/ # parse_indexer_subtitles` 打桩为 None 强制走 PyQuery 回退;再恢复原函数跑一次记录 `mp_rust`)。 # MP 自带站点定义在加密资源 `user.sites.v3.bin`(Cython `SitesHelper`),未认证用户 `get_indexer()` 返回 {}, # `get_indexsites()` 只含 id/name/url,因此 indexer 配置改用本项目 `definitions/*.json`(同 MP schema)。 # 输入先由 `deno run -A packages/adapters/test/differential/export-inputs.ts` 导出 tk-* fixture。 UMT_ROOT = REPO DEFAULT_OUT = UMT_ROOT / "packages/adapters/test/differential" DEFINITIONS_DIR = UMT_ROOT / "packages/adapters/src/sites/definitions" _BOOTSTRAPPED = False def _mp_root() -> Path: root = os.environ.get("MP_ROOT") if not root: raise SystemExit("需要环境变量 MP_ROOT(MoviePilot 源码根目录)") return Path(root).resolve() def _bootstrap() -> Path: """隔离 CONFIG_DIR 并建表(复用 MP 测试引导),固定进程时区为 UTC,返回 MP_ROOT。""" global _BOOTSTRAPPED root = _mp_root() if _BOOTSTRAPPED: return root # 时区固定为 UTC:RSS 的 astimezone(None) 与站点 normalize_datetime 结果与宿主时区无关 os.environ["TZ"] = "UTC" time.tzset() if str(root) not in sys.path: sys.path.insert(0, str(root)) cwd = os.getcwd() os.chdir(root) try: from app.testing.bootstrap import prepare_backend # noqa: WPS433 prepare_backend() # 同 MP tests/conftest.py configure_plugin_system_services:装配类型化配置快照与 runtime setting from app.application.configuration import ( RuntimeConfiguration, RuntimeSettingsService, configure_runtime_configuration, configure_runtime_settings, ) from app.runtime.config import settings from app.runtime.settings import configure_runtime_setting_provider from app.startup.composition.configuration import ( build_api_runtime_config, build_chain_runtime_config, build_scheduler_runtime_config, ) configure_runtime_configuration(RuntimeConfiguration( api=lambda: build_api_runtime_config(settings), scheduler=lambda: build_scheduler_runtime_config(settings), chain=lambda: build_chain_runtime_config(settings), )) configure_runtime_settings(RuntimeSettingsService(settings)) configure_runtime_setting_provider(lambda key: getattr(settings, key)) _freeze_dateparser_now() finally: os.chdir(cwd) _BOOTSTRAPPED = True return root # MP normalize_datetime 用 dateparser.parse(value) 无 settings,相对时间("3小时前")按当前时刻换算; # 为使期望文件可复现,冻结其相对基准。 FROZEN_NOW = "2026-09-27T12:00:00" def _freeze_dateparser_now() -> None: import datetime as _dt import dateparser as _real from app.foundation import temporal base = _dt.datetime.fromisoformat(FROZEN_NOW) class _Shim: @staticmethod def parse(value, *args, **kwargs): settings = dict(kwargs.pop("settings", None) or {}) settings.setdefault("RELATIVE_BASE", base) return _real.parse(value, *args, settings=settings, **kwargs) temporal.dateparser = _Shim() def _mp_commit(root: Path) -> str: try: return subprocess.check_output(["git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip() except Exception: # noqa: BLE001 return "unknown" def _sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _load_existing(path: Path) -> dict: if path.exists(): try: return json.loads(path.read_text("utf-8")) except Exception: # noqa: BLE001 return {} return {} def _jsonable(v: Any) -> Any: import datetime as _dt if isinstance(v, _dt.datetime): return v.isoformat() if isinstance(v, dict): return {k: _jsonable(x) for k, x in v.items()} if isinstance(v, (list, tuple)): return [_jsonable(x) for x in v] return v # --------------------------------------------------------------------------- RSS def _run_mp_rss_python(xml_text: str) -> dict: """走 RssHelper._parse_impl 的 lxml 回退分支;返回 {status, items}。""" import datetime as _dt from app.adapters.network.http import RequestUtils from app.application.rss import RssHelper, configure_rss_ports, reset_rss_ports raw = xml_text.encode("utf-8") class _Http: def get(self, **_kw): return SimpleNamespace(status_code=200, content=raw, text=xml_text, apparent_encoding="utf-8", encoding="utf-8", headers={}) def post(self, **_kw): # pragma: no cover raise AssertionError("不应 POST") @staticmethod def decode_xml(response, **kw): return RequestUtils.get_decoded_xml_content(response, **kw) class _Browser: @staticmethod def render(**_kw): # pragma: no cover raise AssertionError("不应访问浏览器") class _Parser: @staticmethod def parse(_content, _max): return None # 关闭 Rust,强制 Python 回退 calls: list = [] original = RssHelper._parse_publish_time def _spy(value): dt = original(value) calls.append({"text": value, "naive": bool(dt is not None and dt.tzinfo is None), "wall": dt.replace(tzinfo=None).isoformat() if dt is not None else None}) return dt RssHelper._parse_publish_time = staticmethod(_spy) configure_rss_ports(http=_Http(), browser=_Browser(), parser=_Parser()) try: result = RssHelper().parse("https://differential.invalid/rss") finally: reset_rss_ports() RssHelper._parse_publish_time = staticmethod(original) if result is None: return {"status": "none_expired", "items": []} if result is False: return {"status": "false_error", "items": []} items = [] for it in result: out = {k: it.get(k) for k in ("title", "description", "link", "enclosure", "size")} if "nickname" in it: out["nickname"] = it["nickname"] pd = it.get("pubdate") if pd == "": out["pubdate"] = {"kind": "absent"} else: call = calls.pop(0) if calls else {"text": None, "naive": False, "wall": None} if pd is None: out["pubdate"] = {"kind": "unparsed", "text": call["text"]} else: utc = pd.astimezone(_dt.timezone.utc) out["pubdate"] = { "kind": "naive" if call["naive"] else "aware", "text": call["text"], # naive:MP 以宿主本地时区解释(astimezone(None));wall 为墙上时间 "wall": call["wall"], "utc": utc.strftime("%Y-%m-%dT%H:%M:%S.") + f"{utc.microsecond // 1000:03d}Z", } items.append(out) return {"status": "ok", "items": items} def _run_mp_rss_rust(xml_text: str) -> Optional[dict]: try: import moviepilot_rust # type: ignore except Exception as err: # noqa: BLE001 return {"status": "unavailable", "error": str(err)} stripped = xml_text.strip() if not stripped or not stripped.startswith("<"): return {"status": "not_called", "note": "RssHelper.__parse_with_rust 对非 < 开头内容直接返回 None"} try: return {"status": "ok", "items": _jsonable(moviepilot_rust.parse_rss_items_fast(xml_text, 1001))} except BaseException as err: # noqa: BLE001 return {"status": "error", "error": f"{type(err).__name__}: {err}"} def gen_rss(out_dir: Optional[str] = None) -> Path: """生成 /rss.json;返回文件路径。""" root = _bootstrap() out = Path(out_dir).resolve() if out_dir else DEFAULT_OUT inputs = sorted(glob.glob(str(out / "inputs/rss/*.xml")) + glob.glob(str(out / "inputs/rss/*.txt"))) target = out / "rss.json" existing = _load_existing(target) samples = [] for f in inputs: p = Path(f) data = p.read_bytes() text = data.decode("utf-8") samples.append({ "id": p.stem, "input": {"file": f"inputs/rss/{p.name}", "sha256": _sha256(data)}, "mp": _run_mp_rss_python(text), "mp_rust": _run_mp_rss_rust(text), }) doc = { "kind": "rss", "generatedFrom": { "repo": "MoviePilot", "commit": _mp_commit(root), "function": "app/application/rss.py RssHelper._parse_impl (lxml fallback; parser port -> None)", "rust": "moviepilot_rust.parse_rss_items_fast (reference only)", "tz": "UTC", "generator": "gen-differential.py rss", }, "fields": ["title", "description", "link", "enclosure", "size", "pubdate", "nickname"], "samples": samples, "deviations": existing.get("deviations", {}), } target.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", "utf-8") return target # --------------------------------------------------------------------------- 站点索引 def _run_mp_spider(indexer: dict, html: str, keyword: Optional[str], search_type: str, use_rust: bool) -> dict: from app.adapters.system import rust as rust_accel from app.modules.indexer.spider import SiteSpider orig_t, orig_s = rust_accel.parse_indexer_torrents, rust_accel.parse_indexer_subtitles if not use_rust: rust_accel.parse_indexer_torrents = lambda **_kw: None rust_accel.parse_indexer_subtitles = lambda **_kw: None try: spider = SiteSpider(indexer, keyword=keyword, search_type=search_type) rows = spider.parse(html) return {"is_error": bool(spider.is_error), "torrents": _jsonable(rows)} except BaseException as err: # noqa: BLE001 return {"status": "error", "error": f"{type(err).__name__}: {err}"} finally: rust_accel.parse_indexer_torrents, rust_accel.parse_indexer_subtitles = orig_t, orig_s SUBTITLE_INDEXER = { # 取自 MP tests/test_rust_accel.py::test_rust_subtitle_parser_is_several_times_faster_than_python "id": "hhanclub-subtitles", "name": "憨憨(字幕)", "domain": "https://hhanclub.net/", "public": False, "subtitles": { "list": {"selector": "#subtitles-table > div"}, "fields": { "language_icon": {"selector": "div:nth-child(1) img", "attribute": "src"}, "title": {"selector": 'div:nth-child(2) a[href*="downloadsubs.php"]'}, "download": {"selector": 'div:nth-child(2) a[href*="downloadsubs.php"]', "attribute": "href"}, "size": {"selector": "div:nth-child(3)"}, "date_added": {"selector": "div:nth-child(4) span", "attribute": "title"}, "date_elapsed": {"selector": "div:nth-child(4) span"}, "grabs": {"defualt_value": 0}, "uploader": {"selector": 'div:nth-child(2) a[href*="userdetails.php"]'}, "report": {"selector": 'div:nth-child(5) a[href*="report.php"]', "attribute": "href"}, }, }, } def _subtitle_html() -> str: rows = [] for i in range(3): rows.append(f"""
Example Show S01E0{i + 1} 1080p WEB-DL CHS tester{i}
111.99 KB
1月18天
举报
""") return f'
{"".join(rows)}
' def gen_indexer(out_dir: Optional[str] = None) -> Path: """生成 /indexer.json;返回文件路径。""" root = _bootstrap() out = Path(out_dir).resolve() if out_dir else DEFAULT_OUT target = out / "indexer.json" existing = _load_existing(target) idx_dir = out / "inputs/indexer" defs = {} for f in sorted(DEFINITIONS_DIR.glob("*.json")): d = json.loads(f.read_text("utf-8")) if (d.get("umt") or {}).get("kind") == "nexusphp": defs[d["id"]] = (f, d) def page(name: str) -> tuple[str, dict]: data = (idx_dir / name).read_bytes() return data.decode("utf-8"), {"file": f"inputs/indexer/{name}", "sha256": _sha256(data)} runs: list[tuple[str, dict, dict, str, dict, Optional[str], str]] = [] list_pages = ["tk-audiences-list.html.txt", "tk-hhanclub-list.html.txt", "tk-chdbits-list.html.txt"] for site_id, (f, d) in defs.items(): for pg in list_pages: sid = f"{site_id}@{pg.removesuffix(".html.txt")}" runs.append((sid, d, {"source": "definition", "id": site_id, "file": str(f.relative_to(UMT_ROOT)), "sha256": _sha256(f.read_bytes())}, pg, {}, "x", "torrents")) # rl-*:真实站点页面脱敏后的样本(DET-2 观众列表页),只配对应站点的定义 for site_id, pg in (("audiences", "tk-audiences-login.html.txt"), ("chdbits", "tk-chdbits-login.html.txt"), ("audiences", "rl-audiences-list.html.txt")): f, d = defs[site_id] runs.append((f"{site_id}@{pg.removesuffix(".html.txt")}", d, {"source": "definition", "id": site_id, "file": str(f.relative_to(UMT_ROOT)), "sha256": _sha256(f.read_bytes())}, pg, {}, "x", "torrents")) for cfg in sorted(idx_dir.glob("*.indexer.json")): d = json.loads(cfg.read_text("utf-8")) name = cfg.name.replace(".indexer.json", "") # 同名 .html.txt 优先,否则用共享的 hw-dsl-rows.html.txt(.txt 后缀避免 biome 按 HTML 做 a11y lint) pg = f"{name}.html.txt" if (idx_dir / f"{name}.html.txt").exists() else "hw-dsl-rows.html.txt" runs.append((f"{name}@{pg.removesuffix(".html.txt")}", d, {"source": "inline", "file": f"inputs/indexer/{cfg.name}", "config": d}, pg, {}, None, "torrents")) samples = [] for sid, indexer, def_meta, pg, _extra, keyword, stype in runs: html, meta = page(pg) samples.append({ "id": sid, "definition": def_meta, "input": meta, "query": {"keyword": keyword, "search_type": stype}, "mp": _run_mp_spider(indexer, html, keyword, stype, use_rust=False), "mp_rust": _run_mp_spider(indexer, html, keyword, stype, use_rust=True), }) # 字幕:本项目未移植(静态登记),仍记录 MP 输出作为将来补迁的期望 sub_html = _subtitle_html() samples.append({ "id": "hhanclub-subtitles@inline", "definition": {"source": "inline", "config": SUBTITLE_INDEXER}, "input": {"inline": sub_html, "sha256": _sha256(sub_html.encode())}, "query": {"keyword": "Example Show", "search_type": "subtitles"}, "mp": _run_mp_spider(SUBTITLE_INDEXER, sub_html, "Example Show", "subtitles", use_rust=False), "mp_rust": _run_mp_spider(SUBTITLE_INDEXER, sub_html, "Example Show", "subtitles", use_rust=True), }) doc = { "kind": "indexer", "generatedFrom": { "repo": "MoviePilot", "commit": _mp_commit(root), "function": "app/modules/indexer/spider/__init__.py SiteSpider.parse (PyQuery fallback)", "rust": "moviepilot_rust.parse_indexer_torrents_fast / parse_indexer_subtitles_fast (reference only)", "definitions": "packages/adapters/src/sites/definitions/*.json (MP 内置定义加密且需认证,不可用)", "tz": "UTC", "dateparserNow": FROZEN_NOW, "generator": "gen-differential.py indexer", }, "fields": ["title", "description", "page_url", "enclosure", "size", "seeders", "peers", "grabs", "downloadvolumefactor", "uploadvolumefactor", "freedate", "hit_and_run", "category", "pubdate", "date_elapsed", "imdbid", "labels"], "samples": samples, "deviations": existing.get("deviations", {}), } target.write_text(json.dumps(doc, ensure_ascii=False, indent=2) + "\n", "utf-8") return target def gen_rss_cmd() -> None: print("rss: wrote", gen_rss(None)) def gen_indexer_cmd() -> None: print("indexer: wrote", gen_indexer(None)) COMMANDS = {"meta": gen_meta, "zhconv": gen_zhconv, "filter": gen_filter_cmd, "rss": gen_rss_cmd, "indexer": gen_indexer_cmd} def main(argv: list[str]) -> int: cmd = argv[1] if len(argv) > 1 else "all" if cmd == "all": # MP 运行时配置是进程级全局状态,每个子命令独立进程执行 for name in COMMANDS: subprocess.run([sys.executable, __file__, name], check=True) return 0 if cmd not in COMMANDS: print(f"unknown command {cmd}; choose from {', '.join(COMMANDS)} / all", file=sys.stderr) return 2 COMMANDS[cmd]() return 0 if __name__ == "__main__": sys.exit(main(sys.argv))