MoviePilot/app/application/classification/execution.py

"""
完整媒体识别结果的分类执行与兼容回退服务。
"""

from __future__ import annotations

from collections.abc import Callable, Mapping
from copy import deepcopy
from enum import Enum
from typing import Final, Protocol, TypeAlias

from app.application.classification.legacy import (
    build_legacy_tmdb_extension_facts,
    resolve_legacy_tmdb_category,
)
from app.domain.classification.evaluator import ClassificationEvaluator
from app.domain.classification.facts import build_classification_facts
from app.domain.context import MediaInfo, MusicAlbumInfo, MusicArtistInfo, MusicInfo
from app.schemas.category import (
    CategoryConfig,
    ClassificationEvaluation,
    ClassificationFacts,
    ClassificationFactValue,
    ClassificationPolicy,
    ClassificationResult,
    ClassificationSelection,
)

# 可以在完整识别出口执行自动分类的标准媒体对象。
ClassificationSubject: TypeAlias = (
    MediaInfo | MusicInfo | MusicAlbumInfo | MusicArtistInfo
)

# 按当前插件注册表校验并提供来源扩展分类事实的端口。
ClassificationExtensionFactsProvider: TypeAlias = Callable[
    [ClassificationSubject],
    Mapping[str, Mapping[str, ClassificationFactValue]],
]

_LEGACY_TMDB_SOURCE: Final[str] = "themoviedb"
_LEGACY_RULE_PREFIX: Final[str] = "legacy."


class ClassificationRuntimePort(Protocol):
    """
    分类执行只需要的活动策略与旧格式快照端口。
    """

    def active_policy(self) -> ClassificationPolicy | None:
        """
        返回活动策略；配置不可用时返回空值。

        :return: 当前活动策略；配置尚未就绪或不可用时返回 None
        """
        ...

    def legacy_config(self) -> CategoryConfig:
        """
        返回隔离的旧格式配置快照。

        :return: 与运行时状态隔离的旧格式分类配置副本
        """
        ...


class ClassificationExecutionPort(Protocol):
    """
    Chain、订阅和整理应用层共享的分类执行端口。
    """

    async def async_build_facts(
        self,
        media: ClassificationSubject,
        *,
        policy: ClassificationPolicy | None = None,
    ) -> ClassificationFacts | None:
        """
        按指定或活动策略异步准备事实，供预览和新旧策略比较复用。

        :param media: 已完成媒体识别的标准对象，构造事实时不原地修改
        :param policy: 构造事实采用的策略；None 使用当前活动策略
        :return: 按策略需要补充后的事实；策略不可用或媒体无法建立标准事实时返回 None
        """
        ...

    def build_facts(
        self,
        media: ClassificationSubject,
        *,
        policy: ClassificationPolicy | None = None,
    ) -> ClassificationFacts | None:
        """
        同步构造与实际分类一致的完整事实快照，不写入媒体或策略。

        :param media: 已完成媒体识别的标准对象，构造事实时不原地修改
        :param policy: 构造事实采用的策略；None 使用当前活动策略
        :return: 按策略需要补充后的事实；策略不可用或媒体无法建立标准事实时返回 None
        """
        ...

    def finalize(
        self,
        media: ClassificationSubject,
        *,
        extensions: Mapping[
            str,
            Mapping[str, ClassificationFactValue],
        ]
        | None = None,
        effective_override: ClassificationSelection | None = None,
        refresh: bool = False,
        allow_enrichment: bool = True,
    ) -> ClassificationSubject:
        """
        分类完整媒体；离线整理可禁止外部事实补充，仍执行当前用户分类策略。

        :param media: 待分类的完整识别结果，返回时与原对象隔离
        :param extensions: 调用方提供的来源扩展事实；None 只使用已注册扩展提供方
        :param effective_override: 显式指定的最终分类；None 时保留媒体已有的显式选择
        :param refresh: 为调用兼容保留的刷新标志；当前实现始终按本次读取的策略重新计算
        :param allow_enrichment: 是否允许外部补充缺失事实；False 时仅使用本地可得事实
        :return: 带本次分类结果的独立媒体副本；策略不可用时保留显式分类或执行兼容回退
        """
        ...

    async def async_finalize(
        self,
        media: ClassificationSubject,
        *,
        extensions: Mapping[
            str,
            Mapping[str, ClassificationFactValue],
        ]
        | None = None,
        effective_override: ClassificationSelection | None = None,
        refresh: bool = False,
        allow_enrichment: bool = True,
    ) -> ClassificationSubject:
        """
        异步分类完整媒体对象，并允许有界补充缺失标准事实。

        :param media: 待分类的完整识别结果，返回时与原对象隔离
        :param extensions: 调用方提供的来源扩展事实；None 只使用已注册扩展提供方
        :param effective_override: 显式指定的最终分类；None 时保留媒体已有的显式选择
        :param refresh: 为调用兼容保留的刷新标志；当前实现始终按本次读取的策略重新计算
        :param allow_enrichment: 是否允许外部补充缺失事实；False 时仅使用本地可得事实
        :return: 带本次分类结果的独立媒体副本；策略不可用时保留显式分类或执行兼容回退
        """
        ...


class ClassificationEnrichmentPort(Protocol):
    """
    分类执行服务消费的同步与异步缺失事实补充端口。
    """

    def enrich(
        self,
        policy: ClassificationPolicy,
        facts: ClassificationFacts,
        media: object,
    ) -> ClassificationFacts:
        """
        同步补充缺失标准事实。

        :param policy: 决定所需字段及补充范围的分类策略
        :param facts: 主来源已确认的标准事实，不得被无依据覆盖
        :param media: 用于提取媒体身份和来源详情的标准对象
        :return: 合并已验证补充信息的事实；无可用补充时保留原有事实
        """
        ...

    async def async_enrich(
        self,
        policy: ClassificationPolicy,
        facts: ClassificationFacts,
        media: object,
    ) -> ClassificationFacts:
        """
        异步补充缺失标准事实。

        :param policy: 决定所需字段及补充范围的分类策略
        :param facts: 主来源已确认的标准事实，不得被无依据覆盖
        :param media: 用于提取媒体身份和来源详情的标准对象
        :return: 合并已验证补充信息的事实；无可用补充时保留原有事实
        """
        ...


class ClassificationExecutionService:
    """
    按当前不可变策略分类完整媒体对象，并隔离来源缓存对象。
    """

    def __init__(
        self,
        runtime: ClassificationRuntimePort,
        *,
        extension_facts_provider: ClassificationExtensionFactsProvider | None = None,
        enrichment: ClassificationEnrichmentPort | None = None,
    ) -> None:
        """
        保存策略运行时、扩展事实校验端口和可选跨来源补充服务。
        """
        self._runtime = runtime
        self._extension_facts_provider = extension_facts_provider
        self._enrichment = enrichment

    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
        if self._enrichment is not None:
            try:
                facts = await self._enrichment.async_enrich(policy, facts, finalized)
            # 详情补充失败时保留主来源事实
            except Exception:  # noqa: BLE001
                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,
        *,
        extensions: Mapping[
            str,
            Mapping[str, ClassificationFactValue],
        ]
        | None = None,
        effective_override: ClassificationSelection | None = None,
        refresh: bool = False,
        allow_enrichment: bool = True,
    ) -> ClassificationSubject:
        """
        复制并分类完整结果；离线调用可禁用外部补充，但仍应用当前策略。
        """
        finalized, policy, facts, effective_override = self._prepare(
            media,
            extensions=extensions,
            effective_override=effective_override,
            refresh=refresh,
            allow_local_music=not allow_enrichment,
        )
        if policy is None or facts is None:
            return finalized
        if allow_enrichment and self._enrichment is not None:
            try:
                facts = self._enrichment.enrich(policy, facts, finalized)
            # 分类补充不得成为识别硬依赖
            except Exception:  # noqa: BLE001
                pass
        return self._apply_evaluation(
            finalized,
            policy,
            facts,
            effective_override=effective_override,
        )

    async def async_finalize(
        self,
        media: ClassificationSubject,
        *,
        extensions: Mapping[
            str,
            Mapping[str, ClassificationFactValue],
        ]
        | None = None,
        effective_override: ClassificationSelection | None = None,
        refresh: bool = False,
        allow_enrichment: bool = True,
    ) -> ClassificationSubject:
        """
        异步分类完整结果，支持只使用本地事实，补充失败时保留原始事实。
        """
        finalized, policy, facts, effective_override = self._prepare(
            media,
            extensions=extensions,
            effective_override=effective_override,
            refresh=refresh,
            allow_local_music=not allow_enrichment,
        )
        if policy is None or facts is None:
            return finalized
        if allow_enrichment and self._enrichment is not None:
            try:
                facts = await self._enrichment.async_enrich(
                    policy,
                    facts,
                    finalized,
                )
            # 分类补充不得成为识别硬依赖
            except Exception:  # noqa: BLE001
                pass
        return self._apply_evaluation(
            finalized,
            policy,
            facts,
            effective_override=effective_override,
        )

    def _prepare(
        self,
        media: ClassificationSubject,
        *,
        extensions: Mapping[str, Mapping[str, ClassificationFactValue]] | None,
        effective_override: ClassificationSelection | None,
        refresh: bool,
        policy_override: ClassificationPolicy | None = None,
        allow_local_music: bool = False,
    ) -> tuple[
        ClassificationSubject,
        ClassificationPolicy | None,
        ClassificationFacts | None,
        ClassificationSelection | None,
    ]:
        """
        复制媒体、读取一次策略并构造同步异步共用的初始事实。
        """
        del refresh
        effective_override = effective_override or _explicit_effective_override(media)
        finalized = deepcopy(media)
        policy = policy_override or self._runtime.active_policy()
        if policy is None:
            self._apply_invalid_policy_fallback(
                finalized,
                effective_override=effective_override,
            )
            return finalized, None, None, effective_override
        registered_extensions = self._registered_extensions(finalized)
        merged_extensions = _classification_extensions(
            policy,
            finalized,
            _merge_supplied_extensions(registered_extensions, extensions),
        )
        try:
            facts = build_classification_facts(
                finalized,
                extensions=merged_extensions,
                allow_local_music=allow_local_music,
            )
        except ValueError:
            finalized.set_library_category("")
            finalized.classification = ClassificationResult(
                effective=(
                    effective_override.model_copy(deep=True)
                    if effective_override
                    else None
                ),
                policy_revision=policy.revision,
                state="not_evaluated",
            )
            if effective_override:
                finalized.set_library_category(
                    _category_path_snapshot(effective_override)
                )
            return finalized, policy, None, effective_override
        return finalized, policy, facts, effective_override

    @staticmethod
    def _apply_evaluation(
        finalized: ClassificationSubject,
        policy: ClassificationPolicy,
        facts: ClassificationFacts,
        *,
        effective_override: ClassificationSelection | None,
    ) -> ClassificationSubject:
        """
        应用纯求值结果、降级保护和人工覆盖，并更新兼容目录分类。
        """
        evaluation = evaluate_classification_facts(policy, facts)
        result = evaluation.result.model_copy(deep=True)
        carried = _degraded_automatic_selection(finalized, policy, result)
        if carried is not None:
            result.effective = carried
        if effective_override:
            result.effective = effective_override.model_copy(deep=True)
        finalized.classification = result
        selection = result.effective or result.recommended
        finalized.set_library_category(
            _category_path_snapshot(selection) if selection else ""
        )
        return finalized

    def _registered_extensions(
        self,
        media: ClassificationSubject,
    ) -> Mapping[str, Mapping[str, ClassificationFactValue]]:
        """
        隔离可选插件事实端口故障，媒体识别结果仍按缺失字段继续分类。
        """
        if self._extension_facts_provider is None:
            return {}
        try:
            return self._extension_facts_provider(media)
        # 扩展事实不得成为识别硬依赖
        except Exception:  # noqa: BLE001
            return {}

    def finalize_many(
        self,
        media_items: Mapping[str, ClassificationSubject],
    ) -> dict[str, ClassificationSubject]:
        """
        逐项分类路径映射，保持键顺序并隔离每个来源结果。
        """
        return {
            key: self.finalize(media)
            for key, media in media_items.items()
        }

    def _apply_invalid_policy_fallback(
        self,
        media: ClassificationSubject,
        *,
        effective_override: ClassificationSelection | None,
    ) -> None:
        """
        策略不可用时仅恢复旧格式 TMDB 目录分类并标记失效状态。
        """
        media.set_library_category("")
        legacy_category = resolve_legacy_tmdb_category(
            self._runtime.legacy_config(),
            media_type=_enum_text(getattr(media, "type", None)),
            media_source=_enum_text(getattr(media, "media_source", None)),
            tmdb_info=getattr(media, "tmdb_info", None),
        )
        selection = (
            ClassificationSelection(
                category_path=[legacy_category],
                source="legacy",
            )
            if legacy_category
            else None
        )
        media.classification = ClassificationResult(
            recommended=selection,
            effective=(
                effective_override.model_copy(deep=True)
                if effective_override
                else selection
            ),
            state="invalid_policy",
        )
        effective = effective_override or selection
        if effective:
            media.set_library_category(_category_path_snapshot(effective))


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.model_copy(deep=True, update={"sources": []})
        for rule in policy.rules
        if rule.id.startswith(_LEGACY_RULE_PREFIX)
    ]
    if not legacy_rules:
        return None
    compatibility_policy = policy.model_copy(
        deep=True,
        update={"rules": legacy_rules},
    )
    compatibility_facts = _legacy_tmdb_compatibility_facts(policy, facts)
    return ClassificationEvaluator.evaluate(
        compatibility_policy,
        compatibility_facts,
        trace=trace,
    )


def _legacy_tmdb_compatibility_facts(
    policy: ClassificationPolicy,
    facts: ClassificationFacts,
) -> ClassificationFacts:
    """
    把跨来源标准字段投影到旧规则的 TMDB 扩展命名空间。
    """
    extensions = {
        str(source): {str(key): value for key, value in values.items()}
        for source, values in facts.extensions.items()
    }
    legacy_info = _legacy_tmdb_info_from_standard_facts(facts)
    for source, values in build_legacy_tmdb_extension_facts(
        policy,
        legacy_info,
    ).items():
        target = extensions.setdefault(source, {})
        for field, value in values.items():
            target.setdefault(field, value)
    return facts.model_copy(
            deep=True,
            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]:
    """
    构造旧规则投影所需的有限 TMDB 字段，不伪造缺失事实。
    """
    media = facts.media
    info: dict[str, object] = {}
    if media.language:
        info["original_language"] = media.language
    if media.countries:
        countries = [str(country) for country in media.countries if country]
        if countries:
            info["origin_country"] = countries
            info["production_countries"] = [
                {"iso_3166_1": country} for country in countries
            ]
    if media.year is not None:
        info["release_date"] = str(media.year)
    for field in (
        "adult",
        "runtime",
        "content_rating",
        "companies",
        "networks",
    ):
        value = getattr(media, field, None)
        if value not in (None, "", []):
            info[field] = value
    return info


def _uses_fallback(result: ClassificationResult) -> bool:
    """
    判断主来源求值是否没有命中具体分类规则。
    """
    selection = result.effective or result.recommended
    return selection is None or selection.source in {None, "fallback", "source_fallback"}


def _classification_extensions(
    policy: ClassificationPolicy,
    media: ClassificationSubject,
    supplied: Mapping[str, Mapping[str, ClassificationFactValue]] | None,
) -> dict[str, dict[str, ClassificationFactValue]]:
    """
    合并受控调用方扩展事实和策略实际需要的旧格式 TMDB 事实。
    """
    merged = {
        str(source): {str(key): value for key, value in values.items()}
        for source, values in (supplied or {}).items()
    }
    tmdb_info = getattr(media, "tmdb_info", None)
    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,
        ).items():
            merged.setdefault(source, {}).update(values)
    return merged


def _merge_supplied_extensions(
    registered: Mapping[str, Mapping[str, ClassificationFactValue]],
    supplied: Mapping[str, Mapping[str, ClassificationFactValue]] | None,
) -> dict[str, dict[str, ClassificationFactValue]]:
    """
    合并已校验插件事实和受信调用方事实，显式调用方值具有最终优先级。
    """
    merged = {
        str(source): {str(key): value for key, value in values.items()}
        for source, values in registered.items()
    }
    for source, values in (supplied or {}).items():
        merged.setdefault(str(source), {}).update(
            {str(key): value for key, value in values.items()}
        )
    return merged


def _explicit_effective_override(
    media: ClassificationSubject,
) -> ClassificationSelection | None:
    """
    保留已装配的人工覆盖，避免补充元数据时被自动推荐覆盖。
    """
    classification = getattr(media, "classification", None)
    if not isinstance(classification, ClassificationResult):
        return None
    effective = classification.effective
    if effective is None or effective.source not in {"manual", "subscription"}:
        return None
    return deepcopy(effective)


def _degraded_automatic_selection(
    media: ClassificationSubject,
    policy: ClassificationPolicy,
    result: ClassificationResult,
) -> ClassificationSelection | None:
    """
    事实缺失把同策略结果打成兜底时，返回可复用的既有自动分类选择。

    补充元数据的入口会先裁剪媒体字段再重新分类，此时缺失事实只会让结果更差，
    不会带来新的判断依据。因此仅在既有快照来自同一策略修订且命中过规则，而本次
    求值同时出现事实缺失和兜底化时才复用；事实读齐后给出的兜底结果仍然生效，
    策略调整和真正的重新分类不会被掩盖。
    """
    if result.state != "partial":
        return None
    proposed = result.effective or result.recommended
    if proposed is not None and proposed.source != "fallback":
        return None
    carried = getattr(media, "classification", None)
    if not isinstance(carried, ClassificationResult):
        return None
    if carried.policy_revision != policy.revision:
        return None
    if carried.state not in {"complete", "partial"}:
        return None
    selection = carried.effective or carried.recommended
    if selection is None or selection.source != "automatic":
        return None
    if not str(selection.category_id or "").strip():
        return None
    return selection.model_copy(deep=True)


def _enum_text(value: object) -> str:
    """
    把字符串枚举或普通值转换为去空白文本。
    """
    if isinstance(value, Enum):
        value = value.value
    return str(value or "").strip()


def _category_path_snapshot(selection: ClassificationSelection) -> str:
    """
    把已校验路径投影为过渡期 library_category 兼容字符串。
    """
    return "/".join(selection.category_path)