MoviePilot/app/application/classification/configuration.py

"""
分类策略初始化、发布、历史和回滚应用服务。
"""

from __future__ import annotations

import threading
from collections.abc import Callable, Iterable
from datetime import datetime, timezone
from functools import partial
from typing import Union, cast

from app.application.classification.contract import (
    ClassificationPolicyConflictError,
    ClassificationPolicyReferenceValidator,
    ClassificationPolicyReferenceViolationError,
    ClassificationPolicyStore,
)
from app.application.database import AsyncDatabaseExecutor
from app.domain.classification.fields import merge_field_definitions
from app.domain.classification.validation import ClassificationPolicyValidator
from app.schemas.category import (
    ClassificationCategory,
    ClassificationCondition,
    ClassificationConditionGroup,
    ClassificationFactScalar,
    ClassificationFieldDefinition,
    ClassificationMediaType,
    ClassificationOperator,
    ClassificationPolicy,
    ClassificationPolicyState,
    ClassificationRule,
    ClassificationTarget,
    ClassificationValidationResult,
)

CLASSIFICATION_POLICY_HISTORY_LIMIT = 10

_DEFAULT_MUSIC_CATEGORIES = (
    ("music.album", "Album", ("Album",)),
    (
        "music.compilation",
        "Album / Compilation",
        ("Album", "Compilation"),
    ),
    ("music.ep", "EP", ("EP",)),
    ("music.single", "Single", ("Single",)),
)

_DEFAULT_MEDIA_CATEGORIES: tuple[tuple[str, ClassificationMediaType, str, tuple[str, ...]], ...] = (
    ("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.uncategorized", "电视剧", "未分类", ("未分类",)),
    ("music.uncategorized", "音乐", "未分类", ("未分类",)),
)


def _build_builtin_media_categories() -> list[ClassificationCategory]:
    """
    构造内置媒体分类目录。
    """
    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
    ]
    return categories


def _build_builtin_movie_rules() -> list[ClassificationRule]:
    """
    构造内置电影分类规则。
    """
    return [
        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="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"),
        ),
    ]


def _build_builtin_tv_rules() -> list[ClassificationRule]:
    """
    构造内置电视剧分类规则。
    """
    return [
        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.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="纪录片",
            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", "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="亚洲剧",
            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"),
        ),
    ]


def build_builtin_classification_policy() -> ClassificationPolicy:
    """
    构造显式区分外语与未知信息的默认模板，不改写已保存的旧策略。
    """
    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=rules,
            fallbacks={
                "电影": "movie.uncategorized",
                "电视剧": "tv.uncategorized",
                "音乐": "music.uncategorized",
            },
        )
    )


class ClassificationPolicyNotInitializedError(RuntimeError):
    """
    表示分类策略服务尚未加载或初始化活动快照。
    """


class ClassificationPolicyRevisionNotFoundError(LookupError):
    """
    表示请求回滚的历史 revision 不在有界历史中。
    """


class ClassificationPolicyValidationError(ValueError):
    """
    表示策略包含阻止发布的结构化校验错误。
    """

    def __init__(self, result: ClassificationValidationResult) -> None:
        """
        保存完整校验结果，供 API 层返回字段级错误。
        """
        self.result = result.model_copy(deep=True)
        super().__init__("分类策略校验失败")


def _build_uncategorized_classification_policy() -> ClassificationPolicy:
    """
    构造旧版仅包含媒体类型兜底分类的初始草稿。
    """
    categories = [
        ClassificationCategory(
            id="movie.uncategorized",
            media_type="电影",
            name="未分类",
            path=["未分类"],
        ),
        ClassificationCategory(
            id="tv.uncategorized",
            media_type="电视剧",
            name="未分类",
            path=["未分类"],
        ),
        ClassificationCategory(
            id="music.uncategorized",
            media_type="音乐",
            name="未分类",
            path=["未分类"],
        ),
    ]
    return ClassificationPolicy(
        categories=categories,
        fallbacks={
            "电影": "movie.uncategorized",
            "电视剧": "tv.uncategorized",
            "音乐": "music.uncategorized",
        },
    )


def with_default_music_classification(policy: ClassificationPolicy) -> ClassificationPolicy:
    """
    为尚未配置音乐分类的策略追加安全、结构化的常用专辑分类。
    """
    if not needs_default_music_classification(policy):
        return policy.model_copy(deep=True)

    categories = [
        *(item.model_copy(deep=True) for item in policy.categories),
        *(
            ClassificationCategory(
                id=category_id,
                media_type="音乐",
                name=name,
                path=list(path),
            )
            for category_id, name, path in _DEFAULT_MUSIC_CATEGORIES
        ),
    ]
    priority = max((item.priority for item in policy.rules), default=-1) + 1
    rules = [*(item.model_copy(deep=True) for item in policy.rules)]

    def append_rule(
        *,
        rule_id: str,
        name: str,
        field: str,
        operator: ClassificationOperator,
        value: Union[str, list[str]],
        category_id: str,
    ) -> None:
        """
        按既有规则之后的顺序追加单个音乐规则。
        """
        nonlocal priority
        rules.append(
            ClassificationRule(
                id=rule_id,
                name=name,
                kind="category",
                priority=priority,
                media_types=["音乐"],
                when=ClassificationCondition(
                    field=field,
                    operator=operator,
                    value=list[ClassificationFactScalar](value) if isinstance(value, list) else value,
                ),
                target=ClassificationTarget(category_id=category_id),
            )
        )
        priority += 1

    # 精选集同时具有 Album 主类型，必须在普通 Album 之前匹配。
    append_rule(
        rule_id="music.compilation.default",
        name="音乐精选集",
        field="music.secondary_types",
        operator="contains_any",
        value=["Compilation"],
        category_id="music.compilation",
    )
    for album_type, suffix, category_id in (
        ("EP", "ep", "music.ep"),
        ("Single", "single", "music.single"),
        ("Album", "album", "music.album"),
    ):
        append_rule(
            rule_id=f"music.{suffix}.default",
            name=f"音乐{album_type}",
            field="music.album_type",
            operator="equals",
            value=album_type,
            category_id=category_id,
        )
    return policy.model_copy(
            deep=True,
            update={"categories": categories, "rules": rules},
        )


def needs_default_music_classification(policy: ClassificationPolicy) -> bool:
    """
    判断音乐侧是否仍为旧版原始兜底，未包含任何用户分类。
    """
    music_rules = [item for item in policy.rules if "音乐" in item.media_types]
    music_categories = [
        item for item in policy.categories if item.media_type == "音乐"
    ]
    if music_rules or len(music_categories) != 1:
        return False
    category = music_categories[0]
    return bool(
        category.id == "music.uncategorized"
        and category.name == "未分类"
        and category.path == ["未分类"]
        and category.enabled
        and not category.labels
        and policy.fallbacks.get("音乐") == category.id
    )


def build_default_classification_policy() -> ClassificationPolicy:
    """
    构造带稳定兜底和常用音乐专辑分类的初始草稿。
    """
    return build_builtin_classification_policy()


class ClassificationPolicyConfigurationService:
    """
    维护分类策略的进程内完整快照和数据库 CAS 发布语义。
    """

    def __init__(
        self,
        store: ClassificationPolicyStore,
        *,
        extra_fields: Iterable[ClassificationFieldDefinition] = (),
        extra_fields_provider: Callable[
            [], Iterable[ClassificationFieldDefinition]
        ] | None = None,
        clock: Callable[[], datetime] | None = None,
        async_executor: AsyncDatabaseExecutor | None = None,
        reference_validator: ClassificationPolicyReferenceValidator | None = None,
    ) -> None:
        """
        注入状态仓储、字段目录、时钟、异步端口和外部引用校验器。
        """
        self._store = store
        self._extra_fields = tuple(
            field.model_copy(deep=True)
            for field in merge_field_definitions(extra_fields)
            if field.id.startswith("extensions.")
        )
        self._extra_fields_provider = extra_fields_provider
        self._clock = clock or (lambda: datetime.now(timezone.utc))
        self._async_executor = async_executor
        self._reference_validator = reference_validator
        self._lock = threading.RLock()
        self._state: ClassificationPolicyState | None = None

    def initialize(
        self,
        initial_policy: ClassificationPolicy | None = None,
    ) -> ClassificationPolicy:
        """
        加载已发布状态；配置不存在时以指定草稿或默认策略发布 revision 1。
        """
        with self._lock:
            stored = self._store.load()
            if stored is not None:
                self._publish_memory_snapshot(stored)
                return self.active()
            draft = initial_policy or build_default_classification_policy()
            return self._publish_locked(
                draft=draft,
                expected_revision=0,
                current_state=None,
            )

    def reload(self) -> ClassificationPolicy:
        """
        从持久化事实源重新加载完整状态包并替换进程内快照。
        """
        with self._lock:
            stored = self._store.load()
            if stored is None:
                raise ClassificationPolicyNotInitializedError("分类策略尚未初始化")
            self._publish_memory_snapshot(stored)
            return self.active()

    def active(self) -> ClassificationPolicy:
        """
        返回与内部活动引用隔离的当前策略副本。
        """
        with self._lock:
            return self._require_state().active.model_copy(deep=True)

    def state(self) -> ClassificationPolicyState:
        """
        返回与内部活动引用隔离的完整状态包副本。
        """
        with self._lock:
            return self._require_state().model_copy(deep=True)

    def history(self) -> tuple[ClassificationPolicy, ...]:
        """
        按 revision 从新到旧返回隔离的有界历史快照。
        """
        with self._lock:
            return tuple(
                policy.model_copy(deep=True)
                for policy in self._require_state().history
            )

    def validate(self, draft: ClassificationPolicy) -> ClassificationValidationResult:
        """
        使用发布时相同字段和外部引用约束校验草稿，但不修改状态。
        """
        with self._lock:
            policy_result = ClassificationPolicyValidator.validate(
                draft,
                extra_fields=self._current_extra_fields(),
            )
            if self._reference_validator is None:
                return policy_result
            reference_result = self._reference_validator.validate(draft)
            issues = [
                *(issue.model_copy(deep=True) for issue in policy_result.issues),
                *(issue.model_copy(deep=True) for issue in reference_result.issues),
            ]
            return ClassificationValidationResult(
                valid=not any(issue.severity == "error" for issue in issues),
                issues=issues,
            )

    def register_extra_fields(
        self,
        extra_fields: Iterable[ClassificationFieldDefinition],
    ) -> tuple[ClassificationFieldDefinition, ...]:
        """
        合并动态来源字段目录并返回隔离后的当前扩展字段快照。
        """
        with self._lock:
            merged = merge_field_definitions((*self._extra_fields, *tuple(extra_fields)))
            self._extra_fields = tuple(
                field.model_copy(deep=True)
                for field in merged
                if field.id.startswith("extensions.")
            )
            return tuple(field.model_copy(deep=True) for field in self._extra_fields)

    def extra_fields(self) -> tuple[ClassificationFieldDefinition, ...]:
        """
        返回与内部字段目录隔离的动态来源字段定义。
        """
        with self._lock:
            return tuple(
                field.model_copy(deep=True)
                for field in self._current_extra_fields()
            )

    def _current_extra_fields(self) -> tuple[ClassificationFieldDefinition, ...]:
        """
        合并静态兼容字段和插件运行时当前快照，不缓存可卸载声明。
        """
        try:
            dynamic = (
                tuple(self._extra_fields_provider())
                if self._extra_fields_provider is not None
                else ()
            )
        # 可选插件目录不可阻断静态策略管理
        except Exception:  # noqa: BLE001
            dynamic = ()
        return tuple(
            field.model_copy(deep=True)
            for field in merge_field_definitions((*self._extra_fields, *dynamic))
            if field.id.startswith("extensions.")
        )

    def publish(
        self,
        draft: ClassificationPolicy,
        *,
        expected_revision: int,
    ) -> ClassificationPolicy:
        """
        校验草稿并以 CAS 发布下一个 revision。
        """
        with self._lock:
            current_state = self._require_state()
            return self._publish_locked(
                draft=draft,
                expected_revision=expected_revision,
                current_state=current_state,
            )

    def rollback(
        self,
        target_revision: int,
        *,
        expected_revision: int,
    ) -> ClassificationPolicy:
        """
        选择历史内容并以新的单调 revision 发布，不复用旧版本号。
        """
        with self._lock:
            current_state = self._require_state()
            target = next(
                (
                    policy
                    for policy in current_state.history
                    if policy.revision == target_revision
                ),
                None,
            )
            if target is None:
                raise ClassificationPolicyRevisionNotFoundError(
                    f"分类策略历史 revision {target_revision} 不存在"
                )
            return self._publish_locked(
                draft=target,
                expected_revision=expected_revision,
                current_state=current_state,
            )

    async def async_initialize(
        self,
        initial_policy: ClassificationPolicy | None = None,
    ) -> ClassificationPolicy:
        """
        在线程执行端口中初始化分类策略，避免阻塞异步启动流程。
        """
        return await self._run_async(partial(self.initialize, initial_policy))

    async def async_reload(self) -> ClassificationPolicy:
        """
        在线程执行端口中重新加载分类策略。
        """
        return await self._run_async(self.reload)

    async def async_publish(
        self,
        draft: ClassificationPolicy,
        *,
        expected_revision: int,
    ) -> ClassificationPolicy:
        """
        在线程执行端口中校验并原子发布分类策略。
        """
        return await self._run_async(
            partial(self.publish, draft, expected_revision=expected_revision)
        )

    async def async_rollback(
        self,
        target_revision: int,
        *,
        expected_revision: int,
    ) -> ClassificationPolicy:
        """
        在线程执行端口中把历史内容发布为新的 revision。
        """
        return await self._run_async(
            partial(
                self.rollback,
                target_revision,
                expected_revision=expected_revision,
            )
        )

    def _publish_locked(
        self,
        *,
        draft: ClassificationPolicy,
        expected_revision: int,
        current_state: ClassificationPolicyState | None,
    ) -> ClassificationPolicy:
        """
        在进程锁内构造、校验、CAS 持久化并发布完整状态包。
        """
        current_revision = current_state.active.revision if current_state else 0
        if expected_revision != current_revision:
            raise ClassificationPolicyConflictError(
                expected_revision=expected_revision,
                current_revision=current_revision,
            )
        candidate = draft.model_copy(
            deep=True,
            update={
                "revision": current_revision + 1,
                "updated_at": self._clock(),
            },
        )
        validation = self.validate(candidate)
        if not validation.valid:
            raise ClassificationPolicyValidationError(validation)
        history = [] if current_state is None else [
            current_state.active.model_copy(deep=True),
            *(policy.model_copy(deep=True) for policy in current_state.history),
        ]
        desired = ClassificationPolicyState(
            active=candidate,
            history=history[:CLASSIFICATION_POLICY_HISTORY_LIMIT],
        )
        try:
            self._store.compare_and_set(
                expected_revision=expected_revision,
                state=desired,
            )
        except ClassificationPolicyReferenceViolationError as error:
            raise ClassificationPolicyValidationError(error.result) from error
        except Exception:
            stored = self._store.load()
            if stored is not None:
                self._publish_memory_snapshot(stored)
            raise
        self._publish_memory_snapshot(desired)
        return desired.active.model_copy(deep=True)

    def _publish_memory_snapshot(self, state: ClassificationPolicyState) -> None:
        """
        以深拷贝整体替换内部快照，禁止暴露持久化对象引用。
        """
        self._state = state.model_copy(deep=True)

    def _require_state(self) -> ClassificationPolicyState:
        """
        返回内部活动状态；未初始化时明确失败。
        """
        if self._state is None:
            raise ClassificationPolicyNotInitializedError("分类策略尚未初始化")
        return self._state

    async def _run_async(
        self,
        operation: Callable[[], ClassificationPolicy],
    ) -> ClassificationPolicy:
        """
        通过注入的数据库执行端口运行一个同步短事务操作。
        """
        if self._async_executor is None:
            raise RuntimeError("分类策略异步数据库执行端口尚未配置")
        return cast(
            ClassificationPolicy,
            await self._async_executor.run(operation),
        )