MoviePilot/app/api/endpoints/transfer.py

"""
提供文件整理任务查询、执行、重试和控制接口。
"""

import re
from pathlib import Path
from typing import Annotated, Any, List, Literal, Optional

from fastapi import Depends, HTTPException, Query, status

from app.adapters.web.security.access import verify_apitoken, verify_token
from app.api.dependencies.auth import get_current_active_manage_user
from app.api.dependencies.history import get_transfer_execution_repository, get_transfer_history_lookup_service
from app.api.endpoints.transferhistory import (
    restore_manual_transfer_history_batch,
    restore_manual_transfer_history_metadata,
)
from app.api.response import (
    COLLECTION_MAX_PAGE_SIZE,
    CompatibleCountParam,
    CompatiblePageParam,
    ResponseAPIRouter,
)
from app.application.configuration import get_api_runtime_config_snapshot
from app.application.directory import DirectoryHelper
from app.application.history import TransferHistoryLookupService
from app.application.transfer.execution import (
    TransferExecutionCommand,
    TransferExecutionConflictError,
    TransferExecutionRepository,
    TransferExecutionState,
    TransferManualReviewDecision,
    TransferManualReviewQuery,
    TransferManualReviewTaskView,
    TransferStepResult,
)
from app.chain.media import MediaChain
from app.chain.transfer.facade import TransferChain
from app.runtime.log import logger
from app.runtime.stop import runtime_stop_state
from app.schemas.common import NameData as _SchemaNameData
from app.schemas.file import FileItem
from app.schemas.file import FileItem as _SchemaFileItem
from app.schemas.response import Response as _SchemaResponse
from app.schemas.system import TransferDirectoryConf as _SchemaTransferDirectoryConf
from app.schemas.token import TokenPayload as _SchemaTokenPayload
from app.schemas.transfer import EpisodeFormat as _SchemaEpisodeFormat
from app.schemas.transfer import EpisodeFormatRecommendData as _SchemaEpisodeFormatRecommendData
from app.schemas.transfer import EpisodeFormatRecommendItem, ManualTransferItem
from app.schemas.transfer import ManualTransferHistoryInfo as _SchemaManualTransferHistoryInfo
from app.schemas.transfer import ManualTransferResultData as _SchemaManualTransferResultData
from app.schemas.transfer import ManualTransferSubmissionData as _SchemaManualTransferSubmissionData
from app.schemas.transfer import ManualTransferSubmissionItem as _SchemaManualTransferSubmissionItem
from app.schemas.transfer import ManualTransferTargetPath as _SchemaManualTransferTargetPath
from app.schemas.transfer import TransferJob as _SchemaTransferJob
from app.schemas.transfer import TransferManualReviewData as _SchemaTransferManualReviewData
from app.schemas.transfer import TransferManualReviewDiscardData as _SchemaTransferManualReviewDiscardData
from app.schemas.transfer import TransferManualReviewDiscardRequest as _SchemaTransferManualReviewDiscardRequest
from app.schemas.transfer import TransferManualReviewPageData as _SchemaTransferManualReviewPageData
from app.schemas.transfer import TransferManualReviewRequest as _SchemaTransferManualReviewRequest
from app.schemas.transfer import TransferManualReviewTaskData as _SchemaTransferManualReviewTaskData
from app.schemas.transfer import TransferQueuePageData as _SchemaTransferQueuePageData
from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaType

router = ResponseAPIRouter()


def _public_transfer_message(message: Optional[object]) -> Optional[str]:
    """
    把整理链返回的错误转换为前端可直接展示的文案。
    """
    if message is None or not str(message).strip():
        return None
    from app.runtime.errors import public_error_message

    return public_error_message(message, context="transfer")


def _public_transfer_result(data: dict[str, Any]) -> dict[str, Any]:
    """
    裁剪整理结果中的错误字段，保留预览数据的原有结构。
    """
    from app.application.transfer.feedback import classify_transfer_failure

    result = dict(data)
    if result.get("message"):
        result["message"] = _public_transfer_message(result["message"])
    items = result.get("items")
    if isinstance(items, list):
        public_items = []
        for item in items:
            if not isinstance(item, dict):
                public_items.append(item)
                continue
            public_item = {
                **item,
                "message": _public_transfer_message(item.get("message")),
            }
            if public_item.get("success") is False:
                feedback = classify_transfer_failure(
                    public_item.get("message"),
                    overwrite_skipped=bool(public_item.get("overwrite_skipped")),
                )
                public_item["failure_stage"] = public_item.get("failure_stage") or feedback.stage.value
                public_item["recovery_action"] = public_item.get("recovery_action") or feedback.action
            public_items.append(public_item)
        result["items"] = public_items
    return result


def _build_failure_preview_item(
        file_item: FileItem, message: Optional[str]
) -> dict[str, Any]:
    """
    构造手动整理预览失败项。
    """
    from app.application.transfer.feedback import classify_transfer_failure

    feedback = classify_transfer_failure(message)
    return {
        "source": file_item.path if file_item else None,
        "source_storage": file_item.storage if file_item else None,
        "source_item": file_item.model_dump(exclude={"children", "url", "thumbnail"}) if file_item else None,
        "target": None,
        "target_dir": None,
        "success": False,
        "message": _public_transfer_message(message),
        "failure_stage": feedback.stage.value,
        "recovery_action": feedback.action,
        "overwrite_skipped": False,
        "type": None,
        "title": None,
        "season": None,
        "episode": None,
        "episode_end": None,
        "part": None,
        "org_string": None,
        "apply_words": [],
        "resource_team": None,
        "customization": None,
    }


def _restore_history_fileitem(payload: object) -> Optional[FileItem]:
    """
    从整理历史的 JSON 快照恢复文件项，拒绝缺失或非对象载荷。
    """
    if not isinstance(payload, dict):
        return None
    return FileItem.model_validate(payload)


def _format_manual_transfer_failure(
    *,
    message: Optional[object],
    source_path: Optional[str],
    target_path: Optional[str] = None,
) -> str:
    """
    把手动整理失败转换为包含路径、阶段和下一步动作的用户提示。
    """
    from app.application.transfer.feedback import format_transfer_failure_message

    public_message = _public_transfer_message(message) or "整理失败"
    return format_transfer_failure_message(
        public_message,
        source_path=source_path,
        target_path=target_path,
    )


def _merge_transfer_messages(messages: List[str]) -> str:
    """
    合并手动整理批量预览提示信息，并统一转换错误文案。
    """
    valid_messages = [
        public_message
        for msg in messages
        if msg
        for public_message in [_public_transfer_message(msg)]
        if public_message
    ]
    if not valid_messages:
        return ""
    return "、".join(valid_messages[:2]) + (
        f"，等{len(valid_messages)}条消息" if len(valid_messages) > 2 else ""
    )


def _manual_submission_items(
    fileitem: FileItem, success: bool, result: object, target_path: Optional[Path],
) -> list[_SchemaManualTransferSubmissionItem]:
    """
    映射真实执行回执，旧返回值最多确认接收，不能推断入库完成。
    """
    if isinstance(result, dict):
        items = _public_transfer_result(result).get("items")
        if isinstance(items, list):
            return [_SchemaManualTransferSubmissionItem.model_validate(item) for item in items]
    message = _public_transfer_message(result)
    data = _public_transfer_result({"items": [{
        "source": fileitem.path, "success": success,
        "state": "accepted" if success else "failed", "message": message,
        "target_dir": target_path.as_posix() if target_path else None,
    }]})
    return [_SchemaManualTransferSubmissionItem.model_validate(item) for item in data["items"]]


def _manual_submission_response(
    success: bool, items: list[_SchemaManualTransferSubmissionItem],
) -> Any:
    """
    保留部分成功文件，并继续为旧客户端提供包含失败路径的总提示。
    """
    errors = [
        _format_manual_transfer_failure(
            message=item.message, source_path=item.source,
            target_path=item.target or item.target_dir,
        )
        for item in items if item.state in {"failed", "retry_wait", "manual_review"} and item.message
    ]
    message = _merge_transfer_messages(errors or [item.message for item in items if item.message and not success])
    return _SchemaResponse(
        success=success, message=message,
        data=_SchemaManualTransferSubmissionData(items=items, message=message),
    )


def _is_music_file_batch(fileitems: List[FileItem]) -> bool:
    """
    判断显式选中项是否为可按专辑识别的多音轨批次。
    """
    if len(fileitems) < 2 or any(fileitem.type != "file" for fileitem in fileitems):
        return False
    audio_extensions = set(get_api_runtime_config_snapshot().audio_extensions)
    return all(
        Path(fileitem.path or fileitem.name or "").suffix.lower() in audio_extensions
        for fileitem in fileitems
    )


def _selected_music_fileitems(
    fileitems: List[FileItem],
    *,
    explicitly_selected: bool,
    media_type: Optional[MediaType],
) -> Optional[List[FileItem]]:
    """
    仅在用户明确选择音乐类型时，将显式多选音轨作为专辑批次处理。
    """
    if not explicitly_selected or media_type != MediaType.MUSIC:
        return None
    if not _is_music_file_batch(fileitems):
        return None
    paths = [Path(fileitem.path) for fileitem in fileitems if fileitem.path]
    parent_counts: dict[Path, int] = {}
    for path in paths:
        parent_counts[path.parent] = parent_counts.get(path.parent, 0) + 1
    album_roots = [
        parent
        for parent in parent_counts
        if all(path.parent == parent or parent in path.parents for path in paths)
    ]
    if not album_roots:
        return fileitems
    album_root = max(album_roots, key=lambda path: len(path.parts))
    filtered = []
    for fileitem, path in zip(fileitems, paths):
        if path.parent == album_root:
            filtered.append(fileitem)
            continue
        relative_parent = path.parent.relative_to(album_root)
        first_directory = relative_parent.parts[0] if relative_parent.parts else ""
        if re.fullmatch(r"(?:cd|disc|disk)\s*\d{1,2}", first_directory.strip(), re.IGNORECASE):
            filtered.append(fileitem)
    if filtered and len(filtered) < len(fileitems):
        logger.info(
            "音乐专辑批次忽略 %s 个非碟片子目录中的重复或附加音频",
            len(fileitems) - len(filtered),
        )
        return filtered
    return fileitems


def _manual_review_actor(current_user: object) -> str:
    """
    按名称、用户名和用户 ID 的稳定顺序提取人工复核操作者。
    """
    for attribute in ("name", "username", "id"):
        value = getattr(current_user, attribute, None)
        if value is not None and str(value).strip():
            return str(value).strip()
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail="当前管理用户缺少可审计身份",
    )


def _manual_review_task_data(
    task: TransferManualReviewTaskView,
) -> _SchemaTransferManualReviewTaskData:
    """
    把 Application 人工复核投影映射为严格公开响应。
    """
    return _SchemaTransferManualReviewTaskData.model_validate({
            "task_id": task.task_id,
            "source": {
                "storage": task.source.storage,
                "path": task.source.path,
            },
            "state": task.state.value,
            "step": {
                "operation_id": task.step.operation_id,
                "kind": task.step.kind,
                "intent": task.step.intent,
                "evidence": task.step.evidence,
                "error": _public_transfer_message(task.step.error),
            },
            "review_revision": task.review_revision,
        })


@router.get(
    "/tasks/manual-reviews",
    summary="分页查询 durable 整理人工复核任务",
    response_model=_SchemaResponse[_SchemaTransferManualReviewPageData],
)
def list_transfer_manual_reviews(
    state_filter: Literal["manual_review", "retry_wait"] = Query(
        default="manual_review",
        alias="state",
    ),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=30, ge=1, le=100),
    current_user: object = Depends(get_current_active_manage_user),
    repository: TransferExecutionRepository = Depends(
        get_transfer_execution_repository
    ),
) -> Any:
    """
    分页返回待复核或已判定等待 durable 恢复的任务。
    """
    del current_user
    result = TransferManualReviewQuery(repository).list(
        state=TransferExecutionState(state_filter),
        page=page,
        page_size=page_size,
    )
    return _SchemaResponse(
        success=True,
        data=_SchemaTransferManualReviewPageData(
            items=[_manual_review_task_data(item) for item in result.items],
            total=result.total,
            page=result.page,
            page_size=result.page_size,
        ),
    )


@router.get(
    "/tasks/{task_id}/manual-review",
    summary="查询 durable 整理人工复核详情",
    response_model=_SchemaResponse[_SchemaTransferManualReviewTaskData],
)
def get_transfer_manual_review(
    task_id: str,
    current_user: object = Depends(get_current_active_manage_user),
    repository: TransferExecutionRepository = Depends(
        get_transfer_execution_repository
    ),
) -> Any:
    """
    按任务标识返回严格裁剪的人工复核详情。
    """
    del current_user
    task = TransferManualReviewQuery(repository).get(task_id=task_id)
    if task is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="人工复核任务不存在",
        )
    return _SchemaResponse(success=True, data=_manual_review_task_data(task))


@router.post(
    "/tasks/{task_id}/manual-review",
    summary="人工判定整理步骤的外部执行结果",
    response_model=_SchemaResponse[_SchemaTransferManualReviewData],
)
def resolve_transfer_manual_review(
    task_id: str,
    review: _SchemaTransferManualReviewRequest,
    current_user: object = Depends(get_current_active_manage_user),
    repository: TransferExecutionRepository = Depends(
        get_transfer_execution_repository
    ),
) -> Any:
    """
    提交无租约人工判定，并返回不含 attempt 与 lease 的公开状态。
    """
    result = (
        TransferStepResult(payload=dict(review.result_payload))
        if review.result_payload is not None
        else None
    )
    try:
        resolved = TransferExecutionCommand(repository).resolve_manual_review(
            task_id=task_id,
            operation_id=review.operation_id,
            decision=TransferManualReviewDecision(review.decision),
            actor=_manual_review_actor(current_user),
            reason=review.reason,
            result=result,
        )
    except TransferExecutionConflictError as error:
        logger.warning(f"整理人工复核请求冲突：{error}", exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="整理任务状态已变化，请刷新后重试",
        ) from error
    return _SchemaResponse(
        success=True,
        data=_SchemaTransferManualReviewData(
            task_id=resolved.task_id,
            operation_id=resolved.operation_id,
            decision=(
                "applied"
                if resolved.decision is TransferManualReviewDecision.APPLIED
                else "not_applied"
            ),
            state=resolved.state.value,
            review_revision=resolved.review_revision,
        ),
    )


@router.post(
    "/tasks/{task_id}/manual-review/discard",
    summary="放弃人工复核整理任务但保留文件",
    response_model=_SchemaResponse[_SchemaTransferManualReviewDiscardData],
)
def discard_transfer_manual_review(
    task_id: str,
    review: _SchemaTransferManualReviewDiscardRequest,
    current_user: object = Depends(get_current_active_manage_user),
    repository: TransferExecutionRepository = Depends(get_transfer_execution_repository),
) -> Any:
    """解除无法确认的整理任务；不删除、不移动且不回滚任何文件。"""
    try:
        discarded = TransferExecutionCommand(repository).discard_manual_review(
            task_id=task_id,
            operation_id=review.operation_id,
            actor=_manual_review_actor(current_user),
            reason=review.reason,
        )
    except TransferExecutionConflictError as error:
        logger.warning(f"整理人工复核放弃请求冲突：{error}", exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="整理任务状态已变化，请刷新后重试",
        ) from error
    return _SchemaResponse(
        success=True,
        data=_SchemaTransferManualReviewDiscardData(
            task_id=discarded.task_id,
            operation_id=discarded.operation_id,
            discarded=discarded.discarded,
            message=discarded.message,
        ),
    )


@router.get(
    "/name",
    summary="查询整理后的名称",
    response_model=_SchemaResponse[_SchemaNameData],
)
def query_name(
    path: str, filetype: str, _: _SchemaTokenPayload = Depends(verify_token)
) -> Any:
    """
    查询整理后的名称

    :param path: 文件路径
    :param filetype: 文件类型
    :param _: Token校验
    """
    context = MediaChain().recognize_by_path(
        path,
        obtain_images=False,
    )
    if not context or not context.media_info:
        return _SchemaResponse(success=False, message="未识别到媒体信息")
    new_path = TransferChain().recommend_name(
        meta=context.meta_info, mediainfo=context.media_info
    )
    if not new_path:
        return _SchemaResponse(success=False, message="未识别到新名称")
    if filetype == "dir":
        media_type = context.media_info.type
        if media_type is None:
            return _SchemaResponse(success=False, message="未识别到媒体类型")
        media_path = DirectoryHelper.get_media_root_path(
            rename_format=get_api_runtime_config_snapshot().rename_format(
                media_type
            ),
            rename_path=Path(new_path),
            media_type=media_type,
        )
        if media_path:
            new_name = media_path.name
        else:
            # 模板未提供媒体根目录时，沿用已有父目录层级推导名称。
            parents = Path(new_path).parents
            if len(parents) > 2:
                new_name = parents[1].name
            else:
                new_name = parents[0].name
    else:
        new_name = Path(new_path).name
    return _SchemaResponse(success=True, data={"name": new_name})


@router.get("/queue", summary="查询整理队列", response_model=List[_SchemaTransferJob])
async def query_queue(_: _SchemaTokenPayload = Depends(verify_token), page: CompatiblePageParam = None, count: CompatibleCountParam = None) -> Any:
    """
    查询整理队列

    :param _: Token校验
    """
    return TransferChain().get_queue_tasks()


@router.get(
    "/queue/page",
    summary="查询受限整理队列快照",
    response_model=_SchemaTransferQueuePageData,
)
async def query_queue_page(
    _: _SchemaTokenPayload = Depends(verify_token),
    page: int = Query(1, ge=1),
    count: int = Query(100, ge=1, le=COLLECTION_MAX_PAGE_SIZE),
) -> _SchemaTransferQueuePageData:
    """
    返回前端所需的有限整理队列窗口，避免序列化完整云盘目录树。
    """
    items, total = TransferChain().get_queue_tasks_page(page, count)
    return _SchemaTransferQueuePageData(
        items=items,
        total=total,
        page=page,
        count=count,
    )


@router.delete(
    "/queue", summary="从整理队列中删除任务", response_model=_SchemaResponse[None]
)
async def remove_queue(
    fileitem: _SchemaFileItem, _: _SchemaTokenPayload = Depends(verify_token)
) -> Any:
    """
    查询整理队列

    :param fileitem: 文件项
    :param _: Token校验
    """
    TransferChain().remove_from_queue(fileitem)
    # 取消整理
    if not fileitem.path:
        return _SchemaResponse(success=False, message="文件路径为空")
    runtime_stop_state.stop_transfer(fileitem.path)
    return _SchemaResponse(success=True)


def _resolve_manual_transfer_source_fileitems(
    transer_item: ManualTransferItem,
    history_query: TransferHistoryLookupService,
) -> tuple[List[FileItem], Optional[str]]:
    """
    从手动整理请求中解析源文件项。
    """
    if transer_item.logids:
        fileitems: List[FileItem] = []
        for logid in transer_item.logids:
            history = history_query.get(logid)
            if not history:
                return [], f"整理记录不存在，ID：{logid}"
            if history.status and ("move" in (history.mode or "")):
                fileitem = _restore_history_fileitem(history.dest_fileitem)
            else:
                fileitem = _restore_history_fileitem(history.src_fileitem)
            if fileitem is None:
                return [], f"整理记录缺少文件项，ID：{logid}"
            fileitems.append(fileitem)
        return fileitems, None

    if transer_item.logid:
        history = history_query.get(transer_item.logid)
        if not history:
            return [], f"整理记录不存在，ID：{transer_item.logid}"
        if history.status and ("move" in (history.mode or "")):
            fileitem = _restore_history_fileitem(history.dest_fileitem)
        else:
            fileitem = _restore_history_fileitem(history.src_fileitem)
        if fileitem is None:
            return [], f"整理记录缺少文件项，ID：{transer_item.logid}"
        return [fileitem], None

    if transer_item.fileitems:
        return [fileitem for fileitem in transer_item.fileitems if fileitem], None
    if transer_item.fileitem:
        return [transer_item.fileitem], None
    return [], None


def _deduplicate_fileitems(fileitems: List[FileItem]) -> List[FileItem]:
    """
    按存储和路径去重文件项。
    """
    dedup_fileitems: List[FileItem] = []
    seen_paths = set()
    for current_fileitem in fileitems:
        storage = current_fileitem.storage or "local"
        path = current_fileitem.path
        if not path:
            continue
        key = (storage, path)
        if key in seen_paths:
            continue
        seen_paths.add(key)
        dedup_fileitems.append(current_fileitem)
    return dedup_fileitems


def _build_manual_transfer_target_path(
    directory: Optional[_SchemaTransferDirectoryConf] = None,
) -> _SchemaManualTransferTargetPath:
    """
    根据目录配置生成手动整理目的路径响应。
    """
    if not directory or not directory.library_path:
        return _SchemaManualTransferTargetPath()

    return _SchemaManualTransferTargetPath(
        target_storage=directory.library_storage or "local",
        target_path=directory.library_path,
        transfer_type=directory.transfer_type,
        scrape=directory.scraping or False,
        library_type_folder=directory.library_type_folder or False,
        library_category_folder=directory.library_category_folder or False,
    )


def _get_manual_transfer_target_key(
    directory: _SchemaTransferDirectoryConf,
) -> tuple[Optional[str], Optional[str]]:
    """
    生成目的目录唯一键。
    """
    return (
        directory.library_storage or "local",
        Path(directory.library_path).as_posix() if directory.library_path else None,
    )


@router.post(
    "/manual/target-path",
    summary="匹配手动转移目的路径",
    response_model=_SchemaResponse[_SchemaManualTransferTargetPath],
)
def match_manual_transfer_target_path(
    transer_item: ManualTransferItem,
    history_query: TransferHistoryLookupService = Depends(
        get_transfer_history_lookup_service
    ),
    _: object = Depends(get_current_active_manage_user),
) -> Any:
    """
    根据源文件匹配手动整理目的路径。

    :param transer_item: 手工整理项
    :param history_query: 整理历史投影服务
    :param _: Token校验
    """
    src_fileitems, error_message = _resolve_manual_transfer_source_fileitems(
        transer_item=transer_item,
        history_query=history_query,
    )
    if error_message:
        return _SchemaResponse(success=False, message=error_message)

    matched_directories: List[_SchemaTransferDirectoryConf] = []
    target_storage = transer_item.target_storage or None
    for src_fileitem in _deduplicate_fileitems(src_fileitems):
        if not src_fileitem.path:
            return _SchemaResponse(success=False, message="源文件路径为空")
        directory = DirectoryHelper().get_dir(
            media=None,
            storage=src_fileitem.storage or "local",
            src_path=Path(src_fileitem.path),
            target_storage=target_storage,
        )
        if not directory or not directory.library_path:
            return _SchemaResponse(
                success=True,
                data=_SchemaManualTransferTargetPath().model_dump(),
            )
        matched_directories.append(directory)

    if not matched_directories:
        return _SchemaResponse(
            success=True,
            data=_SchemaManualTransferTargetPath().model_dump(),
        )

    first_directory = matched_directories[0]
    first_key = _get_manual_transfer_target_key(first_directory)
    if any(
        _get_manual_transfer_target_key(directory) != first_key
        for directory in matched_directories[1:]
    ):
        return _SchemaResponse(
            success=True,
            data=_SchemaManualTransferTargetPath().model_dump(),
        )

    return _SchemaResponse(
        success=True,
        data=_build_manual_transfer_target_path(first_directory).model_dump(),
    )


@router.post(
    "/manual/history",
    summary="查询手动转移成功历史",
    response_model=_SchemaResponse[_SchemaManualTransferHistoryInfo],
)
def query_manual_transfer_history(
    transer_item: ManualTransferItem,
    history_query: TransferHistoryLookupService = Depends(
        get_transfer_history_lookup_service
    ),
    _: object = Depends(get_current_active_manage_user),
) -> Any:
    """
    查询文件或目录命中的成功整理记录。

    :param transer_item: 手工整理项
    :param history_query: 整理历史投影服务
    :param _: Token校验
    """
    src_fileitems, error_message = _resolve_manual_transfer_source_fileitems(
        transer_item=transer_item,
        history_query=history_query,
    )
    if error_message:
        return _SchemaResponse(success=False, message=error_message)

    histories = TransferChain().get_manual_transfer_histories(
        _deduplicate_fileitems(src_fileitems)
    )
    history_info = _SchemaManualTransferHistoryInfo(
        reorganize=bool(histories),
        history_count=len(histories),
    )
    return _SchemaResponse(success=True, data=history_info.model_dump())


@router.post(
    "/manual",
    summary="手动转移",
    response_model=_SchemaResponse[_SchemaManualTransferSubmissionData | _SchemaManualTransferResultData],
)
def manual_transfer(
    transer_item: ManualTransferItem,
    background: Optional[bool] = False,
    history_query: TransferHistoryLookupService = Depends(
        get_transfer_history_lookup_service
    ),
    _: object = Depends(get_current_active_manage_user),
) -> Any:
    """
    解析手动整理 HTTP 请求并委托兼容用例处理器。

    :param transer_item: 手工整理项
    :param background: 后台运行
    :param history_query: 整理历史投影服务
    :param _: Token校验
    """
    return _route_manual_transfer(
        transer_item=transer_item,
        background=background,
        history_query=history_query,
    )


def _route_manual_transfer(
    transer_item: ManualTransferItem,
    background: Optional[bool],
    history_query: TransferHistoryLookupService,
) -> Any:
    """
    执行历史恢复、批量预览与 TransferChain 兼容编排。
    """
    if not transer_item.logids:
        return _execute_manual_transfer(
            transer_item=transer_item,
            background=background,
            history_query=history_query,
        )

    (
        src_fileitems,
        force,
        downloader,
        download_hash,
        history_error,
    ) = restore_manual_transfer_history_batch(
        transer_item=transer_item,
        history_query=history_query,
    )
    if history_error or not src_fileitems:
        return _SchemaResponse(
            success=False,
            message=history_error or "缺少参数",
        )
    transer_item.fileitems = src_fileitems
    transer_item.logids = None
    return _execute_manual_transfer(
        transer_item=transer_item,
        background=background,
        history_query=history_query,
        force=force,
        downloader=downloader,
        download_hash=download_hash,
    )


def _execute_manual_transfer(
    transer_item: ManualTransferItem,
    background: Optional[bool],
    history_query: TransferHistoryLookupService,
    force: bool = False,
    downloader: Optional[str] = None,
    download_hash: Optional[str] = None,
) -> Any:
    """
    执行已还原源文件项的手动整理兼容编排。
    """
    src_fileitems: List[FileItem] = []
    cleanup_dest_fileitem: Optional[FileItem] = None
    target_path = Path(transer_item.target_path) if transer_item.target_path else None
    if transer_item.logid:
        # 查询历史记录
        history = history_query.get(transer_item.logid)
        if not history:
            return _SchemaResponse(
                success=False, message=f"整理记录不存在，ID：{transer_item.logid}"
            )
        # 失败历史必须经过整理链的重试/重整判定，不能绕过旧任务直接重新准入。
        force = bool(history.status)
        # 下载器与 Hash 是同一组下载上下文，重新识别时由当前文件路径重新匹配。
        downloader = history.downloader if transer_item.from_history else None
        download_hash = history.download_hash if transer_item.from_history else None
        if history.status and ("move" in (history.mode or "")):
            # 重新整理成功的转移，则使用成功的 dest 做 in_path
            fileitem = _restore_history_fileitem(history.dest_fileitem)
        else:
            # 源路径
            fileitem = _restore_history_fileitem(history.src_fileitem)
            if (
                history.dest_fileitem
                and not transer_item.preview
                and not transer_item.reorganize
            ):
                cleanup_dest_fileitem = _restore_history_fileitem(history.dest_fileitem)
        src_fileitems = [fileitem] if fileitem else []

        # 从历史数据获取信息
        if transer_item.from_history:
            restore_manual_transfer_history_metadata(transer_item, history)

    elif transer_item.fileitems:
        src_fileitems = [fileitem for fileitem in transer_item.fileitems if fileitem]
    elif transer_item.fileitem:
        src_fileitems = [transer_item.fileitem]
    else:
        return _SchemaResponse(success=False, message="缺少参数")

    src_fileitems = _deduplicate_fileitems(src_fileitems)
    if not src_fileitems:
        return _SchemaResponse(success=False, message="缺少参数")

    # 类型（“自动/auto/none”按未指定处理）
    mtype = None
    type_name = str(transer_item.type_name).strip() if transer_item.type_name else ""
    if type_name and type_name.lower() not in {"自动", "auto", "none"}:
        # 先按 Agent 使用的 movie/tv/music 取值解析，再回退到原生中文枚举值。
        try:
            mtype = MediaType.from_agent(type_name) or MediaType(type_name)
        except ValueError:
            return _SchemaResponse(
                success=False, message=f"不支持的媒体类型：{type_name}"
            )

    def _resolve_music_type(file_item: FileItem) -> Optional[str]:
        """
        为未显式指定实体的旧客户端按源项类型补全音乐命名空间。
        """
        if mtype != MediaType.MUSIC or transer_item.music_type:
            return transer_item.music_type
        return (
            MUSIC_ENTITY_ALBUM
            if file_item.type == "dir"
            else MUSIC_ENTITY_RECORDING
        )
    # 自定义格式
    epformat = None
    if (
        transer_item.episode_offset
        or transer_item.episode_part
        or transer_item.episode_detail
        or transer_item.episode_format
    ):
        epformat = _SchemaEpisodeFormat(
            format=transer_item.episode_format,
            detail=transer_item.episode_detail,
            part=transer_item.episode_part,
            offset=transer_item.episode_offset,
        )
    explicit_selected_files = bool(transer_item.fileitems or transer_item.logids)
    selected_music_fileitems = _selected_music_fileitems(
        src_fileitems, explicitly_selected=explicit_selected_files, media_type=mtype
    )
    mtype = MediaType.MUSIC if selected_music_fileitems is not None else mtype
    explicit_selected_files = explicit_selected_files and selected_music_fileitems is None

    release_kwargs: dict[str, Any] = (
        {"musicbrainz_release_id": transer_item.musicbrainz_release_id}
        if transer_item.musicbrainz_release_id is not None else {}
    )

    # 前端显式传入文件列表时，按选中的文件逐个处理，避免将目录整体展开。
    if explicit_selected_files:
        preview_items: List[dict[str, Any]] = []
        submission_items: list[_SchemaManualTransferSubmissionItem] = []
        error_messages: List[str] = []
        all_success = True
        for src_fileitem in src_fileitems:
            state, errormsg = TransferChain().manual_transfer(
                fileitem=src_fileitem,
                target_storage=transer_item.target_storage,
                target_path=target_path,
                media_source=transer_item.media_source,
                media_id=transer_item.media_id,
                music_type=_resolve_music_type(src_fileitem),
                music_release_regions=transer_item.music_release_regions,
                music_release_scripts=transer_item.music_release_scripts,
                music_cue_enable=transer_item.music_cue_enable,
                **release_kwargs,
                mtype=mtype,
                season=transer_item.season,
                episode_group=transer_item.episode_group,
                transfer_type=transer_item.transfer_type,
                epformat=epformat,
                min_filesize=transer_item.min_filesize,
                scrape=transer_item.scrape,
                library_type_folder=transer_item.library_type_folder,
                library_category_folder=transer_item.library_category_folder,
                force=force,
                background=background,
                downloader=downloader,
                download_hash=download_hash,
                preview=transer_item.preview,
                reorganize=transer_item.reorganize,
                skip_success=transer_item.skip_success,
                sync_extra_files=False,
                cleanup_dest_fileitem=cleanup_dest_fileitem,
                report_results=not transer_item.preview,
            )
            if transer_item.preview:
                if isinstance(errormsg, dict):
                    preview_items.extend(
                        _public_transfer_result(errormsg).get("items") or []
                    )
                    if errormsg.get("message"):
                        message = _public_transfer_message(errormsg.get("message"))
                        error_messages.append(message or "")
                    if not state:
                        all_success = False
                else:
                    if errormsg:
                        public_message = _public_transfer_message(errormsg)
                        if public_message:
                            error_messages.append(public_message)
                    preview_items.append(
                        _build_failure_preview_item(
                            src_fileitem,
                            _public_transfer_message(errormsg),
                        )
                    )
                    all_success = False
            else:
                all_success = all_success and state
                submission_items.extend(_manual_submission_items(src_fileitem, state, errormsg, target_path))

        if transer_item.preview:
            merged_preview_items: List[dict[str, Any]] = []
            seen_sources = set()
            for preview_item in preview_items:
                source = (preview_item.get("source_storage"), preview_item.get("source"))
                if source in seen_sources:
                    continue
                seen_sources.add(source)
                merged_preview_items.append(
                    _public_transfer_result(preview_item)
                    if isinstance(preview_item, dict)
                    else preview_item
                )
            merged_message = _merge_transfer_messages(error_messages)
            preview_data = {
                "summary": {
                    "total": len(merged_preview_items),
                    "success": len(
                        [item for item in merged_preview_items if item.get("success")]
                    ),
                    "failed": len(
                        [item for item in merged_preview_items if not item.get("success")]
                    ),
                },
                "items": merged_preview_items,
                "message": merged_message,
            }
            return _SchemaResponse(
                success=True,
                message=merged_message or "",
                data=preview_data,
            )

        return _manual_submission_response(all_success, submission_items)

    src_fileitem = src_fileitems[0]
    # 开始转移
    state, errormsg = TransferChain().manual_transfer(
        fileitem=src_fileitem,
        target_storage=transer_item.target_storage,
        target_path=target_path,
        media_source=transer_item.media_source,
        media_id=transer_item.media_id,
        music_type=(
            MUSIC_ENTITY_ALBUM
            if selected_music_fileitems is not None
            else _resolve_music_type(src_fileitem)
        ),
        music_release_regions=transer_item.music_release_regions,
        music_release_scripts=transer_item.music_release_scripts,
        music_cue_enable=transer_item.music_cue_enable,
        **release_kwargs,
        mtype=mtype,
        season=transer_item.season,
        episode_group=transer_item.episode_group,
        transfer_type=transer_item.transfer_type,
        epformat=epformat,
        min_filesize=transer_item.min_filesize,
        scrape=transer_item.scrape,
        library_type_folder=transer_item.library_type_folder,
        library_category_folder=transer_item.library_category_folder,
        force=force,
        background=background,
        downloader=downloader,
        download_hash=download_hash,
        preview=transer_item.preview,
        reorganize=transer_item.reorganize,
        skip_success=transer_item.skip_success,
        sync_extra_files=selected_music_fileitems is None,
        cleanup_dest_fileitem=cleanup_dest_fileitem,
        selected_fileitems=selected_music_fileitems,
        report_results=not transer_item.preview,
    )
    if not transer_item.preview:
        return _manual_submission_response(state, _manual_submission_items(src_fileitem, state, errormsg, target_path))
    # 预览失败仍返回完整预览列表，实际提交的总状态已在上方独立处理。
    if isinstance(errormsg, dict):
        public_result = _public_transfer_result(errormsg)
        return _SchemaResponse(success=True, message=public_result.get("message") or "", data=public_result)
    if state:
        return _SchemaResponse(success=True, data=errormsg or {})
    return _SchemaResponse(success=False, message=_format_manual_transfer_failure(
        message=errormsg, source_path=src_fileitem.path,
        target_path=target_path.as_posix() if target_path else None,
    ))


@router.post(
    "/episode-format/recommend",
    summary="推荐集数定位模板",
    response_model=_SchemaResponse[_SchemaEpisodeFormatRecommendData],
)
def recommend_episode_format(
    recommend_item: EpisodeFormatRecommendItem,
    _: object = Depends(get_current_active_manage_user),
) -> Any:
    """
    根据目录样本推荐集数定位模板

    :param recommend_item: 推荐请求
    :param _: Token校验
    """
    target_path = recommend_item.fileitem.path if recommend_item.fileitem else None
    logger.info(f"开始推荐集数定位模板：{target_path}")
    state, errmsg, data = TransferChain().recommend_episode_format(
        fileitem=recommend_item.fileitem,
        fileitems=recommend_item.fileitems,
    )
    if not state:
        logger.warn(f"推荐集数定位模板失败：{target_path} - {errmsg}")
        return _SchemaResponse(
            success=False,
            message=_public_transfer_message(errmsg) or "",
        )
    logger.info(
        f"推荐集数定位模板成功：{target_path} - 规则 {data.get('rule_name') if data else None}"
    )
    return _SchemaResponse(success=True, data=data)


@router.get("/now", summary="立即执行下载器文件整理", response_model=_SchemaResponse[None])
def now(_: Annotated[str, Depends(verify_apitoken)]) -> Any:
    """
    立即执行下载器文件整理 API_TOKEN认证（?token=xxx）
    """
    TransferChain().process()
    return _SchemaResponse(success=True)