UHD_Media_Tool/scripts/mp-sync/gen-rules-differential.py

#!/usr/bin/env python3
"""规则组(W9-D)差分期望:MP 规则串解析 + 多规则组串联过滤的 pri 与 trace。

用法(必须用 MP 自己的 venv):

    MP_ROOT=/opt/moviepilot-plus/upstream/MoviePilot \
      $MP_ROOT/.venv/bin/python scripts/mp-sync/gen-rules-differential.py

输出 packages/domain/test/differential/rules-cases.json:
- parse:每个规则串按 `>` 切层后 RuleParser.parse(layer).as_list()(pyparsing,Python 参考实现)
- cases:FilterModule.filter_torrents(Python 回退路径)对每个 (规则组链 × 种子) 的 pri_order;
  trace 为 Python 路径的 debug 日志(逐条规则失败原因),rustTrace 为 filter_torrents_with_trace_fast 的调试输出
- deviations 不由本脚本生成,重跑时保留已有登记。
只读调用上游,不复制上游源码。
"""
from __future__ import annotations

import importlib.util
import json
import os
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Optional
from unittest.mock import patch

HERE = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("gen_differential", HERE / "gen-differential.py")
gd = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
assert spec and spec.loader
spec.loader.exec_module(gd)  # type: ignore[union-attr]

OUT = gd.DOMAIN_DIFF / "rules-cases.json"
GiB = 1024 ** 3

EXTRA_TORRENTS: list[dict[str, Any]] = [
    {"id": "t29", "title": "Infernal.Affairs.2002.1080p.BluRay.x264.DTS-HK-GRP", "description": "无间道 粤语 中字", "size": 12 * GiB, "seeders": 6},
    {"id": "t30", "title": "Movie.2019.1080p.WEB-DL.SDR.H264-GRP", "description": "", "size": 5 * GiB, "seeders": 5},
    {"id": "t31", "title": "Show.S01E05.2160p.WEB-DL.HDR10+.H265-GRP", "description": "官种 特效中字", "labels": ["官种"], "size": 4 * GiB, "seeders": 30, "downloadvolumefactor": 0.5},
    {"id": "t32", "title": "Movie.2018.1080p.BluRay.Remux.AVC.Atmos-GRP", "description": "国配 杜比全景声", "size": 28 * GiB, "seeders": 2},
]
TORRENTS = gd.TORRENTS + EXTRA_TORRENTS

# groups:规则组链(按顺序串联取交集,pri 取最后一组);单组即单条规则串
RULES: list[dict[str, Any]] = [
    *({"id": r["id"], "groups": [r["rule"]], **({"mediainfo": r["mediainfo"]} if r.get("mediainfo") else {})} for r in gd.RULES),
    {"id": "X01", "groups": ["SPECSUB & CNSUB > CNSUB & 4K > 1080P & !BLU"]},
    {"id": "X02", "groups": ["HKVOI"]},
    {"id": "X03", "groups": ["SDR"]},
    {"id": "X04", "groups": ["HDR & !DOLBY"]},
    {"id": "X05", "groups": ["DOLBY | HDR"]},
    {"id": "X06", "groups": ["H264 > H265"]},
    {"id": "X07", "groups": ["!(BLU | REMUX) & 4K"]},
    {"id": "X08", "groups": ["BLURAY"]},
    {"id": "X09", "groups": ["FREE | SEED5"]},
    {"id": "X10", "groups": ["UHD & REMUX > 4K"]},
    {"id": "X11", "groups": ["CNVOI | HKVOI > !CNVOI"]},
    {"id": "X12", "groups": ["!!4K"]},
    {"id": "X13", "groups": ["4K & (CNSUB | SPECSUB) & !3D"]},
    {"id": "X14", "groups": ["ATMOS & (REMUX | BLU)"]},
    {"id": "X15", "groups": ["GZ > FREE > 4K"]},
    {"id": "X16", "groups": ["CNVOI"], "mediainfo": {"original_language": "zh"}},
    {"id": "X17", "groups": ["SIZESMALL > SIZEBIG"]},
    {"id": "X18", "groups": ["1080P & CNSUB | 4K & !CNSUB"]},
    {"id": "X19", "groups": ["60FPS | 3D > ATMOS"]},
    {"id": "M01", "groups": ["4K | 1080P", "CNSUB > !CNSUB"]},
    {"id": "M02", "groups": ["!BLU", "4K > 1080P > 720P"]},
    {"id": "M03", "groups": ["4K > 1080P", "WEBDL"]},
    {"id": "M04", "groups": ["CNSUB", "SIZEMOVIE", "H265 > H264"]},
    {"id": "M05", "groups": ["REMUX > BLU > WEBDL", "!DOLBY"]},
]


def main() -> int:
    root = gd._filter_mp_root(None)
    if str(root) not in sys.path:
        sys.path.insert(0, str(root))
    old = os.getcwd()
    os.chdir(root)
    try:
        from app.adapters.system import rust as rust_accel
        from app.application.rules import BUILTIN_RULE_SET, RuleParser
        from app.domain.context import TorrentInfo
        from app.modules import filter as filter_mod
        from app.modules.filter import FilterModule
    finally:
        os.chdir(old)

    rule_set = {**BUILTIN_RULE_SET, **gd.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),
        )

    parser = RuleParser()
    parse_out: dict[str, Any] = {}
    for r in RULES:
        for g in r["groups"]:
            if g in parse_out:
                continue
            with patch("app.application.rules._filter_rule_parser", None):
                parse_out[g] = [parser.expr.parse_string(layer.strip()).as_list()[0] for layer in g.split(">")]

    def run(r: dict, t: dict) -> tuple[Optional[int], list[str], list[str]]:
        module = FilterModule()
        groups = [SimpleNamespace(name=f"{r['id']}#{i}", rule_string=g) for i, g in enumerate(r["groups"])]
        module.rulehelper = _RuleHelper(groups)
        module.rule_set = rule_set
        mi = SimpleNamespace(**r["mediainfo"]) if r.get("mediainfo") else None
        logs: list[str] = []
        with patch("app.modules.filter.rust_accel.is_enabled", return_value=False), \
                patch.object(filter_mod.logger, "debug", side_effect=lambda m, *a, **k: logs.append(str(m))):
            res = module.filter_torrents(rule_groups=[g.name for g in groups], torrent_list=[mk(t)], mediainfo=mi)
        pri = res[0].pri_order if res else None
        rust_trace: list[str] = []
        native = getattr(rust_accel, "_moviepilot_rust", None)
        if rust_accel.is_enabled() and native is not None and hasattr(native, "filter_torrents_with_trace_fast"):
            try:
                _m, rust_trace = native.filter_torrents_with_trace_fast(
                    [vars(g) for g in groups], [mk(t)], rule_set, mi, filter_mod._rust_parse_options(),
                )
            except BaseException:  # noqa: BLE001
                rust_trace = []
        return pri, logs, [str(x) for x in rust_trace]

    cases = []
    for r in RULES:
        for t in TORRENTS:
            pri, logs, rtrace = run(r, t)
            # 去掉每行重复的 "种子 S - <标题> " 前缀;Rust trace 与 Python 相同时不重复存
            prefix = f"种子 S - {t['title']} "
            strip = lambda xs: [x[len(prefix):].strip() if x.startswith(prefix) else x for x in xs]  # noqa: E731
            mp: dict[str, Any] = {"pri": pri, "trace": strip(logs)}
            if rtrace and rtrace != logs:
                mp["rustTrace"] = strip(rtrace)
            cases.append({"caseId": f"{r['id']}:{t['id']}", "ruleId": r["id"], "torrentId": t["id"], "mp": mp})

    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": gd._filter_mp_commit(root),
            "engine": "python-fallback (FilterModule, rust_accel disabled); parse = pyparsing RuleParser",
            "generator": "scripts/mp-sync/gen-rules-differential.py",
            "semantics": "groups 串联取交集,pri = 最后一组命中层的 pri_order(100 起逐层 -1);null = 被过滤",
        },
        "customRules": gd.CUSTOM_RULES,
        "torrents": TORRENTS,
        "rules": RULES,
        "parse": parse_out,
        "cases": cases,
        "deviations": deviations,
    }
    OUT.write_text(json.dumps(doc, ensure_ascii=False, indent=1) + "\n", "utf-8")
    print(f"rules: rules={len(RULES)} cases={len(cases)} parse={len(parse_out)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())