MoviePilot/app/chain/transfer/execution.py

"""
持久整理步骤执行、探测与检查点推进。
"""

import time
import traceback
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Mapping, Optional, Tuple, Union, cast

from app.application.audio import use_music_cue
from app.application.directory import DirectoryHelper
from app.application.history.retry import (
    max_failed_retries,
)
from app.application.transfer.execution import (
    TransferExecutionCheckpoint,
    TransferExecutionCommand,
    TransferExecutionConflictError,
    TransferExecutionLeaseLostError,
    TransferExecutionRepository,
    TransferExecutionSnapshot,
    TransferOperationObservation,
    TransferOperationObservationState,
    TransferStepIntent,
    TransferStepResult,
    TransferStepState,
)
from app.application.transfer.workflow import (
    TransferAdmission,
    TransferAdmissionConflictError,
    TransferLeaseLostError,
    TransferTask,
)
from app.chain.media import MediaChain
from app.chain.tmdb import TmdbChain
from app.chain.transfer.contract import _TransferOwnerBase
from app.chain.transfer.music import defer_music_recognition, refresh_music_retry_context
from app.chain.transfer.records import apply_download_history_classification
from app.domain.context import MediaInfo, MusicInfo
from app.domain.media import is_music_media_source
from app.domain.meta.metamusic import MetaMusic
from app.runtime.log import logger
from app.schemas.transfer import TransferInfo
from app.schemas.types import (
    MediaSource,
    MediaType,
)


class _TransferRetryDeferred(RuntimeError):
    """
    表示已把确定失败交回唯一持久重试调度器。
    """


class _TransferManualReviewRequired(RuntimeError):
    """
    表示外部结果无法严格判定，任务已隔离等待人工复核。
    """


class _TransferRetryExhausted(RuntimeError):
    """
    表示步骤重试预算耗尽，任务可进入原子失败结算。
    """

    def __init__(self, message: str, snapshot: TransferExecutionSnapshot) -> None:
        """
        保存失败原因和数据库建立的结算检查点。
        """
        super().__init__(message)
        self.snapshot = snapshot


def _download_history_media_type(
        history_type: Optional[str],
        fallback: Optional[MediaType],
) -> Optional[MediaType]:
    """
    使用下载历史身份类型；缺少有效类型时回退到整理调用提示。
    """
    try:
        media_type = MediaType(history_type)
    except (TypeError, ValueError):
        return fallback
    return fallback if media_type == MediaType.UNKNOWN else media_type


class _DurableTransferStepRunner:
    """
    以稳定顺序、lease 和 attempt fencing 执行整理外部步骤。
    """

    _RETRY_DELAY_SECONDS = 30

    def __init__(
            self,
            *,
            task_id: str,
            lease_token: str,
            checkpoint_fingerprint: str,
            repository: TransferExecutionRepository,
    ) -> None:
        """
        绑定单个任务的持久执行命令与全局步骤序号。
        """
        self._task_id = task_id
        self._lease_token = lease_token
        self._checkpoint_fingerprint = checkpoint_fingerprint
        self._repository = repository
        self._command = TransferExecutionCommand(self._repository)
        snapshot = self._repository.get_snapshot(task_id=task_id)
        existing_steps = list(snapshot.steps) if snapshot else []
        matching_ordinals = [
            step.ordinal
            for step in existing_steps
            if step.checkpoint_fingerprint == checkpoint_fingerprint
        ]
        self._ordinal = (
            min(matching_ordinals)
            if matching_ordinals
            else max((step.ordinal for step in existing_steps), default=-1) + 1
        )
        self._operation_ids = [
            step.operation_id
            for step in existing_steps
            if step.ordinal < self._ordinal
        ]

    @staticmethod
    def _retry_due_at() -> str:
        """
        生成固定退避后的 UTC 调度时间。
        """
        return (
            datetime.now(timezone.utc)
            + timedelta(seconds=_DurableTransferStepRunner._RETRY_DELAY_SECONDS)
        ).strftime("%Y-%m-%d %H:%M:%S.%f")

    def _execute_attempt(
            self,
            *,
            step: Any,
            execute: Callable[[], TransferStepResult],
            observe: Callable[[], TransferOperationObservation],
    ) -> TransferStepResult:
        """
        在事务外执行副作用，异常后先探测结果再决定重试或人工复核。
        """
        try:
            result = execute()
        except Exception as error:
            try:
                observation = observe()
            except Exception as observe_error:
                observation = TransferOperationObservation(
                    state=TransferOperationObservationState.UNKNOWN,
                    evidence=TransferStepResult(payload={
                        "execute_error": str(error),
                        "observe_error": str(observe_error),
                    }),
                )
            if observation.state is TransferOperationObservationState.APPLIED:
                completed = self._command.complete(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    result=observation.evidence,
                )
                if completed.result is None:
                    raise RuntimeError("探测完成的整理步骤缺少结果证据") from error
                return completed.result
            if observation.state is not TransferOperationObservationState.NOT_APPLIED:
                reason = (
                    f"整理步骤 {step.operation_id} 执行异常后外部结果为 "
                    f"{observation.state.value}，禁止自动重放：{error}"
                )
                self._command.manual_review(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    error=reason,
                    evidence=observation.evidence,
                )
                raise _TransferManualReviewRequired(reason) from error
            snapshot = self._repository.get_snapshot(task_id=self._task_id)
            if snapshot is None:
                raise RuntimeError("整理执行快照在步骤失败后丢失") from error
            if snapshot.retry_count >= max_failed_retries():
                exhausted = self._command.exhaust(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    error=str(error),
                    evidence=observation.evidence,
                )
                raise _TransferRetryExhausted(str(error), exhausted) from error
            self._command.defer(
                task_id=self._task_id,
                lease_token=self._lease_token,
                step=step,
                error=str(error),
                retry_due_at=self._retry_due_at(),
                evidence=observation.evidence,
            )
            raise _TransferRetryDeferred(str(error)) from error
        completed = self._command.complete(
            task_id=self._task_id,
            lease_token=self._lease_token,
            step=step,
            result=result,
        )
        if completed.result is None:
            raise RuntimeError("整理步骤完成后缺少持久结果")
        return completed.result

    def run(
            self,
            *,
            phase: str,
            kind: str,
            payload: Mapping[str, Any],
            execute: Callable[[], TransferStepResult],
            observe: Callable[[], TransferOperationObservation],
    ) -> TransferStepResult:
        """
        幂等消费一个步骤，遗留 STARTED 必须先取得严格观察结论。
        """
        intent = TransferStepIntent.create(
            task_id=self._task_id,
            checkpoint_fingerprint=self._checkpoint_fingerprint,
            ordinal=self._ordinal,
            phase=phase,
            kind=kind,
            payload=payload,
        )
        self._ordinal += 1
        self._operation_ids.append(intent.operation_id)
        step = self._command.prepare(
            task_id=self._task_id,
            lease_token=self._lease_token,
            intent=intent,
        )
        if step.state is TransferStepState.SUCCEEDED:
            if step.result is None:
                raise RuntimeError("已成功整理步骤缺少结果证据")
            return step.result
        if step.state is TransferStepState.MANUAL_REVIEW:
            raise _TransferManualReviewRequired(
                step.last_error or "整理步骤正在等待人工复核"
            )
        if step.state is TransferStepState.PREPARED:
            step = self._command.begin(
                task_id=self._task_id,
                lease_token=self._lease_token,
                operation_id=step.operation_id,
            )
        elif step.state is TransferStepState.FAILED:
            step = self._command.resume_failed(
                task_id=self._task_id,
                lease_token=self._lease_token,
                step=step,
            )
        elif step.state is TransferStepState.STARTED:
            observation = observe()
            if observation.state is TransferOperationObservationState.APPLIED:
                completed = self._command.complete(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    result=observation.evidence,
                )
                if completed.result is None:
                    raise RuntimeError("探测完成的整理步骤缺少结果证据")
                return completed.result
            if observation.state is TransferOperationObservationState.NOT_APPLIED:
                step = self._command.restart_after_not_applied(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    evidence=observation.evidence,
                )
            else:
                reason = (
                    f"遗留步骤 {step.operation_id} 外部结果为 "
                    f"{observation.state.value}，禁止自动重放"
                )
                self._command.manual_review(
                    task_id=self._task_id,
                    lease_token=self._lease_token,
                    step=step,
                    error=reason,
                    evidence=observation.evidence,
                )
                raise _TransferManualReviewRequired(reason)
        else:
            raise RuntimeError(f"不支持的整理步骤状态：{step.state}")
        return self._execute_attempt(step=step, execute=execute, observe=observe)

    def checkpoint(self, transferinfo: TransferInfo) -> TransferExecutionCheckpoint:
        """
        提交足以独立重放终态结算的聚合执行结果。
        """
        outcome = (
            "overwrite_skipped"
            if transferinfo.overwrite_skipped
            else "succeeded" if transferinfo.success else "failed"
        )
        checkpoint = TransferExecutionCheckpoint.create(
            payload={
                "outcome": outcome,
                "transferinfo": transferinfo.model_dump(mode="json"),
            },
            operation_ids=tuple(self._operation_ids),
            skip_reason=(
                (transferinfo.message or "zero-side-effect transfer")
                if not self._operation_ids
                else None
            ),
        )
        snapshot = self._command.checkpoint(
            task_id=self._task_id,
            lease_token=self._lease_token,
            checkpoint=checkpoint,
        )
        if snapshot.checkpoint is None:
            raise RuntimeError("整理执行检查点提交后无法回读")
        return snapshot.checkpoint


class TransferExecutionOwner(_TransferOwnerBase):
    """
    持有整理准入租约、外部步骤执行与结果检查点。
    """

    def _TransferChain__handle_transfer_execution_error(
            self,
            task: TransferTask,
            error: Exception,
            *,
            preview: bool,
    ) -> Tuple[bool, str]:
        """
        在插件可替换执行入口之外收口异常，防止未记账失败无限回放。
        """
        if (not preview and not task.terminal_settled and task.admission_task_id and task.lease_token
                and not isinstance(error, (TransferAdmissionConflictError, TransferLeaseLostError,
                                           TransferExecutionConflictError, TransferExecutionLeaseLostError,
                                           _TransferRetryDeferred, _TransferRetryExhausted,
                                           _TransferManualReviewRequired))):
            try:
                isolated = self.transfer_execution_repository.isolate_unhandled_failure(
                    task_id=task.admission_task_id, lease_token=task.lease_token, error=str(error),
                )
                if isolated:
                    error = _TransferManualReviewRequired(str(error))
            except Exception as record_error:
                logger.error(f"记录整理执行异常失败：{record_error}")
        if isinstance(error, _TransferManualReviewRequired):
            logger.warning(
                f"{task.fileitem.name} 已转入人工复核，未自动重放；"
                f"请打开整理队列查看详情，确认外部操作状态后再决定是否重试：{error}"
            )
            message = "整理任务已转入人工复核，请打开整理队列查看详情"
        else:
            logger.error(
                f"{task.fileitem.name} 整理任务处理出现错误：{error} - {traceback.format_exc()}"
            )
            message = (
                str(error) if isinstance(error, TransferAdmissionConflictError)
                else "整理任务处理失败，请稍后重试"
            )
        if not preview:
            self._TransferChain__fail_transfer_task(task, error)
        return False, message

    def _TransferChain__register_claimed_admission(
            self,
            admission: TransferAdmission,
    ) -> None:
        """
        把仓储 claim 加入续期集合，覆盖恢复构造阶段的同步 I/O。
        """
        if (
                admission.lease_owner != self._worker_owner_id
                or not admission.lease_token
        ):
            raise TransferLeaseLostError(
                f"整理任务 claim 投影无效：{admission.task_id}"
            )
        with self._worker_state_lock:
            self._owned_leases[admission.task_id] = (
                admission.lease_token,
                time.monotonic() + self._WORKER_LEASE_SECONDS,
            )
        self._TransferChain__ensure_lease_heartbeat_owner()

    def _TransferChain__bind_claimed_admission(
            self,
            task: TransferTask,
            admission: TransferAdmission,
    ) -> None:
        """
        校验仓储 claim 投影并把 token 私有绑定到执行任务。
        """
        if admission.task_id != task.admission_task_id:
            raise TransferLeaseLostError(
                f"整理任务 claim 投影无效：{admission.task_id}"
            )
        self._TransferChain__register_claimed_admission(admission)
        assert admission.lease_owner is not None
        assert admission.lease_token is not None
        task.bind_execution_lease(
            owner_id=admission.lease_owner,
            lease_token=admission.lease_token,
        )

    def _TransferChain__claim_admitted_task(
            self,
            task: TransferTask,
            task_id: str,
    ) -> None:
        """
        取得已准入任务的唯一租约，并绑定到内存任务。
        """
        claimed = cast(
            Optional[TransferAdmission],
            self._transfer_admissions.claim_task(
                task_id=task_id,
                owner_id=self._worker_owner_id,
                lease_seconds=self._WORKER_LEASE_SECONDS,
            ),
        )
        if claimed is None:
            raise TransferLeaseLostError(f"整理任务已由其他 worker claim：{task_id}")
        try:
            self._TransferChain__bind_claimed_admission(task, claimed)
        except Exception as err:
            self._TransferChain__release_admission_claim(claimed, error=str(err))
            raise

    def _TransferChain__admit_transfer(self, task: TransferTask) -> TransferAdmission:
        """
        持久化源文件并在内存入队前取得执行租约。
        """
        fileitem = task.fileitem if task else None
        if not fileitem or not fileitem.storage or not fileitem.path:
            raise ValueError("整理任务缺少源文件身份")
        planning_input = task.planning_input or self._TransferChain__build_planning_input(task)
        task.bind_planning_input(planning_input)
        admission = cast(
            TransferAdmission,
            self._transfer_admissions.admit(
                storage=fileitem.storage,
                src_path=fileitem.path,
                planning_input=planning_input,
                replace_inactive=bool(task.manual),
            ),
        )
        task.bind_admission_task_id(admission.task_id)
        self._TransferChain__claim_admitted_task(task, admission.task_id)
        return admission

    def _TransferChain__claim_task_for_execution(self, task: TransferTask) -> None:
        """
        校验队列任务既有租约，并为兼容旧队列项补取唯一租约。
        """
        if task.preview:
            return
        self._TransferChain__ensure_lease_runtime_state()
        if task.lease_token:
            self._TransferChain__assert_owned_lease(task)
            return
        if not task.admission_task_id:
            self._TransferChain__admit_transfer(task)
            return
        self._TransferChain__claim_admitted_task(task, task.admission_task_id)

    def _TransferChain__handle_transfer(
            self,
            task: TransferTask,
            callback: Optional[Callable[[TransferTask, TransferInfo], Tuple[bool, str]]] = None,
    ) -> Optional[Tuple[bool, str]]:
        """
        恢复准入时的 CUE 策略并执行，统一记录 checkpoint 前的失败或异常。
        """
        try:
            cue_enabled = cast(Optional[bool], task.planning_input.options.get("music_cue_enable")) if task.planning_input else None
            with use_music_cue(cue_enabled):
                result = self._TransferChain__perform_transfer(task, callback)
        except Exception as error:
            if not getattr(error, "_transfer_planning_failure_recorded", False):
                self._TransferChain__record_uncheckpointed_failure(task, error)
            raise
        if result and not result[0]:
            self._TransferChain__record_uncheckpointed_failure(task, result[1])
        return result

    @staticmethod
    def _recognize_task_meta(task: TransferTask, *, restore_history: bool = True) -> Optional[Union[MediaInfo, MusicInfo]]:
        """
        按任务限定的来源和类型识别元数据，仅在历史语义兼容时恢复其分类。
        """
        options: dict[str, Any] = {"obtain_images": True}
        if task.media_source:
            options["media_source"] = task.media_source
        if task.mtype:
            options["mtype"] = task.mtype
        info = MediaChain().recognize_by_meta(task.meta, **options)
        if info and restore_history and task.download_history:
            info = apply_download_history_classification(info, task.download_history)
        return info

    def _TransferChain__perform_transfer(
            self,
            task: TransferTask,
            callback: Optional[Callable[[TransferTask, TransferInfo], Tuple[bool, str]]] = None,
    ) -> Optional[Tuple[bool, str]]:
        """
        执行冻结计划，或在元数据完整时识别并规划新任务。

        未规划任务缺少文件名元数据时走可结算的规划拒绝，不猜测季集或执行文件操作。
        """
        try:
            if task.plan_checkpoint is not None:
                return cast(
                    Optional[Tuple[bool, str]],
                    self._TransferChain__handle_planned_transfer(task, callback),
                )
            refresh_music_retry_context(self, task)
            if task.meta is None:
                transferinfo = self._TransferChain__checkpoint_planning_rejection(
                    task,
                    "整理规划输入缺少文件名元数据，请重新预览并确认季集信息后重新整理",
                )
                return callback(task, transferinfo) if callback else (
                    transferinfo.success, transferinfo.message or ""
                )
            transferhis = self.transfer_history_repository
            # 显式标注联合：下面既会赋回音乐识别结果（MusicInfo），也会赋回影视识别
            # 结果（MediaInfo），不标注时会被推断成其中一种，另一种就成了假错误
            mediainfo: Optional[Union[MediaInfo, MusicInfo]] = task.mediainfo
            mediainfo_changed = False
            need_obtain_images = False
            if not mediainfo:
                download_history = task.download_history
                if download_history:
                    task.username = download_history.username
                    history_year_conflict = self._is_movie_year_conflict(
                        task.meta, download_history
                    )
                    # 本次明确按影视整理时，旧音乐身份不能抢占随片音轨的归属。
                    history_music_conflict = task.mtype in (MediaType.MOVIE, MediaType.TV) and (
                        download_history.type == MediaType.MUSIC.value
                        or is_music_media_source(download_history.media_source)
                        or self._download_history_music_type(download_history) is not None
                    )
                    if (
                            download_history.media_source
                            and download_history.media_id
                            and not history_year_conflict
                            and not history_music_conflict
                    ):
                        # 下载记录中已存在识别信息。这里不再重复标注类型：函数开头
                        # 已把 mediainfo 声明为 MediaInfo | MusicInfo | None，重复
                        # 声明会遮蔽它，把音乐识别结果判成类型错误
                        mediainfo = MediaChain().recognize_media(
                            mtype=_download_history_media_type(download_history.type, task.mtype),
                            media_source=download_history.media_source,
                            media_id=download_history.media_id,
                            music_type=self._download_history_music_type(download_history),
                            episode_group=download_history.episode_group,
                        )
                        need_obtain_images = True
                        if mediainfo:
                            mediainfo = apply_download_history_classification(mediainfo, download_history)
                    else:
                        if history_year_conflict:
                            logger.info(
                                f"{task.fileitem.name} 文件年份 {task.meta.year if task.meta else None} 与下载记录年份 "
                                f"{download_history.year} 不一致，按文件名重新识别"
                            )
                        mediainfo = self._recognize_task_meta(task, restore_history=not history_music_conflict)
                else:
                    mediainfo = self._recognize_task_meta(task)

                # 音乐必须先经过音乐元数据模块识别；远端不可用时再保留本地标签结果，
                # 避免因离线兜底提前赋值而跳过音乐识别链。
                if not mediainfo and isinstance(task.meta, MetaMusic):
                    mediainfo = self._music_info_from_meta(task.meta)

                # 按名称识别时已在识别链路补图，这里只补齐显式ID识别的场景。
                if mediainfo and need_obtain_images:
                    self.obtain_images(mediainfo=mediainfo)

                if mediainfo and task.media_source:
                    mediainfo.scrape_source = task.media_source

                if not mediainfo:
                    if task.preview:
                        return False, "未识别到媒体信息"
                    transferinfo = self._TransferChain__checkpoint_planning_rejection(
                        task,
                        "未识别到媒体信息",
                    )
                    if callback:
                        return callback(task, transferinfo)
                    return transferinfo.success, transferinfo.message or ""

                mediainfo_changed = True

            # accepted/planned 恢复使用持久快照；在线补充会让首次执行与重放产生漂移。
            if not task.planning_context_restored:
                supplemented = MediaChain().supplement_tmdb_info(mediainfo, task.meta)
                mediainfo = cast(Union[MediaInfo, MusicInfo], supplemented)
            task.mediainfo = mediainfo
            if deferred := defer_music_recognition(self, task):
                return deferred

            # 只有 TMDB 主源沿用历史 TMDB 标题，避免辅助 ID 改写其它识别源标题。
            if (
                    not self.runtime_config.scrape_follow_tmdb
                    and mediainfo.media_source == MediaSource.TMDB
            ):
                transfer_history = transferhis.get_by_media_identity(
                    media_source=mediainfo.media_source.value,
                    media_id=mediainfo.media_id,
                    mtype=cast(MediaType, mediainfo.type).value,
                )
                if transfer_history and mediainfo.title != transfer_history.title:
                    mediainfo.title = transfer_history.title
                    mediainfo_changed = True

            if mediainfo_changed:
                task.mediainfo = mediainfo
                if not self.jobview.migrate_task(task):
                    logger.info(f"{task.fileitem.name} 已存在整理任务，跳过重复处理")
                    return False, f"{task.fileitem.name} 已在整理队列中"

            if (
                    mediainfo.type == MediaType.TV
                    and mediainfo.tmdb_id
                    and not task.episodes_info
                    and not task.planning_context_restored
            ):
                # 第零季表示特别篇，不能按空值回退为第一季。
                season_num = cast(MediaInfo, mediainfo).season
                if season_num is None:
                    season_seq = task.meta.season_seq
                    season_num = int(season_seq) if season_seq.isdigit() else 1
                task.episodes_info = TmdbChain().tmdb_episodes(
                    tmdbid=mediainfo.tmdb_id,
                    season=season_num,
                    episode_group=mediainfo.episode_group,
                )

            if not task.target_directory:
                if task.target_path:
                    # 指定目标路径，`手动整理`场景下使用，忽略源目录匹配，使用指定目录匹配
                    task.target_directory = DirectoryHelper().get_dir(
                        media=task.mediainfo,
                        dest_path=task.target_path,
                        target_storage=task.target_storage,
                    )
                else:
                    # 启用源目录匹配时，根据源目录匹配下载目录，否则按源目录同盘优先原则，如无源目录，则根据媒体信息获取目标目录
                    task.target_directory = DirectoryHelper().get_dir(
                        media=task.mediainfo,
                        storage=task.fileitem.storage,
                        src_path=Path(cast(str, task.fileitem.path)),
                        target_storage=task.target_storage,
                    )
            if not task.target_storage and task.target_directory:
                task.target_storage = task.target_directory.library_storage

            if error_message := self._transfer_validation_error(task):
                logger.error(f"{task.fileitem.name} {error_message}")
                if task.preview:
                    return False, error_message
                transferinfo = self._TransferChain__checkpoint_planning_rejection(
                    task,
                    error_message,
                )
                if callback:
                    return callback(task, transferinfo)
                return transferinfo.success, transferinfo.message or ""

            self.jobview.running_task(task)

            # 广播事件，请示额外的源、目标存储支持。
            source_oper = self._TransferChain__select_storage_oper(task.fileitem.storage)
            target_oper = self._TransferChain__select_storage_oper(task.target_storage)

            # 纯规划先提交 持久执行检查点，任何文件副作用只能发生在提交之后。
            transferinfo = self._plan_checkpoint_and_execute(
                task,
                source_oper=source_oper,
                target_oper=target_oper,
            )
            if not transferinfo:
                logger.error("文件整理模块运行失败")
                return False, "文件整理模块运行失败"

            # 回调，位置传参：任务、整理结果
            if callback:
                return callback(task, transferinfo)

            return transferinfo.success, transferinfo.message

        finally:
            self.jobview.try_remove_job(task)
            self._finish_scrape_batch_task(task)