diff --git a/app/domain/meta/metabase.py b/app/domain/meta/metabase.py index a3417dea7..6b380ccc4 100644 --- a/app/domain/meta/metabase.py +++ b/app/domain/meta/metabase.py @@ -1,7 +1,7 @@ import logging import traceback -from dataclasses import dataclass -from typing import Union, Optional, List, Self +from dataclasses import asdict, dataclass +from typing import Any, Union, Optional, List, Self import cn2an import regex as re @@ -40,6 +40,100 @@ VIDEO_BIT_RE = re.compile( re.IGNORECASE, ) +_META_OPTIONAL_MERGE_FIELDS = ( + "resource_type", + "resource_pix", + "resource_team", + "customization", + "resource_effect", + "web_source", + "video_encode", + "video_bit", + "audio_encode", + "part", + "episode_group", +) + + +@dataclass(frozen=True, slots=True) +class MetaInfoSnapshot: + """MetaInfo 解析器的不可变稳定输出，用于跨实现等价校验和安全缓存。""" + + kind: str + isfile: bool + title: str + org_string: Optional[str] + subtitle: Optional[str] + type: str + cn_name: Optional[str] + en_name: Optional[str] + original_name: Optional[str] + year: Optional[str] + total_season: int + begin_season: Optional[int] + end_season: Optional[int] + total_episode: int + begin_episode: Optional[int] + end_episode: Optional[int] + part: Optional[str] + resource_type: Optional[str] + resource_effect: Optional[str] + resource_pix: Optional[str] + resource_team: Optional[str] + customization: Optional[str] + web_source: Optional[str] + video_encode: Optional[str] + video_bit: Optional[str] + audio_encode: Optional[str] + apply_words: tuple[str, ...] + media_source: Optional[str] + media_id: Optional[str] + episode_group: Optional[str] + fps: Optional[int] + + @classmethod + def from_meta(cls, meta: "MetaBase") -> "MetaInfoSnapshot": + """从兼容的可变 MetaBase 对象提取不包含临时状态的完整契约。""" + media_type = getattr(meta.type, "value", meta.type) + media_source = getattr(meta.media_source, "value", meta.media_source) + return cls( + kind="anime" if type(meta).__name__ == "MetaAnime" else "video", + isfile=bool(meta.isfile), + title=meta.title or "", + org_string=meta.org_string, + subtitle=meta.subtitle, + type=media_type, + cn_name=meta.cn_name, + en_name=meta.en_name, + original_name=meta.original_name, + year=meta.year, + total_season=meta.total_season, + begin_season=meta.begin_season, + end_season=meta.end_season, + total_episode=meta.total_episode, + begin_episode=meta.begin_episode, + end_episode=meta.end_episode, + part=meta.part, + resource_type=meta.resource_type, + resource_effect=meta.resource_effect, + resource_pix=meta.resource_pix, + resource_team=meta.resource_team, + customization=meta.customization, + web_source=meta.web_source, + video_encode=meta.video_encode, + video_bit=meta.video_bit, + audio_encode=meta.audio_encode, + apply_words=tuple(meta.apply_words or ()), + media_source=media_source, + media_id=meta.media_id, + episode_group=meta.episode_group, + fps=meta.fps, + ) + + def to_dict(self) -> dict[str, Any]: + """返回便于差异报告和序列化的独立字典。""" + return asdict(self) + @dataclass class MetaBase(object): @@ -652,45 +746,19 @@ class MetaBase(object): self.begin_episode = meta.begin_episode self.end_episode = meta.end_episode self.total_episode = meta.total_episode - # 版本 - if not self.resource_type: - self.resource_type = meta.resource_type - # 分辨率 - if not self.resource_pix: - self.resource_pix = meta.resource_pix - # 制作组/字幕组 - if not self.resource_team: - self.resource_team = meta.resource_team - # 自定义占位符 - if not self.customization: - self.customization = meta.customization - # 特效 - if not self.resource_effect: - self.resource_effect = meta.resource_effect - # 视频编码 - if not self.video_encode: - self.video_encode = meta.video_encode - # 视频位深 - if not self.video_bit: - self.video_bit = meta.video_bit - # 音频编码 - if not self.audio_encode: - self.audio_encode = meta.audio_encode + # 普通可选字段统一遵循文件优先、父目录补空，新增字段只维护一份策略。 + for field_name in _META_OPTIONAL_MERGE_FIELDS: + if not getattr(self, field_name): + setattr(self, field_name, getattr(meta, field_name)) # 帧率信息 if not self.fps: self.fps = meta.fps - # Part - if not self.part: - self.part = meta.part # 媒体身份必须原子合并，不能将不同目录层级的来源和ID拼成一对 current_source, current_id = resolve_media_identity(media=self) if current_source and current_id: self.media_source, self.media_id = current_source, current_id else: self.media_source, self.media_id = resolve_media_identity(media=meta) - # 剧集组 - if not self.episode_group and meta.episode_group: - self.episode_group = meta.episode_group def to_dict(self): """ diff --git a/app/domain/meta/metavideo.py b/app/domain/meta/metavideo.py index ee4c020b1..f9bfac9ac 100644 --- a/app/domain/meta/metavideo.py +++ b/app/domain/meta/metavideo.py @@ -127,6 +127,7 @@ class MetaVideo(MetaBase): and title.isdigit() \ and len(title) < 5: self.begin_episode = int(title) + self.total_episode = 1 self.type = MediaType.TV return # 全名为Season xx 及 Sxx 直接返回 diff --git a/app/domain/metainfo.py b/app/domain/metainfo.py index 2b2db4033..39f87d03d 100644 --- a/app/domain/metainfo.py +++ b/app/domain/metainfo.py @@ -1,8 +1,9 @@ import hashlib import logging +from dataclasses import dataclass from pathlib import Path from functools import lru_cache -from typing import Tuple, List, Optional +from typing import Mapping, Tuple, List, Optional import regex as re @@ -88,6 +89,18 @@ _LEGACY_ID_KEYS = ( ) +@dataclass(frozen=True, slots=True) +class _PreparedMetaInput: + """Python 回退解析的阶段输入，保留原文与预处理结果之间的明确边界。""" + + original_title: str + parsed_title: str + subtitle: Optional[str] + isfile: bool + apply_words: tuple[str, ...] + explicit_metainfo: Mapping[str, object] + + def _empty_metainfo() -> dict: """ 返回媒体标签的默认结构，避免不同识别请求之间共享可变状态。 @@ -263,37 +276,38 @@ def _find_metainfo_python(title: str) -> Tuple[str, dict]: return title, _normalize_metainfo_identity(metainfo) -def _build_meta_info( +def _prepare_meta_input( title: str, subtitle: Optional[str] = None, custom_words: List[str] = None, -) -> MetaBase: +) -> _PreparedMetaInput: """ - 根据标题构造元数据 + 应用识别词、显式标签和文件后缀规则，生成稳定的解析阶段输入。 """ - # 原标题 - org_title = title - # 预处理标题 - title, apply_words = WordsMatcher().prepare(title, custom_words=custom_words) - # 获取标题中媒体信息 - title, metainfo = find_metainfo(title) - # 判断是否处理文件 + original_title = title + parsed_title, apply_words = WordsMatcher().prepare(title, custom_words=custom_words) + # 完整 Rust 入口已经失败或被禁用，参考实现不得再次跨边界调用部分 Rust 解析器。 + parsed_title, explicit_metainfo = _find_metainfo_python(parsed_title) media_exts = get_media_extensions() - title_path = Path(title) if title else None + title_path = Path(parsed_title) if parsed_title else None if title_path and title_path.suffix.lower() in media_exts: isfile = True - # 去掉后缀 - title = title_path.stem + parsed_title = title_path.stem else: isfile = False - # 识别 - meta = MetaAnime(title, subtitle, isfile) if is_anime(title) else MetaVideo(title, subtitle, isfile) - # 记录原标题 - meta.title = org_title - # 记录使用的识别词 - meta.apply_words = apply_words or [] - # 修正媒体信息 - media_source, media_id = resolve_media_identity(media=metainfo) + return _PreparedMetaInput( + original_title=original_title, + parsed_title=parsed_title, + subtitle=subtitle, + isfile=isfile, + apply_words=tuple(apply_words or ()), + explicit_metainfo=explicit_metainfo, + ) + + +def _apply_explicit_metainfo(meta: MetaBase, metainfo: Mapping[str, object]) -> None: + """以显式标签覆盖推断字段，保持用户声明拥有最高优先级。""" + media_source, media_id = resolve_media_identity(media=dict(metainfo)) if media_source and media_id: meta.media_source = media_source meta.media_id = media_id @@ -313,6 +327,42 @@ def _build_meta_info( meta.end_episode = metainfo['end_episode'] if metainfo.get('total_episode') is not None: meta.total_episode = metainfo['total_episode'] + + +def _build_meta_info( + title: str, + subtitle: Optional[str] = None, + custom_words: List[str] = None, +) -> MetaBase: + """按准备、分类解析、显式覆盖三个阶段构造 Python MetaInfo。""" + prepared = _prepare_meta_input(title, subtitle, custom_words) + meta = MetaAnime( + prepared.parsed_title, + prepared.subtitle, + prepared.isfile, + ) if is_anime(prepared.parsed_title) else MetaVideo( + prepared.parsed_title, + prepared.subtitle, + prepared.isfile, + ) + meta.title = prepared.original_title + meta.apply_words = list(prepared.apply_words) + _apply_explicit_metainfo(meta, prepared.explicit_metainfo) + return meta + + +def _build_python_meta_info( + title: str, + subtitle: Optional[str] = None, + custom_words: List[str] = None, +) -> MetaBase: + """构造并完成 original_name 的纯 Python 参考解析结果。""" + meta = _build_meta_info(title=title, subtitle=subtitle, custom_words=custom_words) + if meta.apply_words: + original_meta = _build_meta_info(title=title, subtitle=subtitle) + meta.original_name = original_meta.name or meta.name + else: + meta.original_name = meta.name or None return meta @@ -473,13 +523,11 @@ def MetaInfo(title: str, subtitle: Optional[str] = None, custom_words: List[str] ) if rust_meta: return rust_meta - meta = _build_meta_info(title=title, subtitle=subtitle, custom_words=custom_words) - if meta.apply_words: - original_meta = _build_meta_info(title=title, subtitle=subtitle) - meta.original_name = original_meta.name or meta.name - else: - meta.original_name = meta.name - return meta + return _build_python_meta_info( + title=title, + subtitle=subtitle, + custom_words=custom_words, + ) def MetaInfoPath(path: Path, custom_words: List[str] = None, force_video: bool = False) -> MetaBase: @@ -512,16 +560,16 @@ def MetaInfoPath(path: Path, custom_words: List[str] = None, force_video: bool = if rust_meta: return rust_meta # 文件元数据，不包含后缀 - file_meta = MetaInfo(title=path.name, custom_words=custom_words) + file_meta = _build_python_meta_info(title=path.name, custom_words=custom_words) if should_use_parent_title_for_file_stem(path.stem, path.parent.name, file_meta): clear_parsed_title_for_parent_merge(file_meta) # 上级目录元数据 - dir_meta = MetaInfo(title=path.parent.name, custom_words=custom_words) + dir_meta = _build_python_meta_info(title=path.parent.name, custom_words=custom_words) if file_meta.type == MediaType.TV or dir_meta.type != MediaType.TV: # 合并元数据 file_meta.merge(dir_meta) # 上上级目录元数据 - root_meta = MetaInfo(title=path.parent.parent.name, custom_words=custom_words) + root_meta = _build_python_meta_info(title=path.parent.parent.name, custom_words=custom_words) if file_meta.type == MediaType.TV or root_meta.type != MediaType.TV: # 合并元数据 file_meta.merge(root_meta) diff --git a/scripts/benchmark_metainfo_rust.py b/scripts/benchmark_metainfo_rust.py index c3dd6095d..c48fdab58 100644 --- a/scripts/benchmark_metainfo_rust.py +++ b/scripts/benchmark_metainfo_rust.py @@ -11,10 +11,16 @@ sys.path.insert(0, str(PROJECT_ROOT)) from app.domain import metainfo as metainfo_module from app.domain.meta.metaanime import MetaAnime +from app.domain.meta.metabase import MetaInfoSnapshot from app.domain.meta.metamusic import MetaMusic -from app.domain.meta.runtime import get_audio_extensions +from app.domain.meta.runtime import ( + configure_recognition_runtime, + get_audio_extensions, + get_metainfo_accelerator, +) from app.domain.metainfo import MetaInfo, MetaInfoPath from app.adapters.system import rust as rust_accel +from app.runtime.settings import RuntimeSettingsCompat from tests.cases.meta import meta_cases @@ -56,6 +62,20 @@ _MUSIC_CASES: tuple[BenchmarkInput, ...] = ( ) +def configure_benchmark_runtime() -> None: + """注入独立基准所需的文件类型和 Rust 适配器，避免静默测量 Python 回退。""" + settings = RuntimeSettingsCompat() + configure_recognition_runtime( + media_extensions_provider=lambda: ( + *settings.RMT_MEDIAEXT, + *settings.RMT_SUBEXT, + *settings.RMT_AUDIOEXT, + ), + audio_extensions_provider=lambda: settings.RMT_AUDIOEXT, + accelerator=rust_accel, + ) + + def build_video_inputs(repeat: int) -> list[BenchmarkInput]: """构造覆盖影视 MetaInfo 和 MetaInfoPath 生产入口的基准输入。""" inputs: list[BenchmarkInput] = [] @@ -124,24 +144,10 @@ def _enum_value(value: Any) -> Any: def project_video_result(meta: Any) -> dict[str, Any]: - """提取影视识别对外契约字段，排除 Python 解析器的临时内部状态。""" - return { - "kind": "anime" if isinstance(meta, MetaAnime) else "video", - "type": _enum_value(meta.type), - "cn_name": meta.cn_name or "", - "en_name": meta.en_name or "", - "year": meta.year or "", - "part": meta.part or "", - "season": meta.season, - "episode": meta.episode, - "resource_type": meta.edition, - "resource_pix": meta.resource_pix or "", - "video_encode": meta.video_encode or "", - "audio_encode": meta.audio_encode or "", - "fps": meta.fps or None, - "media_source": _enum_value(meta.media_source), - "media_id": meta.media_id, - } + """提取影视识别完整稳定契约，排除解析器的临时内部状态和派生展示字段。""" + result = MetaInfoSnapshot.from_meta(meta).to_dict() + result["apply_words"] = list(result["apply_words"]) + return result def project_music_result(meta: Any) -> dict[str, Any]: @@ -252,9 +258,17 @@ def benchmark_suite( def validate_rust_runtime() -> None: - """确认 Rust 总开关和音乐扩展入口可用，拒绝静默回退形成伪基准。""" + """确认领域入口已注入可工作的 Rust 影视与音乐解析器，拒绝伪基准。""" if not rust_accel.is_enabled(): raise RuntimeError("Rust 加速未启用或 moviepilot-rust 扩展不可用") + if get_metainfo_accelerator() is not rust_accel: + raise RuntimeError("MetaInfo 领域入口未注入 Rust 加速器") + video_probe = rust_accel.parse_metainfo( + "Benchmark Movie 2026 1080p WEB-DL H265", + options={"media_exts": [".mkv"]}, + ) + if not isinstance(video_probe, dict) or video_probe.get("en_name") != "Benchmark Movie": + raise RuntimeError("Rust 影视解析探针未返回有效结果，拒绝测量 Python 回退") if not callable(getattr(rust_accel, "parse_metamusic", None)): raise RuntimeError("MoviePilot 后端缺少 rust_accel.parse_metamusic 适配器") extension = getattr(rust_accel, "_moviepilot_rust", None) @@ -316,6 +330,7 @@ def main() -> int: """运行影视与音乐 Rust/Python 生产入口基准测试。""" args = parse_args() try: + configure_benchmark_runtime() validate_rust_runtime() video_inputs = build_video_inputs(args.repeat_inputs) music_inputs = build_music_inputs(args.repeat_inputs) diff --git a/tests/test_benchmark_metainfo_rust.py b/tests/test_benchmark_metainfo_rust.py index 5996e8865..6712242c2 100644 --- a/tests/test_benchmark_metainfo_rust.py +++ b/tests/test_benchmark_metainfo_rust.py @@ -23,6 +23,19 @@ def test_build_inputs_separates_video_and_music_domains(): for kind, value, _subtitle in music_once if kind == "music_query") +def test_configure_benchmark_runtime_injects_extensions_and_accelerator(monkeypatch): + """独立基准必须显式注入文件类型和 Rust 适配器，不能依赖应用启动副作用。""" + configure_runtime = Mock() + monkeypatch.setattr(benchmark, "configure_recognition_runtime", configure_runtime) + + benchmark.configure_benchmark_runtime() + + kwargs = configure_runtime.call_args.kwargs + assert kwargs["accelerator"] is benchmark.rust_accel + assert ".mkv" in kwargs["media_extensions_provider"]() + assert ".flac" in kwargs["audio_extensions_provider"]() + + def test_parse_input_uses_public_production_entries(monkeypatch): """输入分发应调用 MetaInfo、MetaInfoPath 和 MetaMusic.parse_query 公开入口。""" title_result = object() @@ -140,6 +153,24 @@ def test_video_projection_ignores_python_parser_internal_state(): ) +def test_video_projection_covers_stable_path_merge_fields(): + """差异投影必须覆盖路径合并容易遗漏的平台、效果、位深和范围字段。""" + result = benchmark.MetaInfo("Show S01E01 2026 2160p AMZN WEB-DL HDR H265 10bit") + result.web_source = "Amazon" + result.resource_team = "GROUP" + result.customization = "CUSTOM" + + projected = benchmark.project_video_result(result) + + assert projected["web_source"] == "Amazon" + assert projected["resource_effect"] == "HDR" + assert projected["video_bit"] == "10bit" + assert projected["begin_season"] == 1 + assert projected["begin_episode"] == 1 + assert projected["resource_team"] == "GROUP" + assert projected["customization"] == "CUSTOM" + + def test_validate_rust_runtime_rejects_disabled_and_old_extensions(monkeypatch): """运行前检查应拒绝关闭的 Rust 和缺少音乐入口的旧扩展。""" rust_accel = benchmark.rust_accel @@ -149,6 +180,12 @@ def test_validate_rust_runtime_rejects_disabled_and_old_extensions(monkeypatch): benchmark.validate_rust_runtime() monkeypatch.setattr(rust_accel, "is_enabled", Mock(return_value=True)) + monkeypatch.setattr(benchmark, "get_metainfo_accelerator", Mock(return_value=rust_accel)) + monkeypatch.setattr( + rust_accel, + "parse_metainfo", + Mock(return_value={"en_name": "Benchmark Movie"}), + ) monkeypatch.setattr(rust_accel, "parse_metamusic", Mock(return_value={}), raising=False) monkeypatch.setattr(rust_accel, "_moviepilot_rust", SimpleNamespace()) @@ -161,6 +198,12 @@ def test_validate_rust_runtime_requires_successful_music_probe(monkeypatch): rust_accel = benchmark.rust_accel extension = SimpleNamespace(parse_metamusic_fast=Mock()) monkeypatch.setattr(rust_accel, "is_enabled", Mock(return_value=True)) + monkeypatch.setattr(benchmark, "get_metainfo_accelerator", Mock(return_value=rust_accel)) + monkeypatch.setattr( + rust_accel, + "parse_metainfo", + Mock(return_value={"en_name": "Benchmark Movie"}), + ) monkeypatch.setattr(rust_accel, "_moviepilot_rust", extension) monkeypatch.setattr(rust_accel, "parse_metamusic", Mock(return_value=None), raising=False) @@ -170,6 +213,7 @@ def test_validate_rust_runtime_requires_successful_music_probe(monkeypatch): def test_main_outputs_independent_video_and_music_metrics(monkeypatch, capsys): """主程序应分别输出影视和音乐等价状态、耗时及性能提升。""" + monkeypatch.setattr(benchmark, "configure_benchmark_runtime", Mock()) monkeypatch.setattr(benchmark, "validate_rust_runtime", Mock()) monkeypatch.setattr(benchmark, "build_video_inputs", Mock(return_value=[("title", "V", None)])) monkeypatch.setattr( @@ -212,6 +256,7 @@ def test_main_outputs_independent_video_and_music_metrics(monkeypatch, capsys): def test_main_reports_runtime_failure_with_nonzero_exit(monkeypatch, capsys): """Rust 未就绪时主程序应明确报错并返回非零状态。""" + monkeypatch.setattr(benchmark, "configure_benchmark_runtime", Mock()) monkeypatch.setattr( benchmark, "validate_rust_runtime", diff --git a/tests/test_metainfo.py b/tests/test_metainfo.py index 725363416..b83a2dce8 100644 --- a/tests/test_metainfo.py +++ b/tests/test_metainfo.py @@ -7,9 +7,15 @@ import pytest from app.domain.context import MediaInfo from app.domain.metainfo import MetaInfo, MetaInfoPath, find_metainfo -from app.domain.meta.metabase import MetaBase +from app.domain.meta.metabase import MetaBase, MetaInfoSnapshot from app.domain.meta.metamusic import MetaMusic from app.domain.meta.metaanime import MetaAnime +from app.domain.meta.runtime import ( + configure_recognition_runtime, + get_audio_extensions, + get_media_extensions, + get_metainfo_accelerator, +) from app.application.torrent import TorrentHelper from app.schemas.types import MediaSource, MediaType from tests.cases.meta import meta_cases @@ -100,6 +106,78 @@ def test_metainfopath_with_empty_custom_words(): assert meta is not None +def test_metainfo_snapshot_exposes_complete_immutable_contract(): + """稳定快照应包含路径合并字段，且不受原 MetaBase 后续修改影响。""" + meta = MetaInfo("Show.S01E01.2026.2160p.WEB-DL.HDR.H265.10bit-GROUP.mkv") + meta.web_source = "Amazon" + snapshot = MetaInfoSnapshot.from_meta(meta) + meta.web_source = "Netflix" + + assert snapshot.kind == "video" + assert snapshot.begin_season == 1 + assert snapshot.begin_episode == 1 + assert snapshot.resource_effect == "HDR" + assert snapshot.video_bit == "10bit" + assert snapshot.web_source == "Amazon" + assert snapshot.apply_words == () + + +def test_metainfopath_merges_parent_streaming_platform(): + """文件名缺少平台时应从父目录补充，避免路径识别丢失稳定资源字段。""" + media_extensions = get_media_extensions() + audio_extensions = get_audio_extensions() + accelerator = get_metainfo_accelerator() + configure_recognition_runtime( + media_extensions_provider=lambda: (".mkv",), + audio_extensions_provider=lambda: (), + accelerator=None, + ) + try: + meta = MetaInfoPath(Path("/Show 2024 AMZN WEB-DL/Show.S01E01.mkv")) + finally: + configure_recognition_runtime( + media_extensions_provider=lambda: media_extensions, + audio_extensions_provider=lambda: audio_extensions, + accelerator=accelerator, + ) + + assert meta.web_source == "Amazon" + assert meta.year == "2024" + assert meta.episode == "E01" + + +def test_numeric_video_filename_sets_single_episode_total(): + """纯数字视频文件名表示单集时，范围字段必须保持自洽。""" + media_extensions = get_media_extensions() + audio_extensions = get_audio_extensions() + accelerator = get_metainfo_accelerator() + configure_recognition_runtime( + media_extensions_provider=lambda: (".mkv",), + audio_extensions_provider=lambda: (), + accelerator=None, + ) + try: + meta = MetaInfo("5.mkv") + finally: + configure_recognition_runtime( + media_extensions_provider=lambda: media_extensions, + audio_extensions_provider=lambda: audio_extensions, + accelerator=accelerator, + ) + + assert meta.begin_episode == 5 + assert meta.end_episode is None + assert meta.total_episode == 1 + + +def test_empty_video_title_keeps_optional_original_name_none(): + """无法提取标题时 original_name 保持空值，不使用含义不同的空字符串。""" + meta = MetaInfo("S02E1000.mkv") + + assert meta.name == "" + assert meta.original_name is None + + def test_custom_words_apply_words_recording(): """测试 apply_words 记录功能。""" custom_words = ["替换词 => 新词"]