feat(classification): complete media preview and impact analysis

app/api/endpoints/classification.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from enum import Enum
+from functools import partial
 from typing import cast
 
 from fastapi import Depends, HTTPException, status
@@ -30,10 +32,13 @@ from app.application.classification.contract import (
     ClassificationPolicyConflictError,
     ClassificationPolicyStateCorruptError,
 )
+from app.application.classification.execution import ClassificationExecutionPort
 from app.application.classification.runtime import ClassificationRuntime
 from app.application.history import DownloadHistoryQueryPort
+from app.chain.media import MediaChain
 from app.schemas.category import (
     ClassificationEvaluation,
+    ClassificationFacts,
     ClassificationFieldCatalog,
     ClassificationImpactAnalysis,
     ClassificationImpactRequest,
@@ -48,6 +53,7 @@ from app.schemas.category import (
     ClassificationValidationResult,
 )
 from app.schemas.response import Response
+from app.schemas.types import MediaSource, MediaType
 from app.startup.composition.context import HostRuntime
 
 router = ResponseAPIRouter()
@@ -77,10 +83,61 @@ def _get_analysis_service(
                 runtime.history.download_repository(db),
             ),
             transfer_history=runtime.history.transfer_repository,
+            facts_resolver=partial(
+                _resolve_history_facts,
+                runtime.classification_execution,
+            ),
         ),
     )
 
 
+async def _resolve_history_facts(
+    execution: ClassificationExecutionPort,
+    history: object,
+) -> 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))
+    if not media_source or not media_id or media_type is None:
+        return None
+    try:
+        source = MediaSource(media_source)
+        media = await MediaChain().async_recognize_media(
+            media_source=source,
+            media_id=media_id,
+            mtype=media_type,
+            music_type=str(getattr(history, "music_type", None) or "").strip() or None,
+        )
+        if media is None:
+            return None
+        return await execution.async_build_facts(media)
+    except (TypeError, ValueError):
+        return None
+
+
+def _history_media_type(value: object) -> MediaType | None:
+    """兼容历史记录中的中文和英文媒体类型。"""
+    normalized = _enum_text(value).casefold()
+    aliases = {
+        "电影": MediaType.MOVIE,
+        "movie": MediaType.MOVIE,
+        "电视剧": MediaType.TV,
+        "tv": MediaType.TV,
+        "电视": MediaType.TV,
+        "音乐": MediaType.MUSIC,
+        "music": MediaType.MUSIC,
+    }
+    return aliases.get(normalized)
+
+
+def _enum_text(value: object) -> str:
+    """把枚举或普通值转换为去除首尾空白的文本。"""
+    if isinstance(value, Enum):
+        value = value.value
+    return str(value or "").strip()
+
+
 def _require_active_policy(runtime: ClassificationRuntime) -> ClassificationPolicy:
     """读取活动策略；启动迁移失败时映射为可诊断的 503。"""
     try:
@@ -217,7 +274,7 @@ async def preview_policy(
     _: object = Depends(get_current_active_user_async),
     runtime: ClassificationRuntime = Depends(get_classification_runtime),
 ) -> ClassificationEvaluation | JSONResponse:
-    """对显式标准事实执行活动策略或未发布草稿并返回完整 trace。"""
+    """对选择的媒体信息或兼容事实执行策略，并返回完整匹配说明。"""
     if request.policy is None:
         _require_active_policy(runtime)
     try:
@@ -237,7 +294,7 @@ async def analyze_impact(
     _: object = Depends(get_current_active_superuser_async),
     service: ClassificationAnalysisService = Depends(_get_analysis_service),
 ) -> ClassificationImpactAnalysis | JSONResponse:
-    """比较活动策略与草稿；样本有限且不触发联网识别或任何写入。"""
+    """读取近期历史对应的完整媒体详情后比较策略，不修改媒体或历史数据。"""
     try:
         return await service.impact(
             request.policy,

app/application/classification/analysis.py

@@ -4,11 +4,12 @@ from __future__ import annotations
 
 import asyncio
 from collections import Counter
-from collections.abc import Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Literal, Optional, Protocol, cast
+from types import SimpleNamespace
+from typing import Any, Literal, Optional, Protocol, TypeAlias, cast
 
 from app.application.classification.catalog import (
     build_classification_field_catalog,
@@ -25,6 +26,7 @@ from app.application.history import (
     TransferHistoryQueryPort,
     TransferHistorySnapshot,
 )
+from app.domain.classification.facts import build_classification_facts
 from app.domain.classification.evaluator import ClassificationEvaluator
 from app.domain.classification.validation import (
     MAX_CATEGORY_DEPTH,
@@ -43,17 +45,28 @@ from app.schemas.category import (
     ClassificationImpactAnalysis,
     ClassificationImpactChange,
     ClassificationImpactGroup,
+    ClassificationFactValue,
     ClassificationMediaFacts,
     ClassificationMediaType,
     ClassificationMusicFacts,
     ClassificationPolicy,
     ClassificationPolicyLimits,
+    ClassificationPreviewInput,
     ClassificationPreviewRequest,
     ClassificationResult,
     ClassificationValidationResult,
 )
+from app.schemas.context import MediaInfo as SchemaMediaInfo
+from app.schemas.music import MusicInfo as SchemaMusicInfo
 
 _UNCLASSIFIED_CATEGORY_ID = "__unclassified__"
+_DEFAULT_RESOLVE_CONCURRENCY = 3
+
+ClassificationImpactFactsResolver: TypeAlias = Callable[
+    [DownloadHistorySnapshot | TransferHistorySnapshot],
+    Awaitable[ClassificationFacts | None],
+]
+"""按历史记录重新读取完整媒体信息的异步端口。"""
 
 
 @dataclass(frozen=True, slots=True)
@@ -64,6 +77,8 @@ class ClassificationImpactSampleBatch:
     facts: tuple[ClassificationFacts, ...]
     scanned_count: int
     skipped_count: int
+    unresolved_count: int = 0
+    truncated: bool = False
     warnings: tuple[str, ...] = ()
 
 
@@ -83,10 +98,16 @@ class RecentHistoryClassificationSampleProvider:
         *,
         download_history: DownloadHistoryQueryPort,
         transfer_history: TransferHistoryQueryPort,
+        facts_resolver: ClassificationImpactFactsResolver | None = None,
+        resolve_concurrency: int = _DEFAULT_RESOLVE_CONCURRENCY,
     ) -> None:
         """保存由 API 请求或宿主运行时提供的只读历史端口。"""
+        if resolve_concurrency <= 0:
+            raise ValueError("影响分析详情读取并发上限必须大于 0")
         self._download_history = download_history
         self._transfer_history = transfer_history
+        self._facts_resolver = facts_resolver
+        self._resolve_concurrency = resolve_concurrency
 
     async def load(self, limit: int) -> ClassificationImpactSampleBatch:
         """合并两类近期历史，按时间和 ID 排序后去重并投影事实。"""
@@ -106,7 +127,7 @@ class RecentHistoryClassificationSampleProvider:
             key=lambda item: (item.date or "", item.record_id, item.kind),
             reverse=True,
         )
-        facts: list[ClassificationFacts] = []
+        unique_records: list[tuple[_HistorySampleRecord, ClassificationFacts]] = []
         seen: set[tuple[str, str, str, str]] = set()
         skipped_count = 0
         for record in records:
@@ -126,19 +147,77 @@ class RecentHistoryClassificationSampleProvider:
                 skipped_count += 1
                 continue
             seen.add(identity_key)
-            facts.append(projected)
-            if len(facts) >= limit:
-                break
+            unique_records.append((record, projected))
+
+        if self._facts_resolver is None:
+            facts = [projected for _, projected in unique_records[:limit]]
+            skipped_count += max(0, len(unique_records) - len(facts))
+            return ClassificationImpactSampleBatch(
+                source="recent_history",
+                facts=tuple(facts),
+                scanned_count=len(records),
+                skipped_count=skipped_count,
+                truncated=len(unique_records) > limit,
+                warnings=(
+                    "近期历史仅稳定保存媒体身份、类型、标题和年份；其它字段缺失时相关规则不会命中",
+                ),
+            )
+
+        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)
+        facts = []
+        unresolved_count = 0
+        for item in resolved:
+            if item is None:
+                unresolved_count += 1
+                skipped_count += 1
+                continue
+            facts.append(item)
+
+        warnings = [
+            "系统会按近期下载和整理记录中的来源与编号重新读取完整媒体信息；无法读取的记录不会参与比较",
+        ]
+        if unresolved_count:
+            warnings.append(
+                f"{unresolved_count} 条记录无法获取完整媒体信息，未纳入比较",
+            )
+        if len(unique_records) > limit:
+            warnings.append(f"符合条件的记录超过 {limit} 条，本次最多比较 {limit} 条")
         return ClassificationImpactSampleBatch(
             source="recent_history",
             facts=tuple(facts),
             scanned_count=len(records),
             skipped_count=skipped_count,
-            warnings=(
-                "近期历史仅稳定保存媒体身份、类型、标题和年份；其它字段缺失时相关规则不会命中",
-            ),
+            unresolved_count=unresolved_count,
+            truncated=len(unique_records) > limit,
+            warnings=tuple(warnings),
         )
 
+    async def _resolve_records(
+        self,
+        records: Sequence[tuple[_HistorySampleRecord, ClassificationFacts]],
+    ) -> list[ClassificationFacts | None]:
+        """以固定并发上限重新读取详情，并拒绝身份不一致的返回值。"""
+        if self._facts_resolver is None:
+            return [projected for _, projected in records]
+        semaphore = asyncio.Semaphore(self._resolve_concurrency)
+
+        async def resolve(
+            record: _HistorySampleRecord,
+            projected: ClassificationFacts,
+        ) -> ClassificationFacts | None:
+            async with semaphore:
+                try:
+                    facts = await self._facts_resolver(record.payload)
+                except Exception:  # noqa: BLE001  单条详情失败不应阻断整批分析
+                    return None
+            if facts is None or _classification_identity_key(facts) != _classification_identity_key(projected):
+                return None
+            return facts
+
+        return list(await asyncio.gather(*(resolve(record, projected) for record, projected in records)))
+
 
 @dataclass(frozen=True, slots=True)
 class _HistorySampleRecord:
@@ -192,13 +271,13 @@ 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)
         return ClassificationEvaluator.evaluate(
             policy,
-            request.input.facts,
+            _preview_facts(request.input),
             trace=True,
         )
 
@@ -258,6 +337,8 @@ class ClassificationAnalysisService:
                 facts=selected,
                 scanned_count=len(samples),
                 skipped_count=0,
+                unresolved_count=0,
+                truncated=len(samples) > sample_limit,
                 warnings=warnings,
             )
         if self._sample_provider is None:
@@ -266,6 +347,8 @@ class ClassificationAnalysisService:
                 facts=(),
                 scanned_count=0,
                 skipped_count=0,
+                unresolved_count=0,
+                truncated=False,
                 warnings=("近期历史样本提供器未配置，本次影响分析没有可比较样本",),
             )
         return await self._sample_provider.load(sample_limit)
@@ -299,6 +382,70 @@ def _history_facts(
     )
 
 
+def _classification_identity_key(facts: ClassificationFacts) -> tuple[str, str, str, str]:
+    """返回媒体详情可用于核对的来源、编号、类型和音乐实体键。"""
+    return (
+        facts.identity.media_source,
+        facts.identity.media_id,
+        facts.media.type,
+        facts.music.entity_type if facts.music and facts.music.entity_type else "",
+    )
+
+
+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 build_classification_facts_from_media(media: object) -> ClassificationFacts:
+    """把搜索或识别得到的完整媒体对象转换为统一分类数据。"""
+    return build_classification_facts(
+        cast(Any, media),
+        extensions=_media_extension_facts(media),
+    )
+
+
+def build_classification_facts_from_media_payload(
+    payload: Mapping[str, Any],
+) -> 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)
+
+
+def _media_extension_facts(media: object) -> dict[str, dict[str, ClassificationFactValue]]:
+    """按 extensions.<source>.<field> 命名空间整理媒体携带的扩展字段。"""
+    raw_facts = getattr(media, "classification_facts", None)
+    if not isinstance(raw_facts, Mapping):
+        return {}
+    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]:
+            continue
+        if not _is_classification_fact_value(value):
+            continue
+        extensions.setdefault(parts[1], {})[parts[2]] = cast(ClassificationFactValue, value)
+    return extensions
+
+
+def _is_classification_fact_value(value: object) -> bool:
+    """判断扩展字段是否为分类契约允许的 JSON 标量或标量列表。"""
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return True
+    return isinstance(value, list) and all(
+        item is None or isinstance(item, (str, int, float, bool)) for item in value
+    )
+
+
 def _classification_media_type(value: object) -> ClassificationMediaType | None:
     """兼容历史中使用的中英文媒体类型值。"""
     normalized = _enum_text(value).casefold()
@@ -397,7 +544,8 @@ def _build_impact_analysis(
     sample_count = len(batch.facts)
     changed_count = len(changes)
     truncated = (
-        batch.scanned_count > sample_count + batch.skipped_count
+        batch.truncated
+        or batch.scanned_count > sample_count + batch.skipped_count
         or changed_count > example_limit
     )
     return ClassificationImpactAnalysis(
@@ -408,6 +556,7 @@ def _build_impact_analysis(
         requested_limit=requested_limit,
         scanned_count=batch.scanned_count,
         skipped_count=batch.skipped_count,
+        unresolved_count=batch.unresolved_count,
         truncated=truncated,
         sample_count=sample_count,
         changed_count=changed_count,

app/application/classification/execution.py

@@ -50,6 +50,13 @@ class ClassificationRuntimePort(Protocol):
 class ClassificationExecutionPort(Protocol):
     """Chain、订阅和整理应用层共享的纯分类执行端口。"""
 
+    async def async_build_facts(
+        self,
+        media: ClassificationSubject,
+    ) -> ClassificationFacts | None:
+        """异步构造与实际分类一致的完整事实快照，不写入媒体或策略。"""
+        ...
+
     def finalize(
         self,
         media: ClassificationSubject,
@@ -118,6 +125,26 @@ class ClassificationExecutionService:
         self._extension_facts_provider = extension_facts_provider
         self._enrichment = enrichment
 
+    async def async_build_facts(
+        self,
+        media: ClassificationSubject,
+    ) -> ClassificationFacts | None:
+        """构造影响分析使用的完整事实，并复用插件扩展与跨来源补充规则。"""
+        finalized, policy, facts, _ = self._prepare(
+            media,
+            extensions=None,
+            effective_override=None,
+            refresh=False,
+        )
+        if policy is None or facts is None:
+            return None
+        if self._enrichment is not None:
+            try:
+                facts = await self._enrichment.async_enrich(policy, facts, finalized)
+            except Exception:  # noqa: BLE001  详情补充失败时保留主来源事实
+                pass
+        return facts
+
     def finalize(
         self,
         media: ClassificationSubject,

app/schemas/category.py

@@ -3,6 +3,8 @@ from typing import Dict, Literal, Optional, TypeAlias, Union
 
 from pydantic import BaseModel, ConfigDict, Field, RootModel, field_validator, model_validator
 
+from app.schemas.common import JsonData
+
 
 class CategoryRule(BaseModel):
     """
@@ -523,8 +525,32 @@ class ClassificationFactsPreviewInput(_ClassificationModel):
     facts: ClassificationFacts = Field(description="本次预览使用的标准化分类事实")
 
 
-ClassificationPreviewInput: TypeAlias = ClassificationFactsPreviewInput
-"""首版预览输入联合；C1 将在不破坏 facts 形状的前提下增加身份和历史输入。"""
+class ClassificationMediaPreviewInput(_ClassificationModel):
+    """从媒体搜索结果选择的完整媒体信息，用于直接预览分类结果。"""
+
+    kind: Literal["media"] = Field(default="media", description="预览输入类型")
+    media: dict[str, JsonData] = Field(description="从媒体搜索结果选择的媒体信息")
+
+    @model_validator(mode="after")
+    def validate_media_identity(self) -> "ClassificationMediaPreviewInput":
+        """确保搜索结果包含分类所需的来源、编号和媒体类型。"""
+        source = str(self.media.get("media_source") or "").strip()
+        media_id = str(self.media.get("media_id") or "").strip()
+        media_type = str(self.media.get("type") or "").strip()
+        if not source:
+            raise ValueError("选择的媒体缺少数据来源")
+        if not media_id:
+            raise ValueError("选择的媒体缺少媒体编号")
+        if media_type not in {"电影", "电视剧", "音乐"}:
+            raise ValueError("选择的媒体类型不受分类规则支持")
+        return self
+
+
+ClassificationPreviewInput: TypeAlias = Union[
+    ClassificationFactsPreviewInput,
+    ClassificationMediaPreviewInput,
+]
+"""预览输入联合；前端通常提交搜索结果，旧调用仍可提交标准事实。"""
 
 
 class ClassificationPreviewRequest(_ClassificationModel):
@@ -603,7 +629,11 @@ class ClassificationImpactAnalysis(_ClassificationModel):
     candidate_revision: int = Field(ge=2, description="候选策略预计发布 revision")
     requested_limit: int = Field(ge=1, le=200, description="请求的最大样本数量")
     scanned_count: int = Field(ge=0, description="为生成样本实际扫描的记录数量")
-    skipped_count: int = Field(ge=0, description="因身份缺失、类型无效或重复而跳过的记录数量")
+    skipped_count: int = Field(ge=0, description="未参与比较的记录数量")
+    unresolved_count: int = Field(
+        ge=0,
+        description="身份有效但无法重新获取完整媒体信息的记录数量",
+    )
     truncated: bool = Field(description="是否因样本或示例上限截断结果")
     sample_count: int = Field(ge=0, description="实际参与比较的唯一有效样本数量")
     changed_count: int = Field(ge=0, description="分类结果发生变化的样本数量")

docs/architecture/media-classification-design.md

@@ -1,6 +1,6 @@
 # MoviePilot 多媒体、多数据源自动分类体系设计
 
-- 状态：Draft
+- 状态：V3 首版实现依据
 - 日期：2026-09-02
 - 适用版本：MoviePilot V3
 - 涉及仓库：`MoviePilot`、`MoviePilot-Frontend`，后续插件能力文档涉及 `MoviePilot-Plugins`
@@ -418,6 +418,10 @@ TheMovieDb 模块内部的专用能力：
 豆瓣类型名、Bangumi 标签、AniList Genre 等转换为这些规范键；无法规范化的值仍保留在
 `genre_names` 或音乐 `genres/tags` 中。
 
+在界面中，这些内容统一称为“媒体信息”。“统一风格”只是为了让不同数据源的同一种风格可以使用
+同一条规则匹配，普通用户不需要理解内部键名；搜索并选择媒体后，页面会同时展示来源返回的风格名称
+和用于规则匹配的统一风格。
+
 ### 5.6 插件扩展事实
 
 插件媒体来源可以声明额外分类字段，但必须满足：
@@ -501,9 +505,8 @@ UI 根据用户选择的媒体类型和来源过滤字段，并显示覆盖提
 
 ### 6.2 可选跨源补充
 
-首版默认使用 `primary_only`：只使用当前识别结果和本地解析事实。
-
-后续可增加 `enrich_missing`：
+首版默认使用 `primary_only`：只使用当前识别结果和本地解析事实。选择 `enrich_missing` 后，
+分类执行服务和影响分析会在需要时补充缺失的标准媒体信息：
 
 - 只请求当前策略实际引用且当前缺失的标准事实。
 - 使用独立可选模块方法 `get_media_classification_facts`。
@@ -655,8 +658,10 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 
 字段目录响应同时返回服务端规则限制。枚举型字段的 `options` 使用稳定的
 `{value, label}` 对象，`allow_custom_values` 明确表示前端是否允许目录外输入；前端不得根据字段名硬编码
-输入控件或限制值。预览请求使用带 `kind` 判别字段的输入联合体，首期支持
-`{"kind": "facts", "facts": {...}}`，后续身份查询和历史选择可在不破坏现有客户端的前提下增加分支。
+输入控件或限制值。预览请求使用带 `kind` 判别字段的输入联合体，当前支持：
+
+- `{"kind": "media", "media": {...}}`：前端搜索并选择的完整媒体信息。
+- `{"kind": "facts", "facts": {...}}`：保留给旧调用方和集成测试的标准分类信息。
 
 发布、回滚和影响分析都携带 `expected_revision`。revision 不一致时返回标准 `Response` 包装的
 `409`，`data` 为 `{expected_revision, current_revision}`；领域校验失败返回标准 `Response` 包装的
@@ -719,17 +724,16 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 
 ### 11.1 信息架构
 
-分类策略继续从“设置 -> 目录”进入，但打开独立的大尺寸编辑器。桌面端采用三栏工作区，移动端按步骤切换：
+分类策略从“设置 -> 目录”中的“自动分类策略”入口进入，点击后打开全屏、可滚动的编辑窗口；分类编辑器在窗口内工作。
+窗口内按职责分为“分类树、规则、来源、检查与发布”四个工作区标签；最后一个工作区再分为“结果预览、
+影响分析、版本发布与历史”三个标签。移动端保持同样的分层顺序，不把所有表单压缩到同一屏：
 
 ```text
-+----------------+-------------------------+----------------------+
-| 媒体类型/分类树 | 当前分类的规则列表      | 规则条件检查器       |
-| 电影            | 1. 日本动画            | 数据源：全部         |
-| 电视剧          | 2. 国产剧              | 条件：全部满足       |
-| 音乐            | 3. 未分类              | 字段/操作符/值       |
-+----------------+-------------------------+----------------------+
-| 校验结果 / 单条预览 / 发布前影响分析 / 保存发布                |
-+----------------------------------------------------------------+
++------------------+------------------+------------------+----------------------+
+| 分类树           | 规则             | 来源             | 检查与发布           |
+| 电影 / 电视剧    | 有序规则列表     | 来源默认分类     | 结果预览             |
+| 音乐             | 条件和输出       | 影视 / 音乐      | 影响分析 / 版本历史  |
++------------------+------------------+------------------+----------------------+
 ```
 
 不得继续在每条规则中堆叠固定的 TMDB 表单控件。
@@ -782,11 +786,10 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 
 ### 11.6 单条测试
 
-支持三种测试输入：
+页面支持两种测试输入：
 
-1. 从名称测试结果直接带入当前媒体。
-2. 选择 `media_source + media_id` 获取详情后测试。
-3. 从最近整理历史选择一项重新求值。
+1. 输入关键词搜索并选择一条媒体结果，直接使用结果中的完整媒体信息。
+2. 兼容调用方提交已经构造好的 `ClassificationFacts`；页面不要求用户手工填写来源编号或字段值。
 
 测试面板显示：
 
@@ -799,7 +802,7 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 
 ### 11.7 发布前影响分析
 
-点击“应用”前可用草稿策略对最近 N 条整理历史或订阅样本执行只读比较：
+点击“发布”前可用草稿策略对最近 N 条下载和成功整理历史执行只读比较：
 
 - 分类未变化数量。
 - 分类发生变化数量。
@@ -810,10 +813,9 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 变化较大时由用户二次确认，但不使用模糊的“可能有风险”提示，应明确列出影响数量和示例。
 
 首期后端使用最近下载历史与成功整理历史合并抽样，按媒体身份去重，最多比较 200 条、返回 50 条
-变化示例。历史表只稳定保存身份、类型、标题和年份，因此结果固定标记为估算并返回缺失事实警告；
-`scanned/skipped/truncated` 和按来源、类型分组统计必须随响应返回。显式事实样本优先于历史抽样，
-两种路径都只做纯求值，不触发数据源网络请求、文件移动、订阅修改或历史写入；批量求值离开 API
-事件循环执行。
+变化示例。历史表只保存身份、类型、标题和年份，因此服务端会按记录中的来源和编号重新读取完整媒体详情；
+单条读取失败会计入“无法获取详情”，不会被误算为没有变化。`scanned/skipped/unresolved/truncated`
+和按来源、类型分组统计随响应返回；整个过程不移动文件、不修改订阅或历史，且对详情读取设置固定并发上限。
 
 ### 11.8 无障碍与移动端
 
@@ -951,6 +953,8 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 1. 第一个版本同时写 `library_category` 和兼容 `category`，读侧优先新字段。
 2. 前端、目录、订阅、下载、历史、通知全部迁移后，`category` 只保留序列化兼容属性。
 3. 音乐 `category` 来源数据迁入 `metadata_category`，音乐 UI 不再读取兼容 `category` 展示专辑类型。
+4. 迁移完成后，前端默认隐藏未被现有规则引用的 TMDB 旧字段；仍被旧规则引用的字段会以“旧规则”分组只读显示，
+   直到用户把规则改为新的统一字段。
 
 ## 14. 校验规则
 
@@ -1021,6 +1025,7 @@ API 常规读取只返回 `active`，历史接口按需读取 `history`。选择
 - 分类被引用时禁止删除。
 - revision 冲突时保留本地草稿并提供重新加载/合并选择。
 - 预览展示实际值、失败条件和最终目录。
+- 预览必须通过搜索选择媒体，不再要求用户手工填写事实；编辑分类弹窗和帮助弹窗在透明主题下保持玻璃表面、边框和遮罩。
 - 桌面和窄屏布局不重叠，键盘可完成排序和编辑。
 
 ### 15.5 性能目标

docs/v2-to-v3-overview.md

@@ -104,7 +104,7 @@ V3 将通用媒体身份统一为：
 
 V2 自动分类主要由 TMDB 详情和 `category.yaml` 驱动，只覆盖电影、电视剧。V3 将自动分类独立为统一能力，不再属于某一个元数据来源：
 
-- 电影、电视剧和音乐在同一个“设置 → 自动分类”界面中维护分类树和规则。
+- 电影、电视剧和音乐在“设置 → 目录 → 自动分类策略”打开的全屏窗口中维护分类树和规则。
 - 规则可以使用媒体类型、年份、国家、类型、音乐实体、专辑类型、来源范围等标准事实。
 - TMDB、豆瓣、Bangumi、AniList、IMDb、TVDB、MusicBrainz、TheAudioDB、豆瓣音乐和已登记插件来源都可以进入同一分类流程。
 - 来源专用字段由宿主或插件在受控命名空间中声明，规则编辑器根据后端字段目录动态生成，不再把 TMDB 字段硬编码到前端。
@@ -112,10 +112,17 @@ V2 自动分类主要由 TMDB 详情和 `category.yaml` 驱动，只覆盖电影
 
 分类策略按 revision 版本化保存。发布前必须先通过服务端校验，并可以使用近期历史样本执行有界影响分析；发布后可以查看历史版本和回滚。并发编辑会提示 revision 冲突，不会静默覆盖另一位管理员的修改。
 
+使用时不需要手工填写分类事实：在“结果预览”中输入关键词并选择媒体，系统直接使用搜索结果中的标题、年份、风格、
+国家/地区、分级以及音乐流派、标签和艺术家等信息。影响分析会读取近期下载和整理记录，再按记录中的数据源和编号
+重新获取完整媒体详情；无法获取详情的记录会单独统计，不会被当成“没有变化”。
+
 目录、订阅、下载和整理历史不再只保存易变的分类名称，而是同时记录稳定 `category_id` 和当时的路径快照。分类改名或调整路径后，既有目录和订阅仍能解析到同一个分类；已经建立的整理计划会继续使用计划创建时冻结的分类目标，不会被后续策略修改。
 
 升级时，如果尚未存在 V3 分类策略，系统会读取现有 `category.yaml` 并自动迁移；旧 TMDB 规则的顺序、排除条件、年份范围和兜底语义会保留。迁移完成后不再继续写入 YAML。旧 `GET /api/v1/media/category` 和 `GET /api/v1/media/category/config` 暂时保留为只读投影，旧 `POST /api/v1/media/category/config` 已移除，所有新写入都通过带 revision 校验的策略接口完成。
 
+迁移完成后，前端默认不再显示没有被现有规则引用的 TMDB 旧字段；仍被旧规则使用的字段会标记为“旧规则”，只能查看，
+改写规则时应优先选择新的统一字段。分类编辑弹窗和帮助弹窗支持透明主题，分类下拉选项会避免名称与路径末级重复显示。
+
 ### 3.4 搜索、下载和整理流程更容易处理异常情况
 
 V3 对日常高频操作做了多项增强：
@@ -197,7 +204,7 @@ V3 前端仍然基于 Vue 3、Vuetify 3 和 Vite，并不是推倒重写。因
 - 新增音乐首页、音乐搜索、歌曲详情、专辑详情和艺术家详情页面。
 - 搜索、订阅、探索、推荐、整理、缓存和历史页面支持音乐实体。
 - 新增数据库备份管理面板。
-- 新增统一自动分类设置页，可在同一界面编辑电影、电视剧、音乐分类，预览命中过程并查看发布影响。
+- 目录设置新增“自动分类策略”入口，打开全屏窗口后可编辑电影、电视剧、音乐分类，预览命中过程并查看发布影响。
 - 插件市场支持虚拟分身、来源绑定和换源。
 - 新增首次初始化页面，移除原来体量较大的全功能设置向导。
 - AI 助手支持全屏显示和受保护操作交互。
@@ -493,7 +500,7 @@ V3 本地运行要求 Python 3.14+，优先使用项目 `.venv` 和锁定依赖
 - 确认原账号可以登录，订阅、站点、目录和下载器配置仍在。
 - 检查数据库迁移日志和备份记录。
 - 在“设置 → 关于”确认后端、前端和站点资源均为 V3。
-- 打开“设置 → 自动分类”，确认电影、电视剧分类已从旧配置迁移，并按需补充音乐规则。
+- 打开“设置 → 目录 → 自动分类策略”，确认电影、电视剧分类已从旧配置迁移，并按需补充音乐规则。
 - 手动执行一次站点测试和影视搜索。
 - 选择一个已知资源测试下载与整理。
 - 检查通知渠道和媒体服务器刷新。

tests/test_media_classification_api.py

@@ -272,6 +272,74 @@ def test_preview_returns_condition_path_and_structured_missing_fact_warning() ->
     assert evaluation.warnings[0].source == "themoviedb"
 
 
+def test_preview_selected_media_uses_complete_movie_and_music_details() -> None:
+    """选择搜索结果后应直接使用影视和音乐的完整标准字段进行预览。"""
+    analysis = ClassificationAnalysisService(_service())
+
+    movie = analysis.preview(
+        ClassificationPreviewRequest(
+            input={
+                "kind": "media",
+                "media": {
+                    "media_source": "themoviedb",
+                    "media_id": "550",
+                    "type": "电影",
+                    "title": "搏击俱乐部",
+                    "year": "1999",
+                    "original_language": "en",
+                    "origin_country": ["US"],
+                    "genre_ids": [878],
+                    "genres": [{"id": 878, "name": "科幻"}],
+                    "content_rating": "R",
+                },
+            }
+        )
+    )
+
+    assert movie.facts.identity.media_id == "550"
+    assert movie.facts.media.year == 1999
+    assert movie.facts.media.language == "en"
+    assert movie.facts.media.countries == ["US"]
+    assert movie.facts.media.genre_keys == ["science_fiction"]
+    assert movie.facts.media.genre_names == ["科幻"]
+    assert movie.facts.media.content_rating == "R"
+
+    music = analysis.preview(
+        ClassificationPreviewRequest(
+            input={
+                "kind": "media",
+                "media": {
+                    "media_source": "musicbrainz",
+                    "media_id": "release-1",
+                    "type": "音乐",
+                    "music_type": "album",
+                    "title": "现场专辑",
+                    "album": "现场专辑",
+                    "album_type": "album",
+                    "secondary_types": ["Live"],
+                    "year": 2020,
+                    "genres": ["摇滚"],
+                    "tags": ["现场"],
+                    "artists": ["示例乐队"],
+                    "artist_country": "英国",
+                    "release_status": "official",
+                },
+            }
+        )
+    )
+
+    assert music.facts.identity.media_id == "release-1"
+    assert music.facts.media.type == "音乐"
+    assert music.facts.media.countries == ["GB"]
+    assert music.facts.media.genre_keys == ["rock"]
+    assert music.facts.music is not None
+    assert music.facts.music.entity_type == "album"
+    assert music.facts.music.secondary_types == ["Live"]
+    assert music.facts.music.genres == ["摇滚"]
+    assert music.facts.music.tags == ["现场"]
+    assert music.facts.music.artists == ["示例乐队"]
+
+
 @pytest.mark.asyncio  # type: ignore[misc]
 async def test_recent_history_samples_are_bounded_deduplicated_and_honest() -> None:
     """近期历史样本按身份去重，脏记录跳过，并明确只包含基础事实。"""
@@ -338,6 +406,101 @@ 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_impact_analysis_resolves_complete_history_details_and_reports_gaps() -> None:
+    """影响分析应以历史身份重新读取完整字段，并单独统计无法读取的记录。"""
+    service = _service()
+    policy = build_default_classification_policy()
+    policy.categories.append(
+        ClassificationCategory(
+            id="movie.science-fiction",
+            media_type="电影",
+            name="科幻电影",
+            path=["科幻"],
+        )
+    )
+    policy.rules.append(
+        ClassificationRule(
+            id="rule.movie.science-fiction",
+            name="科幻电影",
+            kind="category",
+            media_types=["电影"],
+            when=ClassificationCondition(
+                field="media.genre_keys",
+                operator="contains_any",
+                value=["science_fiction"],
+            ),
+            target=ClassificationTarget(category_id="movie.science-fiction"),
+        )
+    )
+    records = [
+        DownloadHistorySnapshot(
+            id=10,
+            path="/downloads/movie-10",
+            type="电影",
+            title="基础标题",
+            media_source=MediaSource.TMDB,
+            media_id="10",
+            date="2026-09-02 10:00:00",
+        ),
+        DownloadHistorySnapshot(
+            id=11,
+            path="/downloads/movie-11",
+            type="电影",
+            title="完整标题",
+            media_source=MediaSource.TMDB,
+            media_id="11",
+            date="2026-09-02 09:00:00",
+        ),
+        DownloadHistorySnapshot(
+            id=12,
+            path="/downloads/movie-12",
+            type="电影",
+            title="无法读取",
+            media_source=MediaSource.TMDB,
+            media_id="12",
+            date="2026-09-02 08:00:00",
+        ),
+    ]
+    complete_facts = {
+        "10": _facts(media_id="10"),
+        "11": _facts(media_id="11"),
+    }
+    complete_facts["10"].media.genre_keys = ["science_fiction"]
+    complete_facts["11"].media.genre_keys = ["drama"]
+    resolved_ids: list[str] = []
+
+    async def resolve_history(history: object) -> ClassificationFacts | None:
+        """返回测试中的完整媒体事实，模拟详情接口缺失一条记录。"""
+        media_id = str(getattr(history, "media_id", ""))
+        resolved_ids.append(media_id)
+        return complete_facts.get(media_id)
+
+    provider = RecentHistoryClassificationSampleProvider(
+        download_history=cast(object, _DownloadHistory(records)),
+        transfer_history=cast(object, _TransferHistory([])),
+        facts_resolver=resolve_history,
+    )
+    result = await ClassificationAnalysisService(
+        service,
+        sample_provider=provider,
+    ).impact(
+        policy,
+        expected_revision=1,
+        sample_limit=10,
+        example_limit=20,
+    )
+
+    assert resolved_ids == ["10", "11", "12"]
+    assert result.scanned_count == 3
+    assert result.sample_count == 2
+    assert result.skipped_count == 1
+    assert result.unresolved_count == 1
+    assert result.changed_count == 1
+    assert result.changes[0].identity.media_id == "10"
+    assert any("无法获取完整媒体信息" in warning for warning in result.warnings)
+
+
 @pytest.mark.asyncio  # type: ignore[misc]
 async def test_impact_analysis_checks_revision_and_preserves_statistics() -> None:
     """影响分析以活动 revision 为基线，并保持总量、分组和变化示例一致。"""

tests/test_media_classification_execution.py

@@ -181,6 +181,33 @@ def test_execution_classifies_copy_and_preserves_source_identity() -> None:
     assert source.classification.policy_revision == 1
 
 
+def test_execution_builds_complete_facts_without_mutating_media() -> None:
+    """影响分析事实入口应复用插件字段构造，并保持原媒体对象不变。"""
+    source = MediaInfo(
+        media_source=MediaSource.Douban,
+        media_id="native-1",
+        type=MediaType.MOVIE,
+        title="Example",
+        genres=[{"name": "动画"}],
+        origin_country=["JP"],
+    )
+    service = ClassificationExecutionService(
+        _Runtime(_policy()),
+        extension_facts_provider=lambda media: {
+            "example.source": {"region_group": "east-asia"}
+        },
+    )
+
+    facts = asyncio.run(service.async_build_facts(source))
+
+    assert facts is not None
+    assert facts.identity.media_id == "native-1"
+    assert facts.media.genre_keys == ["animation"]
+    assert facts.media.countries == ["JP"]
+    assert facts.extensions["example.source"]["region_group"] == "east-asia"
+    assert source.classification is None
+
+
 def test_execution_reclassifies_cached_result_after_policy_revision_changes() -> None:
     """缓存对象携带旧 revision 时必须按当前策略重新分类。"""
     runtime = _Runtime(_policy(revision=7))