fix(classification): align migration defaults and source semantics

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)
 

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.<source>.<field> 命名空间整理媒体携带的扩展字段。"""
     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}:

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:

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,

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)
 
 

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,

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",

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,

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()

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(

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",

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

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)

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(

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

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,
     }