test(porting): differential tests against MoviePilot Rust-accelerated parsers

Snapshot differential tests (known deviations registered, current state passes): - title/path recognition: 216 samples vs MP MetaInfo/MetaInfoPath - filter rules: 42 rules x 28 torrents vs MP FilterModule - RSS/indexer parsing: 15 feeds, 72 indexer samples - zhconv: 69 samples Generator: scripts/mp-sync/gen-differential.py; report: docs/porting/mp-rust-logic-parity.md Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-27 22:44:11 +08:00
1 parent b86f75c011
commit 591cc78c7d
@@ -0,0 +1,760 @@
+#!/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:
+    """生成 <out_dir>/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"""<div class="grid"><div><img src="pic/flag/china.gif"></div><div>
+<a href="downloadsubs.php?torrentid={i}&amp;subid={i + 1000}">Example Show S01E0{i + 1} 1080p WEB-DL CHS</a>
+<a href="https://hhanclub.net/userdetails.php?id={i}"><b>tester{i}</b></a></div>
+<div><div>111.99&nbsp;KB</div></div><div><span title="2026-04-21 20:54:37">1月18天</span></div>
+<div><a href="report.php?subtitle={i + 1000}">举报</a></div></div>""")
+    return f'<div id="subtitles-table">{"".join(rows)}</div>'
+
+
+def gen_indexer(out_dir: Optional[str] = None) -> Path:
+    """生成 <out_dir>/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"))
+    for site_id, pg in (("audiences", "tk-audiences-login.html.txt"), ("chdbits", "tk-chdbits-login.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", "")
+        # 同名 <name>.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))