From 889bdfe501bb0b1e43950ae600913643ee79ff9c Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 12 Sep 2026 17:43:36 +0800 Subject: [PATCH] fix(classification): align migration defaults and source semantics --- app/api/endpoints/classification.py | 22 +- app/application/classification/analysis.py | 152 ++++++-- .../classification/configuration.py | 195 +++++++++- app/application/classification/execution.py | 69 +++- app/application/classification/migration.py | 342 ++++++++++++++++-- app/application/classification/projection.py | 91 ++++- app/db/adapters/classification.py | 96 ++++- app/domain/classification/validation.py | 26 ++ app/domain/projection/tmdb.py | 7 +- tests/test_media_classification.py | 17 + tests/test_media_classification_api.py | 60 ++- ...test_media_classification_configuration.py | 6 +- ...ia_classification_configuration_adapter.py | 33 ++ tests/test_media_classification_execution.py | 51 ++- tests/test_media_classification_legacy.py | 121 +++++-- tests/test_media_classification_startup.py | 51 ++- 16 files changed, 1189 insertions(+), 150 deletions(-) diff --git a/app/api/endpoints/classification.py b/app/api/endpoints/classification.py index 37fadd126..1e6a1a410 100644 --- a/app/api/endpoints/classification.py +++ b/app/api/endpoints/classification.py @@ -6,7 +6,7 @@ from enum import Enum from functools import partial from typing import cast -from fastapi import Depends, HTTPException, status +from fastapi import Depends, HTTPException, Query, status from fastapi.responses import JSONResponse from app.api.context import ( @@ -27,6 +27,7 @@ from app.application.classification.configuration import ( ClassificationPolicyNotInitializedError, ClassificationPolicyRevisionNotFoundError, ClassificationPolicyValidationError, + build_default_classification_policy, ) from app.application.classification.contract import ( ClassificationPolicyConflictError, @@ -77,6 +78,7 @@ def _get_analysis_service( """组装分类分析服务及其只读近期历史样本端口。""" return ClassificationAnalysisService( runtime.classification.service, + execution=runtime.classification_execution, sample_provider=RecentHistoryClassificationSampleProvider( download_history=cast( DownloadHistoryQueryPort, @@ -198,8 +200,19 @@ def _validation_response( async def get_policy( _: object = Depends(get_current_active_user_async), runtime: ClassificationRuntime = Depends(get_classification_runtime), + template: str | None = Query( + default=None, + description="读取内置分类策略模板时传入 default；省略则读取活动策略", + ), ) -> ClassificationPolicy: - """返回与运行时内部引用隔离的活动策略和 revision。""" + """返回活动策略，或返回不写入运行时的内置默认策略草稿。""" + if template == "default": + return build_default_classification_policy() + if template is not None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="不支持的分类策略模板", + ) return _require_active_policy(runtime) @@ -278,7 +291,10 @@ async def preview_policy( if request.policy is None: _require_active_policy(runtime) try: - return ClassificationAnalysisService(runtime.service).preview(request) + return ClassificationAnalysisService( + runtime.service, + execution=runtime.classification_execution, + ).preview(request) except ClassificationPolicyValidationError as error: return _validation_response(error) diff --git a/app/application/classification/analysis.py b/app/application/classification/analysis.py index c1641c9d5..923e8ff99 100644 --- a/app/application/classification/analysis.py +++ b/app/application/classification/analysis.py @@ -20,14 +20,18 @@ from app.application.classification.configuration import ( ClassificationPolicyValidationError, ) from app.application.classification.contract import ClassificationPolicyConflictError +from app.application.classification.execution import ( + ClassificationExecutionPort, + evaluate_classification_facts, +) from app.application.history import ( DownloadHistoryQueryPort, DownloadHistorySnapshot, TransferHistoryQueryPort, TransferHistorySnapshot, ) -from app.domain.classification.evaluator import ClassificationEvaluator from app.domain.classification.facts import build_classification_facts +from app.domain.classification.fields import field_definition_map from app.domain.classification.validation import ( MAX_CATEGORY_DEPTH, MAX_CATEGORY_PATH_LENGTH, @@ -238,10 +242,12 @@ class ClassificationAnalysisService: configuration: ClassificationPolicyConfigurationService, *, sample_provider: ClassificationImpactSampleProvider | None = None, + execution: ClassificationExecutionPort | None = None, ) -> None: - """保存策略配置服务和可选近期样本提供器。""" + """保存策略配置、样本提供器和可选的统一执行事实端口。""" self._configuration = configuration self._sample_provider = sample_provider + self._execution = execution def fields(self) -> ClassificationFieldCatalog: """分开返回新规则可选字段和已有规则使用的退役字段。""" @@ -276,10 +282,31 @@ class ClassificationAnalysisService: policy = request.policy or self._configuration.active() if request.policy is not None: self._require_valid(policy) - return ClassificationEvaluator.evaluate( - policy, - _preview_facts(request.input), - trace=True, + facts = self._preview_facts( + request.input, + policy=policy, + use_execution=self._execution is not None, + ) + return evaluate_classification_facts(policy, facts, trace=True) + + def _preview_facts( + self, + input_data: ClassificationPreviewInput, + *, + policy: ClassificationPolicy, + use_execution: bool, + ) -> ClassificationFacts: + """构造预览事实；活动策略优先复用真实执行端口的补充链路。""" + if input_data.kind == "facts": + return input_data.facts + media = _preview_media(input_data.media) + if use_execution and self._execution is not None: + facts = self._execution.build_facts(cast(Any, media), policy=policy) + if facts is not None: + return facts + return build_classification_facts_from_media( + media, + extra_fields=self._configuration.extra_fields(), ) async def impact( @@ -324,22 +351,45 @@ class ClassificationAnalysisService: ) -> ClassificationImpactSampleBatch: """优先使用请求事实，否则委托近期历史提供器生成样本。""" if samples: - selected = tuple( - cast(ClassificationFacts, sample.model_copy(deep=True)) - for sample in samples[:sample_limit] - ) + selected_list: list[ClassificationFacts] = [] + seen: set[tuple[str, str, str, str]] = set() + skipped_count = 0 + for sample in samples: + identity = sample.identity + identity_key = _classification_identity_key(sample) + if not identity.media_source.strip() or not identity.media_id.strip(): + skipped_count += 1 + continue + if identity_key in seen: + skipped_count += 1 + continue + if len(selected_list) >= sample_limit: + skipped_count += 1 + continue + seen.add(identity_key) + selected_list.append(sample.model_copy(deep=True)) + selected = tuple(selected_list) warnings = ( - (f"显式事实共 {len(samples)} 条，仅比较前 {sample_limit} 条",) - if len(samples) > sample_limit - else () + tuple( + item + for item in ( + f"显式事实共 {len(samples)} 条，仅比较前 {sample_limit} 条" + if len(samples) > sample_limit + else "", + "显式事实中存在重复或缺少稳定身份的记录，已跳过" + if skipped_count + else "", + ) + if item + ) ) return ClassificationImpactSampleBatch( source="request", facts=selected, scanned_count=len(samples), - skipped_count=0, + skipped_count=skipped_count, unresolved_count=0, - truncated=len(samples) > sample_limit, + truncated=skipped_count > 0 or len(samples) > sample_limit, warnings=warnings, ) if self._sample_provider is None: @@ -393,48 +443,78 @@ def _classification_identity_key(facts: ClassificationFacts) -> tuple[str, str, ) -def _preview_facts(input_data: ClassificationPreviewInput) -> ClassificationFacts: - """根据预览输入选择兼容事实或媒体搜索结果转换器。""" - if input_data.kind == "facts": - return input_data.facts - return build_classification_facts_from_media_payload(input_data.media) +def _preview_media(payload: Mapping[str, Any]) -> object: + """把预览媒体载荷转换为可供执行服务消费的媒体对象。""" + media_type = _enum_text(payload.get("type")) + model = SchemaMusicInfo if media_type == "音乐" else SchemaMediaInfo + try: + return model.model_validate(dict(payload)) + except ValueError: + # 插件来源不一定属于内置 MediaSource 枚举，使用轻量对象保留其完整字段。 + return SimpleNamespace(**dict(payload)) -def build_classification_facts_from_media(media: object) -> ClassificationFacts: +def build_classification_facts_from_media( + media: object, + *, + extra_fields: Sequence[object] = (), +) -> ClassificationFacts: """把搜索或识别得到的完整媒体对象转换为统一分类数据。""" return build_classification_facts( cast(Any, media), - extensions=_media_extension_facts(media), + extensions=_media_extension_facts(media, extra_fields=extra_fields), ) def build_classification_facts_from_media_payload( payload: Mapping[str, Any], + *, + extra_fields: Sequence[object] = (), ) -> ClassificationFacts: """把前端选择的媒体搜索结果转换为统一分类数据，并兼容插件来源。""" - media_type = _enum_text(payload.get("type")) - model = SchemaMusicInfo if media_type == "音乐" else SchemaMediaInfo - try: - media = model.model_validate(dict(payload)) - except ValueError: - # 插件来源不一定属于内置 MediaSource 枚举，使用轻量对象保留其完整字段。 - media = SimpleNamespace(**dict(payload)) - return build_classification_facts_from_media(media) + return build_classification_facts_from_media( + _preview_media(payload), + extra_fields=extra_fields, + ) -def _media_extension_facts(media: object) -> dict[str, dict[str, ClassificationFactValue]]: +def _media_extension_facts( + media: object, + *, + extra_fields: Sequence[object] = (), +) -> dict[str, dict[str, ClassificationFactValue]]: """按 extensions.. 命名空间整理媒体携带的扩展字段。""" raw_facts = getattr(media, "classification_facts", None) if not isinstance(raw_facts, Mapping): return {} + definitions = field_definition_map(cast(Any, extra_fields)) extensions: dict[str, dict[str, ClassificationFactValue]] = {} for raw_field, value in raw_facts.items(): - parts = str(raw_field or "").split(".", 2) - if len(parts) != 3 or parts[0] != "extensions" or not parts[1] or not parts[2]: + field_id = str(raw_field or "").strip() + if not field_id.startswith("extensions."): continue if not _is_classification_fact_value(value): continue - extensions.setdefault(parts[1], {})[parts[2]] = cast(ClassificationFactValue, value) + definition = definitions.get(field_id) + if definition is None: + parts = field_id.split(".", 2) + if len(parts) != 3 or not parts[1] or not parts[2]: + continue + extensions.setdefault(parts[1], {})[parts[2]] = cast( + ClassificationFactValue, value + ) + continue + extension_sources = [ + source + for source, support in definition.source_support.items() + if support == "extension" + ] + if len(extension_sources) != 1: + continue + source = extension_sources[0] + local_field = field_id.removeprefix(f"extensions.{source}.") + if source and local_field: + extensions.setdefault(source, {})[local_field] = cast(ClassificationFactValue, value) return extensions @@ -494,8 +574,8 @@ def _build_impact_analysis( rule_changed_only_count = 0 became_fallback_count = 0 for facts in batch.facts: - previous = ClassificationEvaluator.evaluate(active, facts).result - proposed = ClassificationEvaluator.evaluate(candidate, facts).result + previous = evaluate_classification_facts(active, facts).result + proposed = evaluate_classification_facts(candidate, facts).result previous_categories[_category_id(previous)] += 1 candidate_categories[_category_id(proposed)] += 1 if "partial" in {previous.state, proposed.state}: diff --git a/app/application/classification/configuration.py b/app/application/classification/configuration.py index 684b6f170..d39c0e6f7 100644 --- a/app/application/classification/configuration.py +++ b/app/application/classification/configuration.py @@ -20,6 +20,7 @@ from app.domain.classification.validation import ClassificationPolicyValidator from app.schemas.category import ( ClassificationCategory, ClassificationCondition, + ClassificationConditionGroup, ClassificationFieldDefinition, ClassificationOperator, ClassificationPolicy, @@ -42,6 +43,196 @@ _DEFAULT_MUSIC_CATEGORIES = ( ("music.single", "Single", ("Single",)), ) +_DEFAULT_MEDIA_CATEGORIES = ( + ("movie.animation", "电影", "动画电影", ("动画电影",)), + ("movie.chinese", "电影", "华语电影", ("华语电影",)), + ("movie.foreign", "电影", "外语电影", ("外语电影",)), + ("tv.dongman.cn", "电视剧", "国漫", ("国漫",)), + ("tv.dongman.jp", "电视剧", "日番", ("日番",)), + ("tv.documentary", "电视剧", "纪录片", ("纪录片",)), + ("tv.kids", "电视剧", "儿童", ("儿童",)), + ("tv.variety", "电视剧", "综艺", ("综艺",)), + ("tv.chinese", "电视剧", "国产剧", ("国产剧",)), + ("tv.western", "电视剧", "欧美剧", ("欧美剧",)), + ("tv.asian", "电视剧", "日韩剧", ("日韩剧",)), + ("tv.uncategorized", "电视剧", "未分类", ("未分类",)), + ("music.uncategorized", "音乐", "未分类", ("未分类",)), +) + + +def build_builtin_classification_policy() -> ClassificationPolicy: + """构造与仓库内 legacy category.yaml 默认值等效的标准分类策略。""" + categories = [ + ClassificationCategory( + id=category_id, + media_type=media_type, + name=name, + path=list(path), + ) + for category_id, media_type, name, path in _DEFAULT_MEDIA_CATEGORIES + ] + rules = [ + ClassificationRule( + id="movie.animation.default", + name="动画电影", + kind="category", + priority=0, + media_types=["电影"], + when=ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["animation"], + ), + target=ClassificationTarget(category_id="movie.animation"), + ), + ClassificationRule( + id="movie.chinese.default", + name="华语电影", + kind="category", + priority=1, + media_types=["电影"], + when=ClassificationCondition( + field="media.language", + operator="in", + value=["zh", "cn", "bo", "za"], + ), + target=ClassificationTarget(category_id="movie.chinese"), + ), + ClassificationRule( + id="tv.dongman.cn.default", + name="国漫", + kind="category", + priority=2, + media_types=["电视剧"], + when=ClassificationConditionGroup( + all=[ + ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["animation"], + ), + ClassificationCondition( + field="media.countries", + operator="contains_any", + value=["CN", "TW", "HK"], + ), + ] + ), + target=ClassificationTarget(category_id="tv.dongman.cn"), + ), + ClassificationRule( + id="tv.dongman.jp.default", + name="日番", + kind="category", + priority=3, + media_types=["电视剧"], + when=ClassificationConditionGroup( + all=[ + ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["animation"], + ), + ClassificationCondition( + field="media.countries", + operator="contains_any", + value=["JP"], + ), + ] + ), + target=ClassificationTarget(category_id="tv.dongman.jp"), + ), + ClassificationRule( + id="tv.documentary.default", + name="纪录片", + kind="category", + priority=4, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["documentary"], + ), + target=ClassificationTarget(category_id="tv.documentary"), + ), + ClassificationRule( + id="tv.kids.default", + name="儿童", + kind="category", + priority=5, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["kids"], + ), + target=ClassificationTarget(category_id="tv.kids"), + ), + ClassificationRule( + id="tv.variety.default", + name="综艺", + kind="category", + priority=6, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.genre_keys", + operator="contains_any", + value=["reality", "talk"], + ), + target=ClassificationTarget(category_id="tv.variety"), + ), + ClassificationRule( + id="tv.chinese.default", + name="国产剧", + kind="category", + priority=7, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.countries", + operator="contains_any", + value=["CN", "TW", "HK"], + ), + target=ClassificationTarget(category_id="tv.chinese"), + ), + ClassificationRule( + id="tv.western.default", + name="欧美剧", + kind="category", + priority=8, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.countries", + operator="contains_any", + value=["US", "FR", "GB", "DE", "ES", "IT", "NL", "PT", "RU", "UK"], + ), + target=ClassificationTarget(category_id="tv.western"), + ), + ClassificationRule( + id="tv.asian.default", + name="日韩剧", + kind="category", + priority=9, + media_types=["电视剧"], + when=ClassificationCondition( + field="media.countries", + operator="contains_any", + value=["JP", "KP", "KR", "TH", "IN", "SG"], + ), + target=ClassificationTarget(category_id="tv.asian"), + ), + ] + return with_default_music_classification( + ClassificationPolicy( + categories=categories, + rules=rules, + fallbacks={ + "电影": "movie.foreign", + "电视剧": "tv.uncategorized", + "音乐": "music.uncategorized", + }, + ) + ) + class ClassificationPolicyNotInitializedError(RuntimeError): """表示分类策略服务尚未加载或初始化活动快照。""" @@ -191,9 +382,7 @@ def needs_default_music_classification(policy: ClassificationPolicy) -> bool: def build_default_classification_policy() -> ClassificationPolicy: """构造带稳定兜底和常用音乐专辑分类的初始草稿。""" - return with_default_music_classification( - _build_uncategorized_classification_policy() - ) + return build_builtin_classification_policy() class ClassificationPolicyConfigurationService: diff --git a/app/application/classification/execution.py b/app/application/classification/execution.py index 7f8bf7867..41dc5e672 100644 --- a/app/application/classification/execution.py +++ b/app/application/classification/execution.py @@ -61,6 +61,15 @@ class ClassificationExecutionPort(Protocol): """异步构造与实际分类一致的完整事实快照，不写入媒体或策略。""" ... + def build_facts( + self, + media: ClassificationSubject, + *, + policy: ClassificationPolicy | None = None, + ) -> ClassificationFacts | None: + """同步构造与实际分类一致的完整事实快照，不写入媒体或策略。""" + ... + def finalize( self, media: ClassificationSubject, @@ -149,6 +158,30 @@ class ClassificationExecutionService: pass return facts + def build_facts( + self, + media: ClassificationSubject, + *, + policy: ClassificationPolicy | None = None, + ) -> ClassificationFacts | None: + """同步构造并按当前策略补充缺失事实，不修改媒体对象。""" + finalized, active_policy, facts, _ = self._prepare( + media, + extensions=None, + effective_override=None, + refresh=True, + policy_override=policy, + ) + if active_policy is None or facts is None: + return None + policy = policy or active_policy + if self._enrichment is not None: + try: + facts = self._enrichment.enrich(policy, facts, finalized) + except Exception: # noqa: BLE001 详情补充失败时保留主来源事实 + pass + return facts + def finalize( self, media: ClassificationSubject, @@ -226,6 +259,7 @@ class ClassificationExecutionService: extensions: Mapping[str, Mapping[str, ClassificationFactValue]] | None, effective_override: ClassificationSelection | None, refresh: bool, + policy_override: ClassificationPolicy | None = None, ) -> tuple[ ClassificationSubject, ClassificationPolicy | None, @@ -236,7 +270,7 @@ class ClassificationExecutionService: del refresh effective_override = effective_override or _explicit_effective_override(media) finalized = deepcopy(media) - policy = self._runtime.active_policy() + policy = policy_override or self._runtime.active_policy() if policy is None: self._apply_invalid_policy_fallback( finalized, @@ -281,10 +315,7 @@ class ClassificationExecutionService: effective_override: ClassificationSelection | None, ) -> 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 + evaluation = evaluate_classification_facts(policy, facts) result = evaluation.result.model_copy(deep=True) if effective_override: result.effective = effective_override.model_copy(deep=True) @@ -356,12 +387,16 @@ class ClassificationExecutionService: def _evaluate_legacy_tmdb_compatibility( policy: ClassificationPolicy, facts: ClassificationFacts, + *, + trace: bool = False, ) -> 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) + rule.model_copy(deep=True, update={"sources": []}) + for rule in policy.rules + if rule.id.startswith(_LEGACY_RULE_PREFIX) ] if not legacy_rules: return None @@ -373,6 +408,7 @@ def _evaluate_legacy_tmdb_compatibility( return ClassificationEvaluator.evaluate( compatibility_policy, compatibility_facts, + trace=trace, ) @@ -393,18 +429,29 @@ def _legacy_tmdb_compatibility_facts( 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}, + update={"extensions": extensions}, ), ) +def evaluate_classification_facts( + policy: ClassificationPolicy, + facts: ClassificationFacts, + *, + trace: bool = False, +) -> ClassificationEvaluation: + """按真实执行路径求值事实，并兼容非 TMDB 来源的旧 TMDB 规则。""" + evaluation = ClassificationEvaluator.evaluate(policy, facts, trace=trace) + legacy_evaluation = _evaluate_legacy_tmdb_compatibility(policy, facts, trace=trace) + if legacy_evaluation is not None and _uses_fallback(evaluation.result): + return legacy_evaluation + return evaluation + + def _legacy_tmdb_info_from_standard_facts( facts: ClassificationFacts, ) -> dict[str, object]: @@ -452,7 +499,7 @@ def _classification_extensions( for source, values in (supplied or {}).items() } tmdb_info = getattr(media, "tmdb_info", None) - if isinstance(tmdb_info, Mapping): + if _enum_text(getattr(media, "media_source", None)) == _LEGACY_TMDB_SOURCE and isinstance(tmdb_info, Mapping): for source, values in build_legacy_tmdb_extension_facts( policy, tmdb_info, diff --git a/app/application/classification/migration.py b/app/application/classification/migration.py index ee598a6a6..9e1c43e4e 100644 --- a/app/application/classification/migration.py +++ b/app/application/classification/migration.py @@ -6,8 +6,11 @@ import hashlib import re from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field -from typing import Final, Literal, Optional, TypeAlias, Union, cast +from typing import Callable, Final, Literal, Optional, TypeAlias, Union, cast +from app.domain.classification.vocabulary import ( + COUNTRY_CODE_ALIASES as _COUNTRY_CODE_ALIASES, +) from app.domain.classification.vocabulary import ( TMDB_GENRE_KEYS as _TMDB_GENRE_KEYS, ) @@ -154,6 +157,21 @@ class _MigrationContext: if value != normalized: aliases[value] = normalized + def register_standard_aliases( + self, + field_id: str, + values: Iterable[str], + *, + canonicalizer: Callable[[str], str] | None = None, + ) -> None: + """登记标准字符串字段的大小写别名，保持旧配置的无大小写比较语义。""" + aliases = self.field_aliases.setdefault(field_id, {}) + for value in values: + canonical = canonicalizer(value) if canonicalizer else value.casefold() + for alias in (value, value.upper(), value.capitalize()): + if alias != canonical: + aliases[alias] = canonical + def build_field_definitions(self) -> tuple[ClassificationFieldDefinition, ...]: """按首次出现顺序构造仅供已有规则解析的退役字段声明。""" return tuple( @@ -377,6 +395,15 @@ def _migrate_media_categories( nodes.append(_tmdb_identity_condition()) category_has_error = True rule_kind, target = _retained_rule_output(category_id, unreachable) + rule_sources = ( + [_TMDB_SOURCE] + if any( + field_id.startswith(_EXTENSION_PREFIX) + for node in nodes + for field_id in _condition_field_ids(node) + ) + else [] + ) rules.append( ClassificationRule( id=f"{category_id}.rule", @@ -385,7 +412,7 @@ def _migrate_media_categories( enabled=not unreachable and not category_has_error, priority=category_index, media_types=[media_type], - sources=[_TMDB_SOURCE], + sources=rule_sources, when=_all_or_single(nodes), target=target, ) @@ -488,7 +515,257 @@ def _migrate_legacy_field( context=context, ) - field_id = context.register_extension_field(raw_field, media_type) + if raw_field == "original_language": + return _standard_string_condition( + "media.language", + tokens, + requires_exists, + context=context, + ) + if raw_field in {"origin_country", "production_countries"}: + return _standard_country_condition( + tokens, + requires_exists, + context=context, + ) + if raw_field == "release_year": + return _standard_numeric_condition( + "media.year", + tokens, + requires_exists, + media_type=media_type, + extension_field="release_year", + context=context, + ) + if raw_field == "runtime": + return _standard_numeric_condition( + "media.runtime", + tokens, + requires_exists, + media_type=media_type, + extension_field="runtime", + context=context, + ) + if raw_field == "adult": + return _standard_boolean_condition( + tokens, + requires_exists, + media_type=media_type, + context=context, + ) + + return _legacy_extension_condition( + raw_field, + tokens, + requires_exists, + media_type=media_type, + context=context, + ) + + +def _standard_string_condition( + field_id: str, + tokens: Sequence[_LegacyToken], + requires_exists: bool, + *, + context: _MigrationContext, +) -> ClassificationConditionNode: + """把旧字符串枚举转换为标准字段的 in/not_in 条件。""" + context.register_standard_aliases( + field_id, + (value for token in tokens for value in token.values), + ) + positives = [ + ClassificationCondition( + field=field_id, + operator="in", + value=[value.casefold() for value in token.values], + ) + for token in tokens + if not token.negative + ] + negatives = [ + ClassificationCondition( + field=field_id, + operator="not_in", + value=[value.casefold() for value in token.values], + ) + for token in tokens + if token.negative + ] + nodes: list[ClassificationConditionNode] = [*positives, *negatives] + if not nodes and requires_exists: + return ClassificationCondition(field=field_id, operator="exists") + return _all_or_single(nodes) + + +def _standard_country_condition( + tokens: Sequence[_LegacyToken], + requires_exists: bool, + *, + context: _MigrationContext, +) -> ClassificationConditionNode: + """把旧电影/电视剧国家枚举转换为统一国家代码列表条件。""" + context.register_standard_aliases( + "media.countries", + (value for token in tokens for value in token.values), + canonicalizer=_standard_country_code, + ) + positives = [ + ClassificationCondition( + field="media.countries", + operator="contains_any", + value=[_standard_country_code(value) for value in token.values], + ) + for token in tokens + if not token.negative + ] + negatives = [ + ClassificationCondition( + field="media.countries", + operator="contains_none", + value=[_standard_country_code(value) for value in token.values], + ) + for token in tokens + if token.negative + ] + nodes: list[ClassificationConditionNode] = [*positives, *negatives] + if not nodes and requires_exists: + return ClassificationCondition(field="media.countries", operator="exists") + return _all_or_single(nodes) + + +def _standard_numeric_condition( + field_id: str, + tokens: Sequence[_LegacyToken], + requires_exists: bool, + *, + media_type: ClassificationMediaType, + extension_field: str, + context: _MigrationContext, +) -> ClassificationConditionNode: + """把可表示的旧数字枚举迁移为标准数值条件，其余值留在 TMDB 扩展字段。""" + positive_nodes: list[ClassificationConditionNode] = [] + negative_nodes: list[ClassificationConditionNode] = [] + extension_field_id: str | None = None + for token in tokens: + numeric_values: list[int] = [] + unknown_values: list[str] = [] + for value in token.values: + if value.isdigit(): + numeric_values.append(int(value)) + else: + unknown_values.append(value) + token_nodes: list[ClassificationConditionNode] = [] + if numeric_values: + token_nodes.append( + ClassificationCondition( + field=field_id, + operator="not_in" if token.negative else "in", + value=numeric_values, + ) + ) + if unknown_values: + if extension_field_id is None: + extension_field_id = context.register_extension_field( + extension_field, + media_type, + ) + context.register_aliases(extension_field_id, unknown_values) + token_nodes.append( + ClassificationCondition( + field=extension_field_id, + operator=("contains_none" if token.negative else "contains_any"), + value=unknown_values, + ) + ) + if not token_nodes: + continue + if token.negative: + negative_nodes.extend(token_nodes) + else: + positive_nodes.append(_any_or_single(token_nodes)) + + nodes: list[ClassificationConditionNode] = [] + if positive_nodes: + nodes.append(_any_or_single(positive_nodes)) + nodes.extend(negative_nodes) + if not nodes and requires_exists: + if extension_field_id is not None: + return ClassificationCondition(field=extension_field_id, operator="exists") + return ClassificationCondition(field=field_id, operator="exists") + return _all_or_single(nodes) + + +def _standard_boolean_condition( + tokens: Sequence[_LegacyToken], + requires_exists: bool, + *, + media_type: ClassificationMediaType, + context: _MigrationContext, +) -> ClassificationConditionNode: + """把旧 TRUE/FALSE 条件迁移为标准成人标记，未知值继续使用 TMDB 扩展。""" + positive_nodes: list[ClassificationConditionNode] = [] + negative_nodes: list[ClassificationConditionNode] = [] + extension_field_id: str | None = None + true_values = {"true", "1", "yes"} + false_values = {"false", "0", "no"} + for token in tokens: + standard_nodes: list[ClassificationConditionNode] = [] + unknown_values: list[str] = [] + for value in token.values: + normalized = value.casefold() + if normalized in true_values | false_values: + is_true = normalized in true_values + if token.negative: + is_true = not is_true + standard_nodes.append( + ClassificationCondition( + field="media.adult", + operator="is_true" if is_true else "is_false", + ) + ) + else: + unknown_values.append(value) + if unknown_values: + if extension_field_id is None: + extension_field_id = context.register_extension_field("adult", media_type) + context.register_aliases(extension_field_id, unknown_values) + standard_nodes.append( + ClassificationCondition( + field=extension_field_id, + operator=("contains_none" if token.negative else "contains_any"), + value=unknown_values, + ) + ) + if not standard_nodes: + continue + if token.negative: + negative_nodes.extend(standard_nodes) + else: + positive_nodes.append(_any_or_single(standard_nodes)) + + nodes: list[ClassificationConditionNode] = [] + if positive_nodes: + nodes.append(_any_or_single(positive_nodes)) + nodes.extend(negative_nodes) + if not nodes and requires_exists: + if extension_field_id is not None: + return ClassificationCondition(field=extension_field_id, operator="exists") + return ClassificationCondition(field="media.adult", operator="exists") + return _all_or_single(nodes) + + +def _legacy_extension_condition( + field_name: str, + tokens: Sequence[_LegacyToken], + requires_exists: bool, + *, + media_type: ClassificationMediaType, + context: _MigrationContext, +) -> ClassificationConditionNode: + """把没有标准等价字段的旧值登记为来源受限扩展条件。""" + field_id = context.register_extension_field(field_name, media_type) context.register_aliases( field_id, (value for token in tokens for value in token.values), @@ -496,6 +773,14 @@ def _migrate_legacy_field( return _legacy_list_condition(field_id, tokens, requires_exists) +def _standard_country_code(value: str) -> str: + """把旧国家枚举值归一为标准事实使用的 ISO 代码。""" + normalized = value.strip() + if len(normalized) == 2 and normalized.isascii() and normalized.isalpha(): + return normalized.upper() + return _COUNTRY_CODE_ALIASES.get(normalized.casefold(), normalized.upper()) + + def _legacy_field_definition( field_id: str, media_types: list[ClassificationMediaType], @@ -592,60 +877,47 @@ def _migrate_genre_tokens( media_type: ClassificationMediaType, context: _MigrationContext, ) -> ClassificationConditionNode: - """已知正向 Genre ID 使用规范风格，其余条件保留原始视图。""" + """已知 Genre ID 使用标准风格键，混入未知值时整体保留 TMDB 原始视图。""" + if any( + value.upper() not in _TMDB_GENRE_KEYS + for token in tokens + for value in token.values + ): + extension_field_id = context.register_extension_field("genre_ids", media_type) + context.register_aliases( + extension_field_id, + (value for token in tokens for value in token.values), + ) + return _legacy_list_condition(extension_field_id, tokens, requires_exists) + positive_nodes: list[ClassificationConditionNode] = [] negative_nodes: list[ClassificationConditionNode] = [] - extension_field_id: Optional[str] = None for token in tokens: - if token.negative: - if extension_field_id is None: - extension_field_id = context.register_extension_field("genre_ids", media_type) - context.register_aliases(extension_field_id, token.values) - negative_nodes.append( - ClassificationCondition( - field=extension_field_id, - operator="contains_none", - value=list(token.values), - ) - ) - continue known_keys: list[str] = [] - unknown_ids: list[str] = [] for value in token.values: if genre_key := _TMDB_GENRE_KEYS.get(value.upper()): _append_unique(known_keys, genre_key) - else: - _append_unique(unknown_ids, value) token_nodes: list[ClassificationConditionNode] = [] if known_keys: token_nodes.append( ClassificationCondition( field="media.genre_keys", - operator="contains_any", + operator="contains_none" if token.negative else "contains_any", value=known_keys, ) ) - if unknown_ids: - if extension_field_id is None: - extension_field_id = context.register_extension_field("genre_ids", media_type) - context.register_aliases(extension_field_id, unknown_ids) - token_nodes.append( - ClassificationCondition( - field=extension_field_id, - operator="contains_any", - value=unknown_ids, - ) - ) if token_nodes: - positive_nodes.append(_any_or_single(token_nodes)) + if token.negative: + negative_nodes.extend(token_nodes) + else: + positive_nodes.append(_any_or_single(token_nodes)) nodes: list[ClassificationConditionNode] = [] if positive_nodes: nodes.append(_any_or_single(positive_nodes)) nodes.extend(negative_nodes) if not nodes and requires_exists: - extension_field_id = context.register_extension_field("genre_ids", media_type) - return ClassificationCondition(field=extension_field_id, operator="exists") + return ClassificationCondition(field="media.genre_keys", operator="exists") return _all_or_single(nodes) diff --git a/app/application/classification/projection.py b/app/application/classification/projection.py index ca383e587..7b07f318c 100644 --- a/app/application/classification/projection.py +++ b/app/application/classification/projection.py @@ -157,11 +157,11 @@ def resolve_legacy_tmdb_category( def _category_rules(policy: ClassificationPolicy) -> dict[str, ClassificationRule]: - """按目标分类 ID 索引 TMDB 主分类规则，保持首条规则优先。""" + """按迁移分类目标 ID 索引主分类规则，保持首条规则优先。""" result: dict[str, ClassificationRule] = {} for rule in policy.rules: category_id = rule.target.category_id or _archived_category_id(rule) - if category_id and _TMDB_SOURCE in rule.sources: + if category_id and category_id.startswith("legacy."): result.setdefault(category_id, rule) return result @@ -211,6 +211,7 @@ def _project_rule( policy, diagnostics, rule.id, + media_type=rule.media_types[0] if rule.media_types else None, ) if field_name is None: continue @@ -250,9 +251,19 @@ def _project_condition( policy: ClassificationPolicy, diagnostics: list[LegacyClassificationDiagnostic], rule_id: str, + *, + media_type: Optional[str], ) -> tuple[Optional[str], list[str], bool]: """把一个迁移器叶子恢复为旧字段、值集合和排除标志。""" - if condition.operator not in {"contains_any", "contains_none", "exists"}: + if condition.operator not in { + "contains_any", + "contains_none", + "exists", + "in", + "not_in", + "is_true", + "is_false", + }: diagnostics.append( _projection_warning( "unsupported_policy_condition", @@ -277,6 +288,72 @@ def _project_condition( continue projected_values.append(genre_id) return "genre_ids", projected_values, condition.operator == "contains_none" + if condition.field == "media.language": + if condition.operator not in {"in", "not_in"}: + diagnostics.append( + _projection_warning( + "unsupported_policy_condition", + f"规则 {rule_id} 的语言操作符 {condition.operator} 无法投影到旧配置", + ["rules", rule_id, "when"], + ) + ) + return None, [], False + aliases = policy.field_aliases.get(condition.field, {}) + return ( + "original_language", + [_original_standard_value(value, aliases) for value in values], + condition.operator == "not_in", + ) + if condition.field == "media.countries": + if condition.operator not in {"contains_any", "contains_none"}: + diagnostics.append( + _projection_warning( + "unsupported_policy_condition", + f"规则 {rule_id} 的国家操作符 {condition.operator} 无法投影到旧配置", + ["rules", rule_id, "when"], + ) + ) + return None, [], False + field_name = "production_countries" if media_type == "电影" else "origin_country" + aliases = policy.field_aliases.get(condition.field, {}) + return ( + field_name, + [_original_standard_value(value, aliases) for value in values], + condition.operator == "contains_none", + ) + if condition.field == "media.year": + if condition.operator not in {"in", "not_in"}: + diagnostics.append( + _projection_warning( + "unsupported_policy_condition", + f"规则 {rule_id} 的年份操作符 {condition.operator} 无法投影到旧配置", + ["rules", rule_id, "when"], + ) + ) + return None, [], False + return "release_year", values, condition.operator == "not_in" + if condition.field == "media.runtime": + if condition.operator not in {"in", "not_in"}: + diagnostics.append( + _projection_warning( + "unsupported_policy_condition", + f"规则 {rule_id} 的片长操作符 {condition.operator} 无法投影到旧配置", + ["rules", rule_id, "when"], + ) + ) + return None, [], False + return "runtime", values, condition.operator == "not_in" + if condition.field == "media.adult": + if condition.operator not in {"is_true", "is_false"}: + diagnostics.append( + _projection_warning( + "unsupported_policy_condition", + f"规则 {rule_id} 的成人标记操作符 {condition.operator} 无法投影到旧配置", + ["rules", rule_id, "when"], + ) + ) + return None, [], False + return "adult", ["TRUE" if condition.operator == "is_true" else "FALSE"], False if condition.field.startswith(_EXTENSION_PREFIX): field_name = condition.field.removeprefix(_EXTENSION_PREFIX) aliases = policy.field_aliases.get(condition.field, {}) @@ -300,6 +377,14 @@ def _original_alias_value(value: str, aliases: Mapping[str, str]) -> str: return value +def _original_standard_value(value: str, aliases: Mapping[str, str]) -> str: + """恢复标准字段迁移前的低位写法，不把生成的标题大小写别名写回旧配置。""" + for alias, canonical in aliases.items(): + if canonical == value and alias == alias.casefold() and alias != value: + return alias + return value + + def _projection_warning( code: str, message: str, diff --git a/app/db/adapters/classification.py b/app/db/adapters/classification.py index 06838a4ef..6b382728c 100644 --- a/app/db/adapters/classification.py +++ b/app/db/adapters/classification.py @@ -34,7 +34,12 @@ DirectoryConfigurationNormalizer = Callable[ def discard_removed_source_fallbacks(value: Any) -> Any: - """读取旧持久化策略时把 TMDB 兜底平移为媒体类型全局兜底。""" + """读取旧持久化策略时兼容已移除的来源级兜底字段。 + + 来源级兜底不能再直接写入新版策略模型，因此将其转换为策略末尾的 + 来源限定 catch-all 分类规则。分类求值器会在已有分类命中后跳过后续 + 分类规则，这与旧版“来源兜底仅在规则未命中时生效”的语义一致。 + """ if not isinstance(value, Mapping): return value @@ -52,16 +57,91 @@ def discard_removed_source_fallbacks(value: Any) -> Any: 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 + if isinstance(fallbacks, dict) and isinstance(tmdb_fallbacks, Mapping): + 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 + _promote_source_fallback_rules( + policy, + { + source: values + for source, values in source_fallbacks.items() + if str(source).strip().casefold() != "themoviedb" + }, + ) return state +_SOURCE_FALLBACK_MEDIA_TYPES = frozenset({"电影", "电视剧", "音乐"}) +_SOURCE_FALLBACK_MEDIA_KEYS = { + "电影": "movie", + "电视剧": "tv", + "音乐": "music", +} + + +def _promote_source_fallback_rules( + policy: dict[str, Any], + source_fallbacks: Mapping[Any, Any], +) -> None: + """把旧来源兜底转换为末尾的来源限定分类规则。""" + raw_rules = policy.get("rules") + rules = raw_rules if isinstance(raw_rules, list) else [] + if rules is not raw_rules: + policy["rules"] = rules + + used_ids = { + str(rule.get("id")) + for rule in rules + if isinstance(rule, Mapping) and rule.get("id") + } + priorities = [ + _safe_priority(rule.get("priority", index), index) + for index, rule in enumerate(rules) + if isinstance(rule, Mapping) + ] + next_priority = max(priorities, default=-1) + 1 + + for raw_source, raw_fallbacks in source_fallbacks.items(): + source = str(raw_source or "").strip().casefold() + if not source or not isinstance(raw_fallbacks, Mapping): + continue + for raw_media_type, raw_category_id in raw_fallbacks.items(): + media_type = str(raw_media_type or "").strip() + category_id = str(raw_category_id or "").strip() + if media_type not in _SOURCE_FALLBACK_MEDIA_TYPES or not category_id: + continue + rule_id = f"compat.source-fallback.{source}.{_SOURCE_FALLBACK_MEDIA_KEYS[media_type]}" + if rule_id in used_ids: + continue + rules.append( + { + "id": rule_id, + "name": f"兼容来源兜底 · {source} · {media_type}", + "kind": "category", + "enabled": True, + "priority": next_priority, + "media_types": [media_type], + "sources": [source], + "when": { + "field": "identity.media_source", + "operator": "equals", + "value": source, + }, + "target": {"category_id": category_id}, + } + ) + used_ids.add(rule_id) + next_priority += 1 + + +def _safe_priority(value: Any, fallback: int) -> int: + """读取损坏旧策略中的优先级，异常值退回稳定列表位置。""" + return value if isinstance(value, int) and not isinstance(value, bool) else fallback + + _COMMON_FALLBACK_IDS = { "电影": "movie.uncategorized", "电视剧": "tv.uncategorized", diff --git a/app/domain/classification/validation.py b/app/domain/classification/validation.py index a79190711..65ec61cd9 100644 --- a/app/domain/classification/validation.py +++ b/app/domain/classification/validation.py @@ -449,6 +449,7 @@ class ClassificationPolicyValidator: cls._validate_condition_value(condition, definition, path, collector) cls._validate_extension_scope(condition.field, rule, definition, path, collector) cls._warn_partial_support(condition.field, rule, definition, path, collector) + cls._validate_unavailable_support(condition.field, rule, definition, path, collector) @classmethod def _validate_condition_value( @@ -589,6 +590,31 @@ class ClassificationPolicyValidator: [*path, "field"], ) + @classmethod + def _validate_unavailable_support( + cls, + field_id: str, + rule: ClassificationRule, + definition: ClassificationFieldDefinition, + path: list[str | int], + collector: _ValidationCollector, + ) -> None: + """阻止规则限定到全部无法提供该字段的数据源。""" + scoped_sources = rule.sources or list(definition.source_support) + if not scoped_sources: + return + unavailable = [ + source + for source in scoped_sources + if definition.source_support.get(source, "unavailable") == "unavailable" + ] + if len(unavailable) == len(scoped_sources): + collector.error( + "unavailable_field_support", + f"字段 {field_id} 在所选来源中均不可用：{', '.join(unavailable)}", + [*path, "field"], + ) + @classmethod def _validate_fallbacks( cls, diff --git a/app/domain/projection/tmdb.py b/app/domain/projection/tmdb.py index 274204f29..f9b3b2632 100644 --- a/app/domain/projection/tmdb.py +++ b/app/domain/projection/tmdb.py @@ -118,10 +118,11 @@ def project( builder = ProjectionBuilder(current) media_type = _media_type(info) media_id = str(info.get("id")) if info.get("id") is not None else None - builder.set("media_source", MediaSource.TMDB) + # 辅助 TMDB 详情可能合并到豆瓣/Bangumi 主对象，不能覆盖其稳定身份。 + builder.set_missing("media_source", MediaSource.TMDB) builder.set("tmdb_info", info) - builder.set("type", media_type) - builder.set("media_id", media_id) + builder.set_missing("type", media_type) + builder.set_missing("media_id", media_id) builder.set("tmdb_id", info.get("id")) if not media_id: return builder.build() diff --git a/tests/test_media_classification.py b/tests/test_media_classification.py index 66fb3d9be..175dccc67 100644 --- a/tests/test_media_classification.py +++ b/tests/test_media_classification.py @@ -414,6 +414,23 @@ def test_rule_source_and_media_type_restrictions_are_applied_before_conditions( assert evaluation.result.recommended.category_id == expected_category +def test_unavailable_source_field_is_rejected() -> None: + """限定到完全不支持字段的来源时，策略不得发布。""" + result = _validation_report( + _policy( + _category_rule( + "rule.unavailable", + _leaf("media.runtime", "gte", 90), + media_types=["电影"], + sources=["tvdb"], + ) + ).model_dump(mode="json") + ) + + assert result.valid is False + assert any(issue.code == "unavailable_field_support" for issue in result.issues) + + def test_trace_reports_actual_values_and_match_decisions() -> None: evaluation = _evaluate( _policy( diff --git a/tests/test_media_classification_api.py b/tests/test_media_classification_api.py index b7663973e..ba1106004 100644 --- a/tests/test_media_classification_api.py +++ b/tests/test_media_classification_api.py @@ -22,6 +22,7 @@ from app.api.routers import API_V1_ROUTER_SPECS from app.application.classification.analysis import ( ClassificationAnalysisService, RecentHistoryClassificationSampleProvider, + build_classification_facts_from_media_payload, ) from app.application.classification.configuration import ( ClassificationPolicyConfigurationService, @@ -240,6 +241,32 @@ def test_field_catalog_exposes_retired_fields_outside_new_rule_options() -> None assert catalog.retired_fields[0].replacement_field == "media.genre_keys" +def test_preview_media_payload_preserves_dotted_plugin_source_facts() -> None: + """扩展来源 ID 含点号时，预览仍按完整命名空间读取事实。""" + field = ClassificationFieldDefinition( + id="extensions.plugin.example.region", + label="区域", + value_type="string", + operators=["equals"], + media_types=["电影"], + source_support={"plugin.example": "extension"}, + ) + + facts = build_classification_facts_from_media_payload( + { + "media_source": "plugin.example", + "media_id": "native-1", + "type": "电影", + "classification_facts": { + "extensions.plugin.example.region": "east-asia", + }, + }, + extra_fields=[field], + ) + + assert facts.extensions == {"plugin.example": {"region": "east-asia"}} + + def test_preview_returns_condition_path_and_structured_missing_fact_warning() -> None: """预览 trace 能定位规则条件，缺失字段提示保留代码、字段和来源。""" service = _service() @@ -273,8 +300,9 @@ def test_preview_returns_condition_path_and_structured_missing_fact_warning() -> trace for trace in evaluation.trace if trace.rule_id == "rule.language" ) assert language_trace.conditions[0].path == ["rules", rule_index, "when"] - assert evaluation.warnings[0].code == "missing_fact" - assert evaluation.warnings[0].field == "media.language" + assert evaluation.warnings + assert all(warning.code == "missing_fact" for warning in evaluation.warnings) + assert any(warning.field == "media.language" for warning in evaluation.warnings) assert evaluation.warnings[0].source == "themoviedb" @@ -412,6 +440,28 @@ async def test_recent_history_samples_are_bounded_deduplicated_and_honest() -> N assert "仅稳定保存" in batch.warnings[0] +@pytest.mark.asyncio # type: ignore[misc] +async def test_explicit_impact_samples_are_deduplicated_by_primary_identity() -> None: + """显式影响样本重复提交时不得放大分类变化统计。""" + service = _service() + analysis = ClassificationAnalysisService(service) + sample = _facts(media_source="douban", media_id="same") + + result = await analysis.impact( + _candidate_policy(), + expected_revision=1, + sample_limit=100, + example_limit=20, + samples=[sample, sample.model_copy(deep=True)], + ) + + assert result.sample_count == 1 + assert result.scanned_count == 2 + assert result.skipped_count == 1 + assert result.truncated is True + assert any("重复" in warning for warning in result.warnings) + + @pytest.mark.asyncio # type: ignore[misc] async def test_impact_analysis_resolves_complete_history_details_and_reports_gaps() -> None: """影响分析应以历史身份重新读取完整字段，并单独统计无法读取的记录。""" @@ -692,6 +742,10 @@ def test_fastapi_envelope_and_conflict_payload_are_frontend_compatible() -> None client = TestClient(app) policy_response = client.get("/api/v1/media/classification/policy") + default_policy_response = client.get( + "/api/v1/media/classification/policy", + params={"template": "default"}, + ) conflict_response = client.put( "/api/v1/media/classification/policy", json={ @@ -703,6 +757,8 @@ def test_fastapi_envelope_and_conflict_payload_are_frontend_compatible() -> None assert policy_response.status_code == 200 assert policy_response.json()["data"]["revision"] == 1 assert "success" not in policy_response.json()["data"] + assert default_policy_response.status_code == 200 + assert default_policy_response.json()["data"] == build_default_classification_policy().model_dump(mode="json") assert conflict_response.status_code == 409 assert conflict_response.json()["data"] == { "code": "classification_revision_conflict", diff --git a/tests/test_media_classification_configuration.py b/tests/test_media_classification_configuration.py index e42768914..47f7209e7 100644 --- a/tests/test_media_classification_configuration.py +++ b/tests/test_media_classification_configuration.py @@ -394,7 +394,7 @@ def test_rollback_rejects_revision_outside_bounded_history() -> None: def test_returned_active_state_and_history_cannot_mutate_internal_snapshot() -> None: """所有读取结果都必须与服务内部活动引用隔离。""" service = _service(_MemoryPolicyStore()) - service.initialize() + initialized = service.initialize() service.publish(_renamed_draft("第二版"), expected_revision=1) active = service.active() @@ -405,7 +405,7 @@ def test_returned_active_state_and_history_cannot_mutate_internal_snapshot() -> history[0].categories[0].name = "篡改历史" assert service.active().categories[0].name == "第二版" - assert service.history()[0].categories[0].name == "未分类" + assert service.history()[0].categories[0].name == initialized.categories[0].name @pytest.mark.asyncio # type: ignore[misc] @@ -428,4 +428,4 @@ async def test_async_entrypoints_delegate_to_same_version_semantics() -> None: 2, 3, ) - assert rolled_back.categories[0].name == "未分类" + assert rolled_back.categories[0].name == initialized.categories[0].name diff --git a/tests/test_media_classification_configuration_adapter.py b/tests/test_media_classification_configuration_adapter.py index 45bc36af3..d01b424e7 100644 --- a/tests/test_media_classification_configuration_adapter.py +++ b/tests/test_media_classification_configuration_adapter.py @@ -107,6 +107,39 @@ def test_old_source_fallback_is_promoted_to_global_fallback() -> None: assert "source_fallbacks" not in normalized["history"][0] +def test_old_non_tmdb_source_fallback_becomes_compatibility_rule() -> None: + """旧非 TMDB 来源兜底应转换为末尾来源限定规则，避免升级丢失行为。""" + state = { + "active": { + "fallbacks": { + "电影": "movie.uncategorized", + "电视剧": "tv.uncategorized", + "音乐": "music.uncategorized", + }, + "rules": [], + "source_fallbacks": { + "douban": {"电影": "legacy.douban.movie"}, + }, + }, + "history": [], + } + + normalized = discard_removed_source_fallbacks(state) + + rules = normalized["active"]["rules"] + assert len(rules) == 1 + assert rules[0]["id"] == "compat.source-fallback.douban.movie" + assert rules[0]["sources"] == ["douban"] + assert rules[0]["media_types"] == ["电影"] + assert rules[0]["when"] == { + "field": "identity.media_source", + "operator": "equals", + "value": "douban", + } + assert rules[0]["target"] == {"category_id": "legacy.douban.movie"} + assert "source_fallbacks" not in normalized["active"] + + 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 188e4a264..1bbc517e2 100644 --- a/tests/test_media_classification_execution.py +++ b/tests/test_media_classification_execution.py @@ -7,7 +7,11 @@ from copy import deepcopy from typing import Optional from unittest.mock import AsyncMock, Mock, patch -from app.application.classification.execution import ClassificationExecutionService +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 @@ -214,6 +218,51 @@ def test_legacy_tmdb_rules_use_non_tmdb_standard_facts() -> None: ) +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( diff --git a/tests/test_media_classification_legacy.py b/tests/test_media_classification_legacy.py index 7080bc294..e31d3cdd6 100644 --- a/tests/test_media_classification_legacy.py +++ b/tests/test_media_classification_legacy.py @@ -1,10 +1,13 @@ """旧版 TMDB 分类配置到新版策略的纯迁移与兼容投影测试。""" from collections.abc import Mapping +from pathlib import Path from typing import Any, Optional, cast import pytest +import yaml +from app.application.classification.configuration import build_builtin_classification_policy from app.application.classification.legacy import ( LegacyClassificationMigrationResult, build_legacy_tmdb_extension_facts, @@ -23,6 +26,7 @@ from app.schemas.category import ( ClassificationConditionNode, ClassificationFacts, ClassificationMediaType, + ClassificationPolicy, ) from app.schemas.types import MediaType @@ -153,6 +157,14 @@ def _category_name( return categories.get(category_id or "", "") +def _policy_category_name(policy: ClassificationPolicy, facts: ClassificationFacts) -> str: + """执行指定策略并返回推荐分类名称，用于比较不同策略的业务结果。""" + evaluation = ClassificationEvaluator.evaluate(policy, facts) + category_id = evaluation.result.recommended.category_id + categories = {str(category.id): str(category.name) for category in policy.categories} + return categories.get(category_id or "", "") + + def _leaf_fields(node: ClassificationConditionNode) -> list[str]: """按条件树顺序提取叶子字段。""" if isinstance(node, ClassificationCondition): @@ -169,8 +181,8 @@ def _leaf_fields(node: ClassificationConditionNode) -> list[str]: return [field for child in children for field in _leaf_fields(child)] -def test_default_style_config_preserves_order_and_uses_safe_standard_fields() -> None: - """默认式配置应保持顺序，已知风格使用标准字段，其余字段保持等价。""" +def test_default_style_config_preserves_order_and_migrates_equivalent_fields() -> None: + """默认式配置应保持顺序，具有等价语义的字段直接迁移到标准字段。""" result = migrate_legacy_category_config(_legacy_config()) assert result.valid @@ -190,22 +202,67 @@ def test_default_style_config_preserves_order_and_uses_safe_standard_fields() -> all_fields = [field for rule in result.policy.rules for field in _leaf_fields(rule.when)] assert "media.genre_keys" in all_fields assert "extensions.themoviedb.genre_ids" not in all_fields - assert "media.language" not in all_fields - assert "media.countries" not in all_fields - assert "extensions.themoviedb.original_language" in all_fields - assert "extensions.themoviedb.production_countries" in all_fields - assert "extensions.themoviedb.origin_country" in all_fields + assert "media.language" in all_fields + assert "media.countries" in all_fields + assert "extensions.themoviedb.original_language" not in all_fields + assert "extensions.themoviedb.production_countries" not in all_fields + assert "extensions.themoviedb.origin_country" not in all_fields assert legacy_extension_fields_from_policy(result.policy) == result.extra_fields assert ClassificationPolicyValidator.validate(result.policy, result.extra_fields).valid - origin_country = next( - field - for field in result.extra_fields - if field.id == "extensions.themoviedb.origin_country" + assert result.extra_fields == () + + +def test_official_category_yaml_has_no_unknown_genre_ids() -> None: + """仓库官方旧模板中的 Genre ID 均应直接迁移为标准风格键。""" + config = yaml.safe_load( + (Path(__file__).resolve().parents[1] / "config" / "category.yaml").read_text(), ) - assert origin_country.label == "原产国家/地区（旧规则）" - assert origin_country.group == "旧规则" - assert origin_country.selectable is False - assert origin_country.replacement_field == "media.countries" + + result = migrate_legacy_category_config(config) + + assert result.valid + assert result.extra_fields == () + assert all( + field_id == "media.genre_keys" + for rule in result.policy.rules + for field_id in _leaf_fields(rule.when) + if "genre" in field_id + ) + + +@pytest.mark.parametrize( + ("media_type", "tmdb_info", "expected"), + [ + ("电影", {"genre_ids": [16]}, "动画电影"), + ("电影", {"original_language": "zh"}, "华语电影"), + ("电影", {"original_language": "en"}, "外语电影"), + ("电视剧", {"genre_ids": [16], "origin_country": ["CN"]}, "国漫"), + ("电视剧", {"genre_ids": [16], "origin_country": ["JP"]}, "日番"), + ("电视剧", {"genre_ids": [99]}, "纪录片"), + ("电视剧", {"genre_ids": [10762]}, "儿童"), + ("电视剧", {"genre_ids": [10764]}, "综艺"), + ("电视剧", {"origin_country": ["CN"]}, "国产剧"), + ("电视剧", {"origin_country": ["US"]}, "欧美剧"), + ("电视剧", {"origin_country": ["JP"]}, "日韩剧"), + ("电视剧", {"origin_country": ["AU"]}, "未分类"), + ], +) +def test_builtin_policy_matches_official_legacy_category_semantics( + media_type: ClassificationMediaType, + tmdb_info: Mapping[str, object], + expected: str, +) -> None: + """内置标准规则应逐个复现官方 category.yaml 的分类结果。""" + config = yaml.safe_load( + (Path(__file__).resolve().parents[1] / "config" / "category.yaml").read_text(), + ) + migrated = migrate_legacy_category_config(config) + facts = _tmdb_facts(migrated, tmdb_info, media_type) + builtin = build_builtin_classification_policy() + + assert _legacy_tmdb_category(config["movie" if media_type == "电影" else "tv"], tmdb_info) == expected + assert _category_name(migrated, facts) == expected + assert _policy_category_name(builtin, facts) == expected def test_deleted_legacy_rules_keep_alias_only_fields_registered() -> None: @@ -220,9 +277,7 @@ def test_deleted_legacy_rules_keep_alias_only_fields_registered() -> None: fields = legacy_extension_fields_from_policy(policy) - assert [field.id for field in fields] == [ - "extensions.themoviedb.original_language" - ] + assert fields == () assert ClassificationPolicyValidator.validate(policy, fields).valid @@ -261,7 +316,7 @@ def test_first_empty_rule_becomes_global_fallback_and_later_entries_are_disabled def test_mixed_genre_ids_keep_positive_or_and_negative_and_semantics() -> None: - """已知与未知 Genre ID 混合时，正值保持 OR，排除值必须全部满足。""" + """用户自定义未知 Genre ID 与已知值混用时仍保持旧的 OR 与排除语义。""" result = migrate_legacy_category_config( { "movie": { @@ -291,6 +346,27 @@ def test_mixed_genre_ids_keep_positive_or_and_negative_and_semantics() -> None: assert _category_name(result, _tmdb_facts(result, {}, "电影")) == "兜底" +def test_known_negative_genre_ids_use_standard_field() -> None: + """官方词表内的排除 Genre ID 也应直接迁移到标准风格字段。""" + result = migrate_legacy_category_config( + { + "movie": { + "非动画纪录": {"genre_ids": "!16,!99"}, + "兜底": None, + }, + "tv": {}, + } + ) + + fields = _leaf_fields(result.policy.rules[0].when) + assert result.valid + assert fields == ["media.genre_keys"] + assert "extensions.themoviedb.genre_ids" not in fields + assert result.extra_fields == () + assert _category_name(result, _tmdb_facts(result, {"genre_ids": [16]}, "电影")) == "兜底" + assert _category_name(result, _tmdb_facts(result, {"genre_ids": [18]}, "电影")) == "非动画纪录" + + @pytest.mark.parametrize("negative", [False, True]) @pytest.mark.parametrize("field_name", ["origin_country", "genre_ids"]) def test_large_value_sets_remain_compact_and_preserve_legacy_semantics( @@ -382,10 +458,6 @@ def test_extension_fact_projection_reproduces_legacy_string_views() -> None: assert by_policy == by_fields assert by_policy == { "themoviedb": { - "runtime": ["120"], - "origin_country": ["CN", "HK"], - "production_countries": ["US", "NONE"], - "release_year": ["2024"], "keywords": ["ONE", "{'ID': 1}"], "genre_ids": ["999"], } @@ -564,6 +636,5 @@ def test_legacy_country_dictionary_keeps_country_codes_and_genre_ids_distinct() """旧地区条件能直接选择代码，未知旧风格编号不能误填为标准风格键。""" result = migrate_legacy_category_config({"tv": {"日韩剧": {"origin_country": "JP,KR", "genre_ids": "999"}}}) fields = {item.id: item for item in result.extra_fields} - countries = fields["extensions.themoviedb.origin_country"] - assert any(item.value == "JP" and item.label == "日本" for item in countries.options) + assert all(field_id != "extensions.themoviedb.origin_country" for field_id in fields) assert not fields["extensions.themoviedb.genre_ids"].options diff --git a/tests/test_media_classification_startup.py b/tests/test_media_classification_startup.py index 9a93d14c1..83cb4d819 100644 --- a/tests/test_media_classification_startup.py +++ b/tests/test_media_classification_startup.py @@ -20,6 +20,7 @@ from app.application.classification.contract import ClassificationPolicyConflict from app.application.classification.legacy import migrate_legacy_category_config from app.application.classification.runtime import ClassificationRuntime from app.schemas.category import ( + ClassificationCategory, ClassificationConditionGroup, ClassificationPolicyState, ) @@ -99,16 +100,31 @@ def _published_legacy_default_state() -> ClassificationPolicyState: """构造旧版本仅含三个未分类兜底的 revision 1 状态。""" policy = build_default_classification_policy() policy.categories = [ - category - for category in policy.categories - if category.id - in { - "movie.uncategorized", - "tv.uncategorized", - "music.uncategorized", - } + ClassificationCategory( + id="movie.uncategorized", + media_type="电影", + name="未分类", + path=["未分类"], + ), + ClassificationCategory( + id="tv.uncategorized", + media_type="电视剧", + name="未分类", + path=["未分类"], + ), + ClassificationCategory( + id="music.uncategorized", + media_type="音乐", + name="未分类", + path=["未分类"], + ), ] policy.rules = [] + policy.fallbacks = { + "电影": "movie.uncategorized", + "电视剧": "tv.uncategorized", + "音乐": "music.uncategorized", + } return ClassificationPolicyState(active=policy.model_copy(update={"revision": 1})) @@ -546,16 +562,17 @@ async def test_legacy_category_get_endpoints_use_classification_runtime_only( assert before.success is True assert before.data == {"movie": {}, "tv": {}} + default_policy = build_default_classification_policy() + expected_categories = { + media_type: [ + category.name + for category in default_policy.categories + if category.media_type == media_type and category.enabled + ] + for media_type in ("电影", "电视剧", "音乐") + } assert categories.root == { - "电影": ["未分类"], - "电视剧": ["未分类"], - "音乐": [ - "未分类", - "Album", - "Album / Compilation", - "EP", - "Single", - ], + **expected_categories, }