diff --git a/app/application/classification/execution.py b/app/application/classification/execution.py index 29bc45fe3..7f8bf7867 100644 --- a/app/application/classification/execution.py +++ b/app/application/classification/execution.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable, Mapping from copy import deepcopy from enum import Enum -from typing import Protocol, TypeAlias +from typing import Final, Protocol, TypeAlias, cast from app.application.classification.legacy import ( build_legacy_tmdb_extension_facts, @@ -16,6 +16,7 @@ from app.domain.classification.facts import build_classification_facts from app.domain.context import MediaInfo, MusicAlbumInfo, MusicArtistInfo, MusicInfo from app.schemas.category import ( CategoryConfig, + ClassificationEvaluation, ClassificationFacts, ClassificationFactValue, ClassificationPolicy, @@ -34,6 +35,9 @@ ClassificationExtensionFactsProvider: TypeAlias = Callable[ ] """按当前插件注册表校验并提供来源扩展分类事实的端口。""" +_LEGACY_TMDB_SOURCE: Final[str] = "themoviedb" +_LEGACY_RULE_PREFIX: Final[str] = "legacy." + class ClassificationRuntimePort(Protocol): """分类执行只需要的活动策略与 legacy 快照端口。""" @@ -278,6 +282,9 @@ class ClassificationExecutionService: ) -> ClassificationSubject: """应用纯求值结果和人工覆盖，并更新兼容目录分类。""" evaluation = ClassificationEvaluator.evaluate(policy, facts) + legacy_evaluation = _evaluate_legacy_tmdb_compatibility(policy, facts) + if legacy_evaluation is not None and _uses_fallback(evaluation.result): + evaluation = legacy_evaluation result = evaluation.result.model_copy(deep=True) if effective_override: result.effective = effective_override.model_copy(deep=True) @@ -346,6 +353,94 @@ class ClassificationExecutionService: media.set_library_category(_category_path_snapshot(effective)) +def _evaluate_legacy_tmdb_compatibility( + policy: ClassificationPolicy, + facts: ClassificationFacts, +) -> ClassificationEvaluation | None: + """让旧 TMDB 分类规则消费非 TMDB 来源已经拥有的标准事实。""" + if facts.identity.media_source == _LEGACY_TMDB_SOURCE: + return None + legacy_rules = [ + rule for rule in policy.rules if rule.id.startswith(_LEGACY_RULE_PREFIX) + ] + if not legacy_rules: + return None + compatibility_policy = policy.model_copy( + deep=True, + update={"rules": legacy_rules}, + ) + compatibility_facts = _legacy_tmdb_compatibility_facts(policy, facts) + return ClassificationEvaluator.evaluate( + compatibility_policy, + compatibility_facts, + ) + + +def _legacy_tmdb_compatibility_facts( + policy: ClassificationPolicy, + facts: ClassificationFacts, +) -> ClassificationFacts: + """把跨来源标准字段投影到旧规则的 TMDB 扩展命名空间。""" + extensions = { + str(source): {str(key): value for key, value in values.items()} + for source, values in facts.extensions.items() + } + legacy_info = _legacy_tmdb_info_from_standard_facts(facts) + for source, values in build_legacy_tmdb_extension_facts( + policy, + legacy_info, + ).items(): + target = extensions.setdefault(source, {}) + for field, value in values.items(): + target.setdefault(field, value) + identity = facts.identity.model_copy( + update={"media_source": _LEGACY_TMDB_SOURCE} + ) + return cast( + ClassificationFacts, + facts.model_copy( + deep=True, + update={"identity": identity, "extensions": extensions}, + ), + ) + + +def _legacy_tmdb_info_from_standard_facts( + facts: ClassificationFacts, +) -> dict[str, object]: + """构造旧规则投影所需的有限 TMDB 字段，不伪造缺失事实。""" + media = facts.media + info: dict[str, object] = {} + if media.language: + info["original_language"] = media.language + if media.countries: + countries = [str(country) for country in media.countries if country] + if countries: + info["origin_country"] = countries + info["production_countries"] = [ + {"iso_3166_1": country} for country in countries + ] + if media.year is not None: + info["release_date"] = str(media.year) + for field in ( + "adult", + "runtime", + "content_rating", + "companies", + "networks", + ): + value = getattr(media, field, None) + if value not in (None, "", []): + info[field] = value + return info + + +def _uses_fallback(result: ClassificationResult) -> bool: + """判断主来源求值是否没有命中具体分类规则。""" + selection = result.effective or result.recommended + return selection is None or selection.source in {None, "fallback", "source_fallback"} + + def _classification_extensions( policy: ClassificationPolicy, media: ClassificationSubject, diff --git a/app/db/adapters/classification.py b/app/db/adapters/classification.py index b86ed6319..06838a4ef 100644 --- a/app/db/adapters/classification.py +++ b/app/db/adapters/classification.py @@ -34,7 +34,7 @@ DirectoryConfigurationNormalizer = Callable[ def discard_removed_source_fallbacks(value: Any) -> Any: - """读取持久化策略时丢弃已删除的来源级默认分类字段，不再恢复其行为。""" + """读取旧持久化策略时把 TMDB 兜底平移为媒体类型全局兜底。""" if not isinstance(value, Mapping): return value @@ -47,9 +47,27 @@ def discard_removed_source_fallbacks(value: Any) -> Any: if isinstance(history, list): policies.extend(item for item in history if isinstance(item, dict)) for policy in policies: - policy.pop("source_fallbacks", None) + source_fallbacks = policy.pop("source_fallbacks", None) + if not isinstance(source_fallbacks, Mapping): + continue + fallbacks = policy.get("fallbacks") + tmdb_fallbacks = source_fallbacks.get("themoviedb") + if not isinstance(fallbacks, dict) or not isinstance(tmdb_fallbacks, Mapping): + continue + for media_type, category_id in tmdb_fallbacks.items(): + if fallbacks.get(media_type) != _COMMON_FALLBACK_IDS.get(media_type): + continue + if category_id: + fallbacks[media_type] = category_id return state + +_COMMON_FALLBACK_IDS = { + "电影": "movie.uncategorized", + "电视剧": "tv.uncategorized", + "音乐": "music.uncategorized", +} + _CONFIGURATION_LOCK_KEYS = ( SystemConfigKey.MediaClassificationPolicy.value, SystemConfigKey.Directories.value, diff --git a/tests/test_media_classification_configuration_adapter.py b/tests/test_media_classification_configuration_adapter.py index d7d732754..45bc36af3 100644 --- a/tests/test_media_classification_configuration_adapter.py +++ b/tests/test_media_classification_configuration_adapter.py @@ -25,6 +25,7 @@ from app.application.directory import normalize_directory_configurations_for_pol from app.db.adapters.classification import ( SystemConfigClassificationPolicyStore, SystemConfigDirectoryConfigurationStore, + discard_removed_source_fallbacks, ) from app.db.models.systemconfig import SystemConfig from app.db.oper.systemconfig import SystemConfigOper @@ -75,6 +76,37 @@ def _store(db: Any) -> tuple[SystemConfigClassificationPolicyStore, SystemConfig ) +def test_old_source_fallback_is_promoted_to_global_fallback() -> None: + """旧策略读取时应保留迁移生成的分类兜底，并移除废弃字段。""" + state = { + "active": { + "fallbacks": { + "电影": "movie.uncategorized", + "电视剧": "tv.uncategorized", + "音乐": "music.uncategorized", + }, + "source_fallbacks": { + "themoviedb": {"电视剧": "legacy.tv.uncategorized"}, + }, + }, + "history": [ + { + "fallbacks": {"电视剧": "tv.uncategorized"}, + "source_fallbacks": { + "themoviedb": {"电视剧": "legacy.tv.previous"}, + }, + } + ], + } + + normalized = discard_removed_source_fallbacks(state) + + assert normalized["active"]["fallbacks"]["电视剧"] == "legacy.tv.uncategorized" + assert normalized["history"][0]["fallbacks"]["电视剧"] == "legacy.tv.previous" + assert "source_fallbacks" not in normalized["active"] + assert "source_fallbacks" not in normalized["history"][0] + + def test_adapter_initializes_and_round_trips_json_datetime(db: Any) -> None: """首次创建原子写入 JSON，并在提交后发布系统配置快照。""" store, oper = _store(db) diff --git a/tests/test_media_classification_execution.py b/tests/test_media_classification_execution.py index 2df448d8c..188e4a264 100644 --- a/tests/test_media_classification_execution.py +++ b/tests/test_media_classification_execution.py @@ -181,6 +181,39 @@ def test_execution_classifies_copy_and_preserves_source_identity() -> None: assert source.classification.policy_revision == 1 +def test_legacy_tmdb_rules_use_non_tmdb_standard_facts() -> None: + """旧 TMDB 分类规则应使用豆瓣等来源已有的标准国家事实。""" + migration = migrate_legacy_category_config( + { + "movie": {}, + "tv": { + "国产剧": {"origin_country": "CN,TW,HK"}, + "未分类": None, + }, + } + ) + source = MediaInfo( + media_source=MediaSource.Douban, + media_id="35593344", + type=MediaType.TV, + title="测试剧", + production_countries=[{"name": "中国大陆"}], + ) + + finalized = ClassificationExecutionService( + _Runtime(migration.policy) + ).finalize(source) + + assert finalized.media_source == MediaSource.Douban + assert finalized.library_category == "国产剧" + assert finalized.classification is not None + assert finalized.classification.effective.category_id == next( + category.id + for category in migration.policy.categories + if category.name == "国产剧" + ) + + def test_execution_builds_complete_facts_without_mutating_media() -> None: """影响分析事实入口应复用插件字段构造，并保持原媒体对象不变。""" source = MediaInfo(