""" 完整识别结果分类执行与 Chain 收口验收测试。 """ from __future__ import annotations import asyncio from copy import deepcopy from typing import Optional from unittest.mock import AsyncMock, Mock, patch from app.application.classification.analysis import build_classification_facts_from_media from app.application.classification.execution import ( ClassificationExecutionService, evaluate_classification_facts, ) from app.application.classification.legacy import migrate_legacy_category_config from app.chain.base import ChainBase from app.domain.context import MediaInfo, MusicArtistInfo, MusicInfo from app.domain.metainfo import MetaInfo from app.schemas.category import ( CategoryConfig, CategoryRule, ClassificationFacts, ClassificationPolicy, ClassificationResult, ClassificationSelection, ) from app.schemas.types import MediaSource, MediaType class _Runtime: """ 提供可变活动策略与隔离 legacy 快照的测试端口。 """ def __init__( self, policy: ClassificationPolicy | None, legacy: CategoryConfig | None = None, ) -> None: """ 保存测试需要的当前策略和旧配置。 """ self.policy = policy self.legacy = legacy or CategoryConfig() def active_policy(self) -> ClassificationPolicy | None: """ 返回活动策略副本。 """ return self.policy.model_copy(deep=True) if self.policy else None def legacy_config(self) -> CategoryConfig: """ 返回 legacy 配置副本。 """ return self.legacy.model_copy(deep=True) def _policy(*, revision: int = 7, movie_category: str = "movie.jp") -> ClassificationPolicy: """ 构造同时覆盖影视、音乐专辑和艺术家的分类策略。 """ return ClassificationPolicy.model_validate( { "schema_version": 2, "revision": revision, "categories": [ { "id": movie_category, "media_type": "电影", "name": "日本动画", "path": ["动画", "日本"], }, { "id": "movie.other", "media_type": "电影", "name": "其它电影", "path": ["其它电影"], }, { "id": "tv.other", "media_type": "电视剧", "name": "其它剧集", "path": ["其它剧集"], }, { "id": "music.live", "media_type": "音乐", "name": "现场音乐", "path": ["音乐", "现场"], }, { "id": "music.artist", "media_type": "音乐", "name": "艺术家", "path": ["音乐", "艺术家"], }, { "id": "music.other", "media_type": "音乐", "name": "其它音乐", "path": ["其它音乐"], }, ], "rules": [ { "id": "rule.movie.jp", "name": "日本动画", "kind": "category", "media_types": ["电影"], "when": { "all": [ { "field": "media.genre_keys", "operator": "contains_any", "value": ["animation"], }, { "field": "media.countries", "operator": "contains_any", "value": ["JP"], }, ] }, "target": {"category_id": movie_category}, }, { "id": "rule.music.live", "name": "现场音乐", "kind": "category", "media_types": ["音乐"], "when": { "all": [ { "field": "music.secondary_types", "operator": "contains_any", "value": ["Live"], } ] }, "target": {"category_id": "music.live"}, }, { "id": "rule.music.artist", "name": "艺术家", "kind": "category", "media_types": ["音乐"], "when": { "all": [ { "field": "music.entity_type", "operator": "equals", "value": "artist", } ] }, "target": {"category_id": "music.artist"}, }, ], "fallbacks": { "电影": "movie.other", "电视剧": "tv.other", "音乐": "music.other", }, } ) def test_execution_classifies_copy_and_preserves_source_identity() -> None: """ 完整来源对象应被复制分类，主身份和来源对象不得被改写。 """ source = MediaInfo( media_source="douban", media_id="1291561", type=MediaType.MOVIE, title="千与千寻", origin_country=["JP"], genres=[{"name": "动画"}], library_category="旧缓存分类", classification=ClassificationResult( policy_revision=1, state="complete", ), ) finalized = ClassificationExecutionService(_Runtime(_policy())).finalize(source) assert finalized is not source assert finalized.media_source == MediaSource.Douban assert finalized.media_id == "1291561" assert finalized.library_category == "动画/日本" assert finalized.classification is not None assert finalized.classification.policy_revision == 7 assert finalized.classification.effective.category_id == "movie.jp" assert source.library_category == "旧缓存分类" 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_legacy_compatibility_preserves_primary_source_provenance() -> None: """ 旧 TMDB 规则兼容求值不得把非 TMDB 主身份伪装成 TMDB。 """ migration = migrate_legacy_category_config( {"movie": {"华语电影": {"original_language": "zh"}}, "tv": {}} ) media = MediaInfo( media_source=MediaSource.Douban, media_id="douban-1", type=MediaType.MOVIE, original_language="zh", ) facts = build_classification_facts_from_media(media) evaluation = evaluate_classification_facts(migration.policy, facts) assert evaluation.facts.identity.media_source == MediaSource.Douban.value assert all( warning.source == MediaSource.Douban.value for warning in evaluation.warnings ) def test_non_tmdb_tmdb_info_does_not_override_primary_standard_facts() -> None: """ 非 TMDB 媒体携带的旧 tmdb_info 不得覆盖主来源标准事实。 """ migration = migrate_legacy_category_config( { "movie": { "华语电影": {"original_language": "zh"}, "外语电影": None, }, "tv": {}, } ) media = MediaInfo( media_source=MediaSource.Douban, media_id="douban-1", type=MediaType.MOVIE, original_language="zh", tmdb_info={"original_language": "en"}, ) finalized = ClassificationExecutionService(_Runtime(migration.policy)).finalize(media) assert finalized.library_category == "华语电影" def test_execution_builds_complete_facts_without_mutating_media() -> None: """ 影响分析事实入口应复用插件字段构造，并保持原媒体对象不变。 """ source = MediaInfo( media_source=MediaSource.Douban, media_id="native-1", type=MediaType.MOVIE, title="Example", genres=[{"name": "动画"}], origin_country=["JP"], ) service = ClassificationExecutionService( _Runtime(_policy()), extension_facts_provider=lambda media: { "example.source": {"region_group": "east-asia"} }, ) facts = asyncio.run(service.async_build_facts(source)) assert facts is not None assert facts.identity.media_id == "native-1" assert facts.media.genre_keys == ["animation"] assert facts.media.countries == ["JP"] assert facts.extensions["example.source"]["region_group"] == "east-asia" assert source.classification is None def test_execution_reclassifies_cached_result_after_policy_revision_changes() -> None: """ 缓存对象携带旧 revision 时必须按当前策略重新分类。 """ runtime = _Runtime(_policy(revision=7)) service = ClassificationExecutionService(runtime) source = MediaInfo( media_source="douban", media_id="1", type=MediaType.MOVIE, origin_country=["JP"], genres=[{"name": "动画"}], ) first = service.finalize(source) runtime.policy = _policy(revision=8, movie_category="movie.other") second = service.finalize(first) assert second.classification is not None assert second.classification.policy_revision == 8 assert second.classification.effective.category_id == "movie.other" assert second.library_category == "其它电影" assert first.classification.policy_revision == 7 def test_execution_refreshes_same_revision_after_auxiliary_facts_change() -> None: """ 来源缓存恢复后即使 revision 相同，也必须按当前完整事实重新求值。 """ service = ClassificationExecutionService(_Runtime(_policy())) source = MediaInfo( media_source="anilist", media_id="1", type=MediaType.MOVIE, ) first = service.finalize(source) first.genre_ids = [16] first.origin_country = ["JP"] reclassified = service.finalize(first) refreshed = service.finalize(first, refresh=True) assert reclassified.library_category == "动画/日本" assert refreshed.library_category == "动画/日本" assert refreshed.classification is not None assert refreshed.classification.effective.rule_id == "rule.movie.jp" def test_execution_preserves_automatic_category_when_supplement_drops_facts() -> None: """ 裁剪过字段的媒体再次分类时，同策略已命中的自动分类不得被兜底覆盖。 """ migration = migrate_legacy_category_config( { "movie": {}, "tv": { "国产剧": {"origin_country": "CN,TW,HK"}, "未分类": None, }, } ) domestic = next( category.id for category in migration.policy.categories if category.name == "国产剧" ) service = ClassificationExecutionService(_Runtime(migration.policy)) source = MediaInfo( media_source=MediaSource.Douban, media_id="35593344", type=MediaType.TV, title="测试剧", production_countries=[{"name": "中国大陆"}], ) classified = service.finalize(source) assert classified.classification.effective.category_id == domestic assert classified.classification.state == "complete" classified.clear() supplemented = service.finalize(classified) assert supplemented.classification is not None assert supplemented.classification.effective.category_id == domestic assert supplemented.classification.effective.source == "automatic" assert supplemented.library_category == "国产剧" # 本次求值确实缺事实，推荐结果和状态仍如实记录，供预览解释差异。 assert supplemented.classification.state == "partial" assert supplemented.classification.recommended.source == "fallback" def test_execution_keeps_fallback_when_complete_facts_no_longer_match() -> None: """ 事实读齐后不再命中规则时，兜底结果必须生效而不是复用旧自动分类。 """ service = ClassificationExecutionService(_Runtime(_policy())) source = MediaInfo( media_source="douban", media_id="1", type=MediaType.MOVIE, origin_country=["JP"], genres=[{"name": "动画"}], ) classified = service.finalize(source) assert classified.classification.effective.category_id == "movie.jp" classified.origin_country = ["US"] reclassified = service.finalize(classified) assert reclassified.classification is not None assert reclassified.classification.state == "complete" assert reclassified.classification.effective.category_id == "movie.other" assert reclassified.classification.effective.source == "fallback" assert reclassified.library_category == "其它电影" def test_execution_applies_manual_effective_override_without_losing_recommendation() -> None: """ 订阅或目录人工覆盖只替换 effective，仍保留自动推荐供 UI 解释。 """ source = MediaInfo( media_source="douban", media_id="1", type=MediaType.MOVIE, origin_country=["JP"], genres=[{"name": "动画"}], ) override = ClassificationSelection( category_path=["收藏", "家庭电影"], source="subscription", ) finalized = ClassificationExecutionService(_Runtime(_policy())).finalize( source, effective_override=override, ) assert finalized.library_category == "收藏/家庭电影" assert finalized.classification is not None assert finalized.classification.recommended.category_id == "movie.jp" assert finalized.classification.effective.source == "subscription" assert finalized.classification.effective.category_path == ["收藏", "家庭电影"] def test_execution_preserves_subscription_override_during_auxiliary_refresh() -> None: """ 补充来源事实再次分类时必须保留已装配的订阅人工覆盖。 """ service = ClassificationExecutionService(_Runtime(_policy())) source = MediaInfo( media_source="douban", media_id="1", type=MediaType.MOVIE, origin_country=["JP"], genres=[{"name": "动画"}], ) overridden = service.finalize( source, effective_override=ClassificationSelection( category_id="movie.manual", category_path=["收藏", "家庭电影"], source="subscription", ), ) refreshed = service.finalize(overridden, refresh=True) assert refreshed.classification.recommended.category_id == "movie.jp" assert refreshed.classification.effective.category_id == "movie.manual" assert refreshed.classification.effective.source == "subscription" assert refreshed.library_category == "收藏/家庭电影" class _StaticEnrichment: """ 为同步和异步收口补充同一组国家事实。 """ def __init__(self) -> None: """ 创建空调用记录。 """ self.calls: list[str] = [] @staticmethod def _enrich(facts: ClassificationFacts) -> ClassificationFacts: """ 复制事实并填入当前缺失的国家。 """ enriched = facts.model_copy(deep=True) enriched.media.countries = ["JP"] return enriched def enrich( self, policy: ClassificationPolicy, facts: ClassificationFacts, media: object, ) -> ClassificationFacts: """ 记录同步调用并返回补充事实。 """ del policy, media self.calls.append("sync") return self._enrich(facts) async def async_enrich( self, policy: ClassificationPolicy, facts: ClassificationFacts, media: object, ) -> ClassificationFacts: """ 记录异步调用并返回同一补充事实。 """ del policy, media self.calls.append("async") return self._enrich(facts) def test_execution_sync_and_async_apply_the_same_enriched_facts() -> None: """ 同步和异步执行服务必须在同一求值边界消费补充事实。 """ source = MediaInfo( media_source=MediaSource.Douban, media_id="1291561", type=MediaType.MOVIE, title="千与千寻", genres=[{"name": "动画"}], ) enrichment = _StaticEnrichment() service = ClassificationExecutionService( _Runtime(_policy()), enrichment=enrichment, ) sync_result = service.finalize(source) async_result = asyncio.run(service.async_finalize(source)) assert enrichment.calls == ["sync", "async"] assert sync_result.classification == async_result.classification assert sync_result.library_category == async_result.library_category == "动画/日本" assert sync_result.media_source == async_result.media_source == MediaSource.Douban assert sync_result.media_id == async_result.media_id == "1291561" assert source.library_category == "" def test_offline_classification_keeps_policy_without_external_enrichment() -> None: """ 离线结果仍遵守用户分类和手工覆盖，但同步异步都不调用外部补充。 """ source = MusicInfo(title="晴天", artists=["周杰伦"], album="叶惠美") enrichment = _StaticEnrichment() service = ClassificationExecutionService(_Runtime(_policy()), enrichment=enrichment) override = ClassificationSelection( category_path=["收藏"], source="manual", category_id="music.other", ) sync_result = service.finalize(source, allow_enrichment=False, effective_override=override) async_result = asyncio.run(service.async_finalize( source, allow_enrichment=False, effective_override=override, )) assert enrichment.calls == [] assert sync_result.classification == async_result.classification assert sync_result.classification.recommended.category_id == "music.other" assert sync_result.library_category == async_result.library_category == "收藏" assert source.media_id is None assert source.library_category == "" def test_music_metadata_category_is_never_promoted_without_a_rule() -> None: """ 音乐来源描述分类只能形成事实，目录分类必须来自规则或兜底。 """ music = MusicInfo( media_source="musicbrainz", media_id="recording-1", title="Live Recording", metadata_category="Album / Live", secondary_types=["Live"], ) finalized = ClassificationExecutionService(_Runtime(_policy())).finalize(music) assert finalized.metadata_category == "Album / Live" assert finalized.library_category == "音乐/现场" assert finalized.classification.effective.rule_id == "rule.music.live" assert music.library_category == "" def test_artist_detail_uses_music_entity_facts() -> None: """ 完整艺术家详情应支持 music.entity_type 分类并保留描述类型。 """ artist = MusicArtistInfo( media_source="musicbrainz", media_id="artist-1", name="Example Band", artist_type="Group", ) finalized = ClassificationExecutionService(_Runtime(_policy())).finalize(artist) assert finalized.library_category == "音乐/艺术家" assert finalized.metadata_category == "Group" assert finalized.classification is not None assert finalized.classification.effective.rule_id == "rule.music.artist" def test_incomplete_identity_remains_not_evaluated() -> None: """ 本地音乐兜底缺少远端身份时不得伪造已完成分类。 """ music = MusicInfo( title="Local Track", metadata_category="Rock", ) finalized = ClassificationExecutionService(_Runtime(_policy())).finalize(music) assert finalized.classification is not None assert finalized.classification.state == "not_evaluated" assert finalized.classification.policy_revision == 7 assert finalized.library_category == "" assert finalized.metadata_category == "Rock" def test_active_legacy_policy_builds_tmdb_extension_facts_at_execution() -> None: """ 迁移策略使用的 TMDB 动态字段应在收口点从受控原始详情构造。 """ migration = migrate_legacy_category_config( CategoryConfig( movie={"特别电影": CategoryRule(custom_flag="yes")}, ) ) assert migration.valid is True media = MediaInfo( media_source="themoviedb", media_id="10", type=MediaType.MOVIE, tmdb_info={"id": 10, "media_type": "movie", "custom_flag": "yes"}, ) finalized = ClassificationExecutionService( _Runtime(migration.policy) ).finalize(media) assert finalized.library_category == "特别电影" assert finalized.classification is not None assert finalized.classification.effective.rule_id is not None def test_invalid_policy_uses_read_only_legacy_tmdb_fallback() -> None: """ 策略不可用时只对 TMDB 恢复旧分类，并明确标记 invalid_policy。 """ legacy = CategoryConfig( movie={ "日本动画": CategoryRule( genre_ids="16", production_countries="JP", ) } ) source = MediaInfo( media_source="themoviedb", media_id="20", type=MediaType.MOVIE, tmdb_info={ "id": 20, "media_type": "movie", "genre_ids": [16], "production_countries": [{"iso_3166_1": "JP"}], }, ) finalized = ClassificationExecutionService(_Runtime(None, legacy)).finalize(source) assert finalized.library_category == "日本动画" assert finalized.classification is not None assert finalized.classification.state == "invalid_policy" assert finalized.classification.effective.source == "legacy" class _OrderingClassificationService: """ 记录共享上报与分类收口顺序的测试执行服务。 """ def __init__(self, events: list[str]) -> None: """ 保存共享顺序记录列表。 """ self.events = events self.calls: list[MediaInfo] = [] def finalize( self, media: MediaInfo, *, effective_override: ClassificationSelection | None = None, refresh: bool = False, ) -> MediaInfo: """ 复制候选并写入可观察分类。 """ del effective_override, refresh self.events.append("finalize") self.calls.append(media) result = deepcopy(media) result.set_library_category("已分类") return result async def async_finalize( self, media: MediaInfo, *, effective_override: ClassificationSelection | None = None, refresh: bool = False, ) -> MediaInfo: """ 记录异步分类收口，并返回独立结果。 """ del effective_override, refresh self.events.append("async_finalize") self.calls.append(media) result = deepcopy(media) result.set_library_category("已分类") return result def _keep_candidate(**kwargs: object) -> Optional[MediaInfo]: """ 保持原生候选不变，隔离插件补充分支。 """ value = kwargs.get("mediainfo") return value if isinstance(value, MediaInfo) else None async def _async_keep_candidate(**kwargs: object) -> Optional[MediaInfo]: """ 异步保持原生候选不变。 """ return _keep_candidate(**kwargs) def test_sync_recognition_finalizes_after_share_report() -> None: """ 同步完整识别应先完成共享上报，再通过应用服务分类返回副本。 """ chain = ChainBase() events: list[str] = [] classifier = _OrderingClassificationService(events) chain.classification_service = classifier candidate = MediaInfo( media_source="themoviedb", media_id="30", type=MediaType.MOVIE, title="Example", ) share = Mock() share.report_recognize_share.side_effect = lambda **_kwargs: events.append("report") with patch.object(chain, "_run_native_media_recognize", return_value=candidate), patch.object( chain, "_supplement_media_recognize", side_effect=_keep_candidate, ), patch("app.chain._recognition._recognition_share_snapshot", return_value=share): result = chain.recognize_media(meta=MetaInfo("Example"), cache=False) assert events == ["report", "finalize"] assert classifier.calls == [candidate] assert isinstance(result, MediaInfo) assert result is not candidate assert result.library_category == "已分类" def test_async_recognition_uses_the_same_finalize_boundary() -> None: """ 异步完整识别应调用同一应用服务的有界异步分类收口。 """ chain = ChainBase() events: list[str] = [] classifier = _OrderingClassificationService(events) chain.classification_service = classifier candidate = MediaInfo( media_source="douban", media_id="40", type=MediaType.TV, title="Example TV", ) share = Mock() share.async_report_recognize_share = AsyncMock( side_effect=lambda **_kwargs: events.append("report") ) async def exercise() -> MediaInfo | None: """ 执行异步识别并返回分类后的媒体。 """ with patch.object( chain, "_async_run_native_media_recognize", AsyncMock(return_value=candidate), ), patch.object( chain, "_async_supplement_media_recognize", side_effect=_async_keep_candidate, ), patch( "app.chain._recognition._recognition_share_snapshot", return_value=share, ): return await chain.async_recognize_media( meta=MetaInfo("Example TV"), cache=False, ) result = asyncio.run(exercise()) assert events == ["report", "async_finalize"] assert classifier.calls == [candidate] assert isinstance(result, MediaInfo) assert result is not candidate assert result.library_category == "已分类"