fix(classification): prepare policy-specific facts and refine defaults

2026-11-01 04:47:03 +08:00 · 2026-10-02 14:14:09 +08:00
parent da932158bb
commit 5f596484b4
10 changed files with 270 additions and 43 deletions
--- a/app/api/endpoints/classification.py
+++ b/app/api/endpoints/classification.py
@@ -96,8 +96,10 @@ def _get_analysis_service(
 async def _resolve_history_facts(
    execution: ClassificationExecutionPort,
    history: object,
-) -> ClassificationFacts | None:
-    """按历史记录中的来源和编号重新读取完整媒体信息。"""
+    active: ClassificationPolicy,
+    candidate: ClassificationPolicy,
+) -> tuple[ClassificationFacts, ClassificationFacts] | None:
+    """只读取一次历史媒体详情，再按两套策略分别补充缺失信息。"""
    media_source = _enum_text(getattr(history, "media_source", None))
    media_id = str(getattr(history, "media_id", None) or "").strip()
    media_type = _history_media_type(getattr(history, "type", None))
@@ -113,7 +115,9 @@ async def _resolve_history_facts(
        )
        if media is None:
            return None
-        return await execution.async_build_facts(media)
+        previous = await execution.async_build_facts(media, policy=active)
+        proposed = await execution.async_build_facts(media, policy=candidate)
+        return (previous, proposed) if previous is not None and proposed is not None else None
    except (TypeError, ValueError):
        return None

@@ -292,10 +296,10 @@ async def preview_policy(
    if request.policy is None:
        _require_active_policy(runtime)
    try:
-        return ClassificationAnalysisService(
+        return await ClassificationAnalysisService(
            runtime.service,
            execution=host_runtime.classification_execution,
-        ).preview(request)
+        ).async_preview(request)
    except ClassificationPolicyValidationError as error:
        return _validation_response(error)

--- a/app/application/classification/analysis.py
+++ b/app/application/classification/analysis.py
@@ -67,10 +67,10 @@ _UNCLASSIFIED_CATEGORY_ID = "__unclassified__"
 _DEFAULT_RESOLVE_CONCURRENCY = 3

 ClassificationImpactFactsResolver: TypeAlias = Callable[
-    [DownloadHistorySnapshot | TransferHistorySnapshot],
-    Awaitable[ClassificationFacts | None],
+    [DownloadHistorySnapshot | TransferHistorySnapshot, ClassificationPolicy, ClassificationPolicy],
+    Awaitable[tuple[ClassificationFacts, ClassificationFacts] | None],
 ]
-"""按历史记录重新读取完整媒体信息的异步端口。"""
+"""一次读取历史媒体详情，再分别按活动和候选策略准备事实的异步端口。"""


@dataclass(frozen=True, slots=True)
@@ -84,13 +84,16 @@ class ClassificationImpactSampleBatch:
    unresolved_count: int = 0
    truncated: bool = False
    warnings: tuple[str, ...] = ()
+    candidate_facts: tuple[ClassificationFacts, ...] | None = None


 class ClassificationImpactSampleProvider(Protocol):
    """为影响分析异步提供近期标准分类事实样本。"""

-    async def load(self, limit: int) -> ClassificationImpactSampleBatch:
-        """读取最多 ``limit`` 个按媒体身份去重的近期事实。"""
+    async def load(
+        self, limit: int, *, active: ClassificationPolicy, candidate: ClassificationPolicy,
+    ) -> ClassificationImpactSampleBatch:
+        """读取最多 ``limit`` 个媒体，分别准备两套策略所需事实。"""
        ...


@@ -113,8 +116,10 @@ class RecentHistoryClassificationSampleProvider:
        self._facts_resolver = facts_resolver
        self._resolve_concurrency = resolve_concurrency

-    async def load(self, limit: int) -> ClassificationImpactSampleBatch:
-        """合并两类近期历史，按时间和 ID 排序后去重并投影事实。"""
+    async def load(
+        self, limit: int, *, active: ClassificationPolicy, candidate: ClassificationPolicy,
+    ) -> ClassificationImpactSampleBatch:
+        """合并去重近期历史，保持活动与候选策略样本按身份成对。"""
        downloads, transfers = await asyncio.gather(
            self._download_history.async_list_by_page(page=1, count=limit),
            self._transfer_history.async_list_by_page(
@@ -169,15 +174,17 @@ class RecentHistoryClassificationSampleProvider:

        records_to_resolve = unique_records[:limit]
        skipped_count += max(0, len(unique_records) - len(records_to_resolve))
-        resolved = await self._resolve_records(records_to_resolve)
+        resolved = await self._resolve_records(records_to_resolve, active, candidate)
        facts = []
+        candidate_facts = []
        unresolved_count = 0
        for item in resolved:
            if item is None:
                unresolved_count += 1
                skipped_count += 1
                continue
-            facts.append(item)
+            facts.append(item[0])
+            candidate_facts.append(item[1])

        warnings = [
            "系统会按近期下载和整理记录中的来源与编号重新读取完整媒体信息；无法读取的记录不会参与比较",
@@ -196,28 +203,35 @@ class RecentHistoryClassificationSampleProvider:
            unresolved_count=unresolved_count,
            truncated=len(unique_records) > limit,
            warnings=tuple(warnings),
+            candidate_facts=tuple(candidate_facts),
        )

    async def _resolve_records(
        self,
        records: Sequence[tuple[_HistorySampleRecord, ClassificationFacts]],
-    ) -> list[ClassificationFacts | None]:
+        active: ClassificationPolicy,
+        candidate: ClassificationPolicy,
+    ) -> list[tuple[ClassificationFacts, ClassificationFacts] | None]:
        """以固定并发上限重新读取详情，并拒绝身份不一致的返回值。"""
        facts_resolver = self._facts_resolver
        if facts_resolver is None:
-            return [projected for _, projected in records]
+            return [(projected, projected) for _, projected in records]
        semaphore = asyncio.Semaphore(self._resolve_concurrency)

        async def resolve(
            record: _HistorySampleRecord,
            projected: ClassificationFacts,
-        ) -> ClassificationFacts | None:
+        ) -> tuple[ClassificationFacts, ClassificationFacts] | None:
+            """在同一并发窗口内准备两份事实，任一身份不符则排除整个样本。"""
            async with semaphore:
                try:
-                    facts = await facts_resolver(record.payload)
+                    facts = await facts_resolver(record.payload, active, candidate)
                except Exception:  # noqa: BLE001  单条详情失败不应阻断整批分析
                    return None
-            if facts is None or _classification_identity_key(facts) != _classification_identity_key(projected):
+            if facts is None or any(
+                _classification_identity_key(item) != _classification_identity_key(projected)
+                for item in facts
+            ):
                return None
            return facts

@@ -278,7 +292,7 @@ class ClassificationAnalysisService:
        return self._configuration.validate(policy)

    def preview(self, request: ClassificationPreviewRequest) -> ClassificationEvaluation:
-        """对搜索结果或兼容事实执行活动策略或合法草稿，并返回完整命中轨迹。"""
+        """对完整媒体详情或兼容事实执行活动策略或合法草稿，并返回完整命中轨迹。"""
        policy = request.policy or self._configuration.active()
        if request.policy is not None:
            self._require_valid(policy)
@@ -289,6 +303,19 @@ class ClassificationAnalysisService:
        )
        return evaluate_classification_facts(policy, facts, trace=True)

+    async def async_preview(self, request: ClassificationPreviewRequest) -> ClassificationEvaluation:
+        """异步补充预览事实，避免网络补充占用 API 事件循环。"""
+        policy = request.policy or self._configuration.active()
+        if request.policy is not None:
+            self._require_valid(policy)
+        if request.input.kind == "media" and self._execution is not None:
+            media = _preview_media(request.input.media)
+            facts = await self._execution.async_build_facts(cast(Any, media), policy=policy)
+            if facts is not None:
+                return evaluate_classification_facts(policy, facts, trace=True)
+        facts = self._preview_facts(request.input, policy=policy, use_execution=False)
+        return evaluate_classification_facts(policy, facts, trace=True)
+
    def _preview_facts(
        self,
        input_data: ClassificationPreviewInput,
@@ -296,7 +323,7 @@ class ClassificationAnalysisService:
        policy: ClassificationPolicy,
        use_execution: bool,
    ) -> ClassificationFacts:
-        """构造预览事实；活动策略优先复用真实执行端口的补充链路。"""
+        """构造预览事实；媒体输入按所选策略复用真实执行端口的补充链路。"""
        if input_data.kind == "facts":
            return input_data.facts
        media = _preview_media(input_data.media)
@@ -333,7 +360,7 @@ class ClassificationAnalysisService:
            ),
        )
        self._require_valid(candidate)
-        batch = await self._sample_batch(samples, sample_limit)
+        batch = await self._sample_batch(samples, sample_limit, active, candidate)
        return _build_impact_analysis(
            active, candidate, batch, sample_limit, example_limit
        )
@@ -348,6 +375,8 @@ class ClassificationAnalysisService:
        self,
        samples: Sequence[ClassificationFacts],
        sample_limit: int,
+        active: ClassificationPolicy,
+        candidate: ClassificationPolicy,
    ) -> ClassificationImpactSampleBatch:
        """优先使用请求事实，否则委托近期历史提供器生成样本。"""
        if samples:
@@ -402,7 +431,7 @@ class ClassificationAnalysisService:
                truncated=False,
                warnings=("近期历史样本提供器未配置，本次影响分析没有可比较样本",),
            )
-        return await self._sample_provider.load(sample_limit)
+        return await self._sample_provider.load(sample_limit, active=active, candidate=candidate)


 def _history_facts(
@@ -573,9 +602,10 @@ def _build_impact_analysis(
    path_only_changed_count = 0
    rule_changed_only_count = 0
    became_fallback_count = 0
-    for facts in batch.facts:
+    candidate_facts = batch.candidate_facts if batch.candidate_facts is not None else batch.facts
+    for facts, proposed_facts in zip(batch.facts, candidate_facts, strict=True):
        previous = evaluate_classification_facts(active, facts).result
-        proposed = evaluate_classification_facts(candidate, facts).result
+        proposed = evaluate_classification_facts(candidate, proposed_facts).result
        previous_categories[_category_id(previous)] += 1
        candidate_categories[_category_id(proposed)] += 1
        if "partial" in {previous.state, proposed.state}:
--- a/app/application/classification/configuration.py
+++ b/app/application/classification/configuration.py
@@ -47,14 +47,16 @@ _DEFAULT_MEDIA_CATEGORIES = (
    ("movie.animation", "电影", "动画电影", ("动画电影",)),
    ("movie.chinese", "电影", "华语电影", ("华语电影",)),
    ("movie.foreign", "电影", "外语电影", ("外语电影",)),
+    ("movie.uncategorized", "电影", "未分类", ("未分类",)),
    ("tv.dongman.cn", "电视剧", "国漫", ("国漫",)),
    ("tv.dongman.jp", "电视剧", "日番", ("日番",)),
+    ("tv.animation.other", "电视剧", "其他动画", ("其他动画",)),
    ("tv.documentary", "电视剧", "纪录片", ("纪录片",)),
    ("tv.kids", "电视剧", "儿童", ("儿童",)),
    ("tv.variety", "电视剧", "综艺", ("综艺",)),
    ("tv.chinese", "电视剧", "国产剧", ("国产剧",)),
    ("tv.western", "电视剧", "欧美剧", ("欧美剧",)),
-    ("tv.asian", "电视剧", "日韩剧", ("日韩剧",)),
+    ("tv.asian", "电视剧", "亚洲剧", ("亚洲剧",)),
    ("tv.uncategorized", "电视剧", "未分类", ("未分类",)),
    ("music.uncategorized", "音乐", "未分类", ("未分类",)),
 )
@@ -103,6 +105,19 @@ def _build_builtin_movie_rules() -> list[ClassificationRule]:
            ),
            target=ClassificationTarget(category_id="movie.chinese"),
        ),
+        ClassificationRule(
+            id="movie.foreign.default",
+            name="外语电影",
+            kind="category",
+            priority=2,
+            media_types=["电影"],
+            when=ClassificationCondition(
+                field="media.language",
+                operator="not_in",
+                value=["zh", "cn", "bo", "za"],
+            ),
+            target=ClassificationTarget(category_id="movie.foreign"),
+        ),
    ]


@@ -153,6 +168,19 @@ def _build_builtin_tv_rules() -> list[ClassificationRule]:
            ),
            target=ClassificationTarget(category_id="tv.dongman.jp"),
        ),
+        ClassificationRule(
+            id="tv.animation.other.default",
+            name="其他动画",
+            kind="category",
+            priority=4,
+            media_types=["电视剧"],
+            when=ClassificationCondition(
+                field="media.genre_keys",
+                operator="contains_any",
+                value=["animation"],
+            ),
+            target=ClassificationTarget(category_id="tv.animation.other"),
+        ),
        ClassificationRule(
            id="tv.documentary.default",
            name="纪录片",
@@ -214,13 +242,17 @@ def _build_builtin_tv_rules() -> list[ClassificationRule]:
            when=ClassificationCondition(
                field="media.countries",
                operator="contains_any",
-                value=["US", "FR", "GB", "DE", "ES", "IT", "NL", "PT", "RU", "UK"],
+                value=[
+                    "US", "CA", "FR", "GB", "DE", "ES", "IT", "NL", "PT", "RU",
+                    "IE", "BE", "CH", "AT", "SE", "NO", "DK", "FI", "IS", "PL",
+                    "CZ", "HU", "GR", "RO", "UA",
+                ],
            ),
            target=ClassificationTarget(category_id="tv.western"),
        ),
        ClassificationRule(
            id="tv.asian.default",
-            name="日韩剧",
+            name="亚洲剧",
            kind="category",
            priority=9,
            media_types=["电视剧"],
@@ -235,13 +267,16 @@ def _build_builtin_tv_rules() -> list[ClassificationRule]:


 def build_builtin_classification_policy() -> ClassificationPolicy:
-    """构造与仓库内 legacy category.yaml 默认值等效的标准分类策略。"""
+    """构造显式区分外语与未知信息的默认模板，不改写已保存的旧策略。"""
+    rules = [*_build_builtin_movie_rules(), *_build_builtin_tv_rules()]
+    for priority, rule in enumerate(rules):
+        rule.priority = priority
    return with_default_music_classification(
        ClassificationPolicy(
            categories=_build_builtin_media_categories(),
-            rules=[*_build_builtin_movie_rules(), *_build_builtin_tv_rules()],
+            rules=rules,
            fallbacks={
-                "电影": "movie.foreign",
+                "电影": "movie.uncategorized",
                "电视剧": "tv.uncategorized",
                "音乐": "music.uncategorized",
            },
@@ -327,6 +362,7 @@ def with_default_music_classification(policy: ClassificationPolicy) -> Classific
        value: Union[str, list[str]],
        category_id: str,
    ) -> None:
+        """按既有规则之后的顺序追加单个音乐规则。"""
        nonlocal priority
        rules.append(
            ClassificationRule(
--- a/app/application/classification/execution.py
+++ b/app/application/classification/execution.py
@@ -57,8 +57,10 @@ class ClassificationExecutionPort(Protocol):
    async def async_build_facts(
        self,
        media: ClassificationSubject,
+        *,
+        policy: ClassificationPolicy | None = None,
    ) -> ClassificationFacts | None:
-        """异步构造与实际分类一致的完整事实快照，不写入媒体或策略。"""
+        """按指定或活动策略异步准备事实，供预览和新旧策略比较复用。"""
        ...

    def build_facts(
@@ -143,13 +145,16 @@ class ClassificationExecutionService:
    async def async_build_facts(
        self,
        media: ClassificationSubject,
+        *,
+        policy: ClassificationPolicy | None = None,
    ) -> ClassificationFacts | None:
-        """构造影响分析使用的完整事实，并复用插件扩展与跨来源补充规则。"""
+        """按指定或活动策略补充事实，隔离候选策略与原媒体对象。"""
        finalized, policy, facts, _ = self._prepare(
            media,
            extensions=None,
            effective_override=None,
            refresh=False,
+            policy_override=policy,
        )
        if policy is None or facts is None:
            return None
--- a/docs/mcp-api.md
+++ b/docs/mcp-api.md
@@ -475,6 +475,8 @@ SMB 配置支持两种明确的路径模式：旧字段 `share: "data"` 保持

 #### 媒体自动分类

+媒体预览应先通过媒体详情接口取得完整信息，避免将搜索摘要中缺失的字段误判为不匹配。媒体输入会按所选策略异步补充缺失信息；显式标准化事实按原值计算。近期历史影响分析只读取一次媒体详情，再分别按活动策略与候选策略补充信息和比较，无法读取详情的记录不视为未变化。
+
 媒体自动分类使用完整、可版本化的策略作为唯一写入合同。先读取当前策略的
 `revision`，再用字段目录中的稳定字段 ID 和操作符构造规则；发布和回滚均使用
 `expected_revision` 做并发校验，成功后产生新的 revision。旧 `/media/category` 与
@@ -485,7 +487,7 @@ SMB 配置支持两种明确的路径模式：旧字段 `share: "data"` 保持
 | GET | `/api/v1/media/classification/fields` | 登录用户读取标准字段、操作符、通用选项、来源候选与策略限制 |
 | GET | `/api/v1/media/classification/policy` | 登录用户读取当前活动策略和 revision |
 | POST | `/api/v1/media/classification/validate` | 超级管理员校验完整草稿，不保存 |
-| POST | `/api/v1/media/classification/preview` | 登录用户对媒体搜索结果或标准化事实执行单次只读预览，可选草稿策略 |
+| POST | `/api/v1/media/classification/preview` | 登录用户对完整媒体详情或标准化事实执行单次只读预览，可选草稿策略 |
 | POST | `/api/v1/media/classification/impact` | 超级管理员比较活动策略与草稿对近期历史或显式样本的有界影响 |
 | GET | `/api/v1/media/classification/history` | 超级管理员读取可回滚的有限历史版本 |
 | PUT | `/api/v1/media/classification/policy` | 超级管理员在 `expected_revision` 匹配时校验并发布完整策略，需要写操作确认 |
--- a/skills/moviepilot-api/api/media.md
+++ b/skills/moviepilot-api/api/media.md
@@ -17,7 +17,11 @@ other clients.
   rules, fallbacks, and aliases that should remain unchanged.
 3. Use `media.classification.policy.validate` for draft-only structural and
   semantic validation. Use `media.classification.policy.preview` with either a
-   selected media result or normalized facts to inspect one classification result.
+   complete media details or normalized facts to inspect one classification result.
+   Fetch media details before previewing; search summaries may omit classification fields.
+   Media inputs use the selected policy for asynchronous metadata enrichment.
+   Explicit facts are evaluated as supplied. Recent-history impact analysis prepares
+   facts separately under the active and candidate policies before comparing them.
   For bounded recent-history or explicit-fact comparisons, use
   `media.classification.policy.impact`.
 4. Publish with `media.classification.policy.update`, sending the complete policy
--- a/tests/test_media_classification.py
+++ b/tests/test_media_classification.py
@@ -32,6 +32,7 @@ def _category(
    name: str,
    path: list[str],
 ) -> dict[str, Any]:
+    """构造指定媒体类型的可用分类目录。"""
    return {
        "id": category_id,
        "media_type": media_type,
@@ -43,6 +44,7 @@ def _category(


 def _base_policy_payload() -> dict[str, Any]:
+    """创建涵盖三类媒体兜底的最小合法策略载荷。"""
    return {
        "schema_version": 2,
        "revision": 1,
@@ -75,6 +77,7 @@ def _category_rule(
    sources: list[str] | None = None,
    labels: list[str] | None = None,
 ) -> dict[str, Any]:
+    """构造带媒体范围与目标分类的条件规则。"""
    return {
        "id": rule_id,
        "name": rule_id,
@@ -96,6 +99,7 @@ def _label_rule(
    when: dict[str, Any],
    labels: list[str],
 ) -> dict[str, Any]:
+    """构造仅累积标签、不改变分类的规则。"""
    return {
        "id": rule_id,
        "name": rule_id,
@@ -110,12 +114,14 @@ def _label_rule(


 def _policy(*rules: dict[str, Any]) -> ClassificationPolicy:
+    """将规则载荷组装为经过结构校验的策略。"""
    payload = _base_policy_payload()
    payload["rules"] = list(rules)
    return ClassificationPolicy.model_validate(payload)


 def _set_nested(payload: dict[str, Any], field: str, value: Any) -> None:
+    """按点分字段路径写入测试事实。"""
    parent = payload
    parts = field.split(".")
    for part in parts[:-1]:
@@ -132,6 +138,7 @@ def _facts(
    media_source: str = "themoviedb",
    values: dict[str, Any] | None = None,
 ) -> ClassificationFacts:
+    """创建稳定身份和可选字段的标准分类事实。"""
    payload: dict[str, Any] = {
        "identity": {
            "media_source": media_source,
@@ -174,6 +181,7 @@ def _evaluate(
    *,
    trace: bool = False,
 ) -> ClassificationEvaluation:
+    """使用默认或指定策略对标准事实执行纯求值。"""
    return ClassificationEvaluator().evaluate(
        policy=policy,
        facts=facts or _facts(),
@@ -182,6 +190,7 @@ def _evaluate(


 def _leaf(field: str, operator: str, value: Any = _MISSING) -> dict[str, Any]:
+    """构造具有可选值的叶子条件载荷。"""
    condition = {"field": field, "operator": operator}
    if value is not _MISSING:
        condition["value"] = value
@@ -222,6 +231,29 @@ def test_default_music_policy_uses_structured_album_categories(
    assert result.result.effective.category_path == category_path


+@pytest.mark.parametrize(  # type: ignore[misc]
+    ("media_type", "values", "category_id"),
+    [
+        ("电影", {"media.genre_keys": ["drama"], "media.language": None}, "movie.uncategorized"),
+        ("电影", {"media.genre_keys": ["drama"], "media.language": "en"}, "movie.foreign"),
+        ("电影", {"media.genre_keys": ["animation"], "media.language": "zh"}, "movie.animation"),
+        ("电视剧", {"media.genre_keys": ["drama"], "media.countries": ["TH"]}, "tv.asian"),
+        ("电视剧", {"media.genre_keys": ["drama"], "media.countries": ["CA"]}, "tv.western"),
+        ("电视剧", {"media.genre_keys": ["animation"], "media.countries": ["US"]}, "tv.animation.other"),
+        ("电视剧", {"media.genre_keys": ["animation"], "media.countries": ["CN"]}, "tv.dongman.cn"),
+    ],
+)
+def test_default_policy_distinguishes_unknown_information_and_region_categories(
+    media_type: str, values: dict[str, Any], category_id: str,
+) -> None:
+    """默认模板区分缺失语言与外语，并为地区、其他动画提供准确分类。"""
+    policy = build_default_classification_policy()
+    result = _evaluate(policy, _facts(media_type=media_type, values=values))
+    assert result.result.recommended.category_id == category_id
+    assert next(category.name for category in policy.categories if category.id == "tv.asian") == "亚洲剧"
+    assert [rule.priority for rule in policy.rules] == list(range(len(policy.rules)))
+
+
@pytest.mark.parametrize(  # type: ignore[misc]
    ("condition", "fact_values"),
    [
@@ -271,6 +303,7 @@ def test_all_operators_have_a_representative_match(
    condition: dict[str, Any],
    fact_values: dict[str, Any],
 ) -> None:
+    """逐项验证支持的操作符能够匹配代表性值。"""
    evaluation = _evaluate(
        _policy(_category_rule("rule.operator", condition)),
        _facts(values=fact_values),
@@ -295,6 +328,7 @@ def test_negative_operators_do_not_match_missing_or_null_values(
    operator: str,
    expected: Any,
 ) -> None:
+    """字段缺失或为空时，否定操作符不能误命中。"""
    evaluation = _evaluate(
        _policy(_category_rule("rule.negative", _leaf(field, operator, expected))),
        _facts(values={field: missing_value}),
@@ -306,6 +340,7 @@ def test_negative_operators_do_not_match_missing_or_null_values(

@pytest.mark.parametrize("missing_value", [_MISSING, None], ids=["missing", "null"])  # type: ignore[misc]
 def test_not_exists_matches_missing_and_null_values(missing_value: Any) -> None:
+    """不存在操作符应匹配缺失及空字段。"""
    evaluation = _evaluate(
        _policy(
            _category_rule(
@@ -320,6 +355,7 @@ def test_not_exists_matches_missing_and_null_values(missing_value: Any) -> None:


 def test_all_any_and_not_groups_can_be_nested() -> None:
+    """嵌套的且、或、排除组合保留逻辑语义。"""
    condition = {
        "all": [
            _leaf("media.year", "gte", 2000),
@@ -340,6 +376,7 @@ def test_all_any_and_not_groups_can_be_nested() -> None:


 def test_first_matching_category_rule_wins() -> None:
+    """分类只使用按顺序最先命中的规则。"""
    always = _leaf("media.type", "equals", "电影")
    evaluation = _evaluate(
        _policy(
@@ -354,6 +391,7 @@ def test_first_matching_category_rule_wins() -> None:


 def test_matching_label_rules_accumulate_with_stable_deduplication() -> None:
+    """标签规则累积结果并稳定去重。"""
    always = _leaf("media.type", "equals", "电影")
    evaluation = _evaluate(
        _policy(
@@ -371,6 +409,7 @@ def test_matching_label_rules_accumulate_with_stable_deduplication() -> None:


 def test_media_type_fallback_is_used_when_no_category_rule_matches() -> None:
+    """无规则命中时使用对应媒体类型的兜底分类。"""
    evaluation = _evaluate(
        _policy(
            _category_rule(
@@ -399,6 +438,7 @@ def test_rule_source_and_media_type_restrictions_are_applied_before_conditions(
    media_source: str,
    expected_category: str,
 ) -> None:
+    """来源与媒体类型范围先于条件求值。"""
    rule = _category_rule(
        "rule.restricted",
        _leaf("media.year", "gte", 2000),
@@ -432,6 +472,7 @@ def test_unavailable_source_field_is_rejected() -> None:


 def test_trace_reports_actual_values_and_match_decisions() -> None:
+    """匹配轨迹保留实际字段值和判断结果。"""
    evaluation = _evaluate(
        _policy(
            _category_rule(
@@ -454,6 +495,7 @@ def _validation_report(
    *,
    fields: Sequence[ClassificationFieldDefinition] | None = None,
 ) -> ClassificationValidationResult:
+    """收集策略结构和语义校验结果。"""
    standard_fields = get_standard_classification_fields()
    extra_fields = tuple(fields or ())
    if extra_fields[:len(standard_fields)] == standard_fields:
@@ -470,6 +512,7 @@ def _assert_validation_error(
    *,
    fields: Sequence[ClassificationFieldDefinition] | None = None,
 ) -> None:
+    """断言无效策略产生指定错误代码。"""
    report = _validation_report(payload, fields=fields)
    error_codes = {
        issue.code for issue in report.issues if issue.severity == "error"
@@ -479,6 +522,7 @@ def _assert_validation_error(


 def test_valid_policy_has_no_validation_errors() -> None:
+    """合法策略不产生阻断发布的校验错误。"""
    payload = _base_policy_payload()
    payload["rules"] = [
        _category_rule(
@@ -516,6 +560,7 @@ def test_duplicate_ids_are_rejected(
    mutate: Callable[[dict[str, Any]], None],
    expected_code: str,
 ) -> None:
+    """重复分类或规则编号应被拒绝。"""
    payload = _base_policy_payload()
    payload["rules"] = [
        _category_rule("rule.duplicate", _leaf("media.year", "gte", 2000))
@@ -540,6 +585,7 @@ def test_duplicate_ids_are_rejected(
    ],
 )
 def test_invalid_category_paths_are_rejected(path: list[str]) -> None:
+    """非法目录路径应在发布前被拒绝。"""
    payload = _base_policy_payload()
    payload["categories"][0]["path"] = path

@@ -547,6 +593,7 @@ def test_invalid_category_paths_are_rejected(path: list[str]) -> None:


 def test_duplicate_paths_are_rejected_within_the_same_media_type() -> None:
+    """同一媒体类型不允许重复分类路径。"""
    payload = _base_policy_payload()
    payload["categories"][1]["path"] = payload["categories"][0]["path"]

@@ -565,6 +612,7 @@ def test_category_rule_targets_must_exist_and_match_the_rule_media_type(
    target: str,
    expected_code: str,
 ) -> None:
+    """规则目标必须存在且与媒体范围兼容。"""
    payload = _base_policy_payload()
    payload["rules"] = [
        _category_rule(
@@ -598,6 +646,7 @@ def test_condition_contract_errors_are_rejected(
    condition: dict[str, Any],
    expected_code: str,
 ) -> None:
+    """条件字段、操作符及值的契约错误应被拒绝。"""
    payload = _base_policy_payload()
    payload["rules"] = [_category_rule("rule.invalid-condition", condition)]

@@ -605,6 +654,7 @@ def test_condition_contract_errors_are_rejected(


 def test_condition_tree_depth_limit_is_enforced() -> None:
+    """条件树不能超过深度上限。"""
    condition: dict[str, Any] = _leaf("media.year", "gte", 2000)
    for _ in range(4):
        condition = {"not": condition}
@@ -615,6 +665,7 @@ def test_condition_tree_depth_limit_is_enforced() -> None:


 def test_policy_rule_count_limit_is_enforced() -> None:
+    """策略规则数量不能超过上限。"""
    payload = _base_policy_payload()
    payload["rules"] = [
        _category_rule(
@@ -628,6 +679,7 @@ def test_policy_rule_count_limit_is_enforced() -> None:


 def test_per_rule_leaf_condition_limit_is_enforced() -> None:
+    """单条规则的叶子条件数量不能超过上限。"""
    payload = _base_policy_payload()
    payload["rules"] = [
        _category_rule(
@@ -645,6 +697,7 @@ def test_per_rule_leaf_condition_limit_is_enforced() -> None:


 def test_every_enabled_media_type_requires_a_fallback() -> None:
+    """每种启用媒体类型都必须配置可用兜底。"""
    payload = _base_policy_payload()
    payload["fallbacks"].pop("音乐")

@@ -652,6 +705,7 @@ def test_every_enabled_media_type_requires_a_fallback() -> None:


 def test_extension_field_namespace_must_match_the_restricted_source() -> None:
+    """扩展字段的命名空间必须与来源限制一致。"""
    standard_fields = list(get_standard_classification_fields())
    field_model = type(standard_fields[0])
    extension_field = field_model.model_validate(
@@ -686,6 +740,7 @@ def test_extension_field_namespace_must_match_the_restricted_source() -> None:


 def test_standard_field_catalog_is_unique_and_exposes_operator_contracts() -> None:
+    """标准字段目录保持唯一并声明操作符合同。"""
    fields = get_standard_classification_fields()
    field_map = {field.id: field for field in fields}

@@ -699,11 +754,13 @@ def test_standard_field_catalog_is_unique_and_exposes_operator_contracts() -> No


 def _percentile_95(samples: list[float]) -> float:
+    """从样本耗时中计算第九十五百分位。"""
    ordered = sorted(samples)
    return ordered[max(0, int(len(ordered) * 0.95) - 1)]


 def test_two_hundred_rules_with_six_conditions_evaluate_under_five_ms_p95() -> None:
+    """验证二百条六条件规则的纯求值性能预算。"""
    rules = []
    for index in range(200):
        rules.append(
--- a/tests/test_media_classification_api.py
+++ b/tests/test_media_classification_api.py
@@ -5,7 +5,9 @@ from __future__ import annotations
 import json
 from collections.abc import Callable
 from datetime import datetime, timezone
+from functools import partial
 from typing import TypeVar, cast
+from unittest.mock import AsyncMock

 import pytest
 from fastapi import FastAPI
@@ -29,8 +31,10 @@ from app.application.classification.configuration import (
    build_default_classification_policy,
 )
 from app.application.classification.contract import ClassificationPolicyConflictError
+from app.application.classification.execution import ClassificationExecutionService
 from app.application.classification.runtime import ClassificationRuntime
 from app.application.history import DownloadHistorySnapshot, TransferHistorySnapshot
+from app.domain.context import MediaInfo
 from app.schemas.category import (
    ClassificationCategory,
    ClassificationCondition,
@@ -45,7 +49,7 @@ from app.schemas.category import (
    ClassificationRule,
    ClassificationTarget,
 )
-from app.schemas.types import MediaSource
+from app.schemas.types import MediaSource, MediaType

 T = TypeVar("T")
 NOW = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc)
@@ -196,6 +200,83 @@ def _response_payload(response: JSONResponse) -> dict[str, object]:
    return cast(dict[str, object], json.loads(response.body))


+class _CountryEnrichment:
+    """模拟按当前规则需要补充国别的来源，禁止预览退回同步网络路径。"""
+
+    def enrich(self, *_args: object) -> ClassificationFacts:
+        """同步调用表示异步预览错误地阻塞了请求处理。"""
+        raise AssertionError("预览必须调用异步事实补充")
+
+    async def async_enrich(
+        self, policy: ClassificationPolicy, facts: ClassificationFacts, _media: object,
+    ) -> ClassificationFacts:
+        """仅在候选策略确实需要国别且启用补充时返回缺失字段。"""
+        result = facts.model_copy(deep=True)
+        if policy.enrichment_mode == "enrich_missing" and any(
+            isinstance(rule.when, ClassificationCondition) and rule.when.field == "media.countries"
+            and "电影" in rule.media_types for rule in policy.rules
+        ):
+            result.media.countries = ["CN"]
+        return result
+
+
+@pytest.mark.asyncio  # type: ignore[misc]
+@pytest.mark.parametrize("change", ["mode", "condition"])  # type: ignore[misc]
+async def test_impact_and_preview_prepare_candidate_facts_independently(
+    monkeypatch: pytest.MonkeyPatch, change: str,
+) -> None:
+    """补充模式或所需字段变化都必须纳入影响分析，并与异步预览保持一致。"""
+    policy = build_default_classification_policy()
+    policy.categories.append(ClassificationCategory(
+        id="movie.cn", name="中国出品", media_type="电影", path=["中国出品"],
+    ))
+    country_rule = ClassificationRule(
+        id="movie.cn.rule", name="中国出品电影", kind="category", media_types=["电影"],
+        when=ClassificationCondition(field="media.countries", operator="contains_any", value=["CN"]),
+        target=ClassificationTarget(category_id="movie.cn"),
+    )
+    if change == "mode":
+        policy.rules.insert(0, country_rule)
+    else:
+        policy.enrichment_mode = "enrich_missing"
+    service = ClassificationPolicyConfigurationService(_MemoryPolicyStore())
+    active = service.initialize(policy)
+    candidate = active.model_copy(deep=True)
+    candidate.enrichment_mode = "enrich_missing"
+    if change == "condition":
+        candidate.rules.insert(0, country_rule)
+    execution = ClassificationExecutionService(ClassificationRuntime(service), enrichment=_CountryEnrichment())
+    payload = {
+        "media_source": "themoviedb", "media_id": "535167", "type": "电影",
+        "title": "测试电影", "original_language": "en", "genre_ids": [18],
+    }
+    source = MediaInfo(**payload)
+    recognize = AsyncMock(return_value=source)
+    monkeypatch.setattr(classification_endpoint.MediaChain, "async_recognize_media", recognize)
+    provider = RecentHistoryClassificationSampleProvider(
+        download_history=cast(object, _DownloadHistory([
+            DownloadHistorySnapshot(id=1, title="测试电影", path="/downloads/test", type="电影",
+                                    media_source=MediaSource.TMDB, media_id="535167"),
+        ])),
+        transfer_history=cast(object, _TransferHistory([])),
+        facts_resolver=partial(classification_endpoint._resolve_history_facts, execution),
+    )
+    analysis = ClassificationAnalysisService(service, sample_provider=provider, execution=execution)
+    impact = await analysis.impact(candidate, expected_revision=active.revision, sample_limit=10, example_limit=10)
+    preview = await analysis.async_preview(ClassificationPreviewRequest(
+        policy=candidate, input={"kind": "media", "media": payload},
+    ))
+    assert impact.sample_count == impact.changed_count == 1
+    assert impact.changes[0].previous.recommended.category_id == "movie.foreign"
+    assert impact.changes[0].candidate.recommended.category_id == "movie.cn"
+    assert preview.result.recommended.category_id == "movie.cn"
+    recognize.assert_awaited_once_with(
+        media_source=MediaSource.TMDB, media_id="535167", mtype=MediaType.MOVIE, music_type=None,
+    )
+    assert source.origin_country == []
+    assert service.active() == active
+
+
 def test_field_catalog_exposes_typed_options_and_server_limits() -> None:
    """字段 API 数据足以驱动前端控件，无需硬编码值类型和结构限制。"""
    catalog = ClassificationAnalysisService(_service()).fields()
@@ -430,7 +511,8 @@ async def test_recent_history_samples_are_bounded_deduplicated_and_honest() -> N
        ),
    )

-    batch = await provider.load(10)
+    policy = build_default_classification_policy()
+    batch = await provider.load(10, active=policy, candidate=policy)

    assert batch.scanned_count == 4
    assert batch.skipped_count == 2
@@ -526,11 +608,14 @@ async def test_impact_analysis_resolves_complete_history_details_and_reports_gap
    complete_facts["11"].media.genre_keys = ["drama"]
    resolved_ids: list[str] = []

-    async def resolve_history(history: object) -> ClassificationFacts | None:
+    async def resolve_history(
+        history: object, _active: ClassificationPolicy, _candidate: ClassificationPolicy,
+    ) -> tuple[ClassificationFacts, ClassificationFacts] | None:
        """返回测试中的完整媒体事实，模拟详情接口缺失一条记录。"""
        media_id = str(getattr(history, "media_id", ""))
        resolved_ids.append(media_id)
-        return complete_facts.get(media_id)
+        facts = complete_facts.get(media_id)
+        return (facts, facts) if facts is not None else None

    provider = RecentHistoryClassificationSampleProvider(
        download_history=cast(object, _DownloadHistory(records)),
--- a/tests/test_media_classification_legacy.py
+++ b/tests/test_media_classification_legacy.py
@@ -247,12 +247,12 @@ def test_official_category_yaml_has_no_unknown_genre_ids() -> None:
        ("电视剧", {"origin_country": ["AU"]}, "未分类"),
    ],
 )
-def test_builtin_policy_matches_official_legacy_category_semantics(
+def test_builtin_and_migration_keep_explicit_region_naming(
    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(),
    )
@@ -262,7 +262,7 @@ def test_builtin_policy_matches_official_legacy_category_semantics(

    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
+    assert _policy_category_name(builtin, facts) == ("亚洲剧" if expected == "日韩剧" else expected)


 def test_deleted_legacy_rules_keep_alias_only_fields_registered() -> None:
--- a/tests/test_media_classification_startup.py
+++ b/tests/test_media_classification_startup.py
@@ -412,8 +412,10 @@ async def test_fresh_environment_initializes_builtin_default_policy(
        "movie.animation",
        "movie.chinese",
        "movie.foreign",
+        "movie.uncategorized",
        "tv.dongman.cn",
        "tv.dongman.jp",
+        "tv.animation.other",
        "tv.documentary",
        "tv.kids",
        "tv.variety",
@@ -430,8 +432,10 @@ async def test_fresh_environment_initializes_builtin_default_policy(
    assert [rule.id for rule in policy.rules] == [
        "movie.animation.default",
        "movie.chinese.default",
+        "movie.foreign.default",
        "tv.dongman.cn.default",
        "tv.dongman.jp.default",
+        "tv.animation.other.default",
        "tv.documentary.default",
        "tv.kids.default",
        "tv.variety.default",
@@ -444,7 +448,7 @@ async def test_fresh_environment_initializes_builtin_default_policy(
        "music.album.default",
    ]
    assert policy.fallbacks == {
-        "电影": "movie.foreign",
+        "电影": "movie.uncategorized",
        "电视剧": "tv.uncategorized",
        "音乐": "music.uncategorized",
    }