xiaoxia-saas/apps/api/app/api/routes/generation_preview.py

"""预览生成路由 — Phase 1：单版本预览接口（创建 + 查询）。

路径前缀：/api/v1/generation/preview（与 /generation/tasks 同体系）
"""

from __future__ import annotations

import logging

from app.auth import AuthenticatedUser, get_current_user
from app.core.task_enqueue import (
    GLOBAL_PENDING_LIMIT,
    USER_PENDING_LIMIT,
    GlobalQueueFull,
    UserPendingLimitExceeded,
    safe_enqueue_generation_task,
)
from app.dependencies import (
    get_generated_video_repository,
    get_generation_task_repository,
)
from app.schemas.generation_task import (
    CreatePreviewGenerationTaskRequest,
    PreviewGenerationTaskResponse,
)
from fastapi import APIRouter, Depends, HTTPException

from packages.application import (
    CreateGenerationTaskCommand,
    CreateGenerationTaskUseCase,
    GetGenerationTaskUseCase,
    ListGeneratedVideosByTaskUseCase,
)

logger = logging.getLogger(__name__)

router = APIRouter()

PREVIEW_RESOLUTION = "854x480"


def _to_preview_response(task, generated_videos: list | None = None) -> PreviewGenerationTaskResponse:
    """将领域任务对象转换为预览响应 DTO。

    Args:
        task: GenerationTask 领域对象
        generated_videos: 生成的视频列表（可选），取第一个作为 video_url

    Returns:
        PreviewGenerationTaskResponse
    """
    video_url = ""
    duration = 0.0
    file_size = 0
    if generated_videos:
        first_video = generated_videos[0]
        video_url = getattr(first_video, "file_url", "") or ""
        duration = float(getattr(first_video, "duration", 0.0) or 0.0)
        file_size = int(getattr(first_video, "file_size", 0) or 0)

    # 从 extra_meta / metadata 中提取统计信息（如果有）
    extra_meta = getattr(task, "extra_meta", {}) or {}
    clip_count = int(extra_meta.get("clip_count", len(getattr(task, "asset_ids", [])) or 0))
    transition_count = int(extra_meta.get("transition_count", max(0, clip_count - 1)))
    material_usage = extra_meta.get("material_usage", {}) or {}

    # 计算生成耗时
    generate_duration = 0.0
    started_at = getattr(task, "started_at", None)
    completed_at = getattr(task, "completed_at", None)
    if started_at and completed_at:
        generate_duration = (completed_at - started_at).total_seconds()

    return PreviewGenerationTaskResponse(
        task_id=task.id,
        status=task.status.value if hasattr(task.status, "value") else str(task.status),
        progress=float(task.progress or 0.0),
        is_preview=bool(getattr(task, "is_preview", True)),
        resolution=getattr(task, "resolution", PREVIEW_RESOLUTION) or PREVIEW_RESOLUTION,
        video_url=video_url,
        duration=duration,
        file_size=file_size,
        clip_count=clip_count,
        transition_count=transition_count,
        material_usage=material_usage,
        error_message=task.error_message or "",
        created_at=task.created_at,
        started_at=started_at,
        finished_at=completed_at,
        generate_duration=generate_duration,
    )


@router.post("/preview", response_model=PreviewGenerationTaskResponse, status_code=201)
def create_preview_generation_task(
    request: CreatePreviewGenerationTaskRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    generation_task_repository=Depends(get_generation_task_repository),
) -> PreviewGenerationTaskResponse:
    """创建预览生成任务。

    预览为完整时长的低清版（480p + 低码率），效果与正式生成一致，仅清晰度降低。

    Args:
        request: 预览任务创建请求（template_id + asset_ids 等）

    Returns:
        201 + 预览任务详情
    """
    user_id = authenticated_user.user.id
    logger.info(
        "[预览生成] 接收请求: user_id=%s, template_id=%s, asset_count=%d",
        user_id,
        request.template_id,
        len(request.asset_ids),
    )

    # 预检查队列限流
    try:
        user_pending = generation_task_repository.count_pending_by_user(user_id)
        global_pending = generation_task_repository.count_pending_total()
        if user_pending + 1 > USER_PENDING_LIMIT:
            raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending + 1, limit=USER_PENDING_LIMIT)
        if global_pending + 1 > GLOBAL_PENDING_LIMIT:
            raise GlobalQueueFull(pending_count=global_pending + 1, limit=GLOBAL_PENDING_LIMIT)
    except UserPendingLimitExceeded as e:
        raise HTTPException(
            status_code=429,
            detail=f"您的待处理任务过多（当前 {e.pending_count - 1}/{e.limit}），请等待完成后再提交",
        ) from e
    except GlobalQueueFull as e:
        raise HTTPException(
            status_code=503,
            detail="系统繁忙，请稍后再试",
        ) from e

    use_case = CreateGenerationTaskUseCase(generation_task_repository)

    try:
        task = use_case.execute(
            CreateGenerationTaskCommand(
                project_id="",
                asset_library_id="",
                strategy_id="",
                voice_library_id="",
                template_id=request.template_id,
                asset_ids=list(request.asset_ids),
                title_ids=list(request.title_ids),
                voice_ids=list(request.voice_ids),
                created_by_user_id=user_id,
                source_edit_plan_id="",
                asset_select_mode="",
                batch_id="",
                video_title=request.video_title,
                resolution=PREVIEW_RESOLUTION,
                bgm_config=request.bgm_config or {},
                auto_retry_enabled=False,
                auto_retry_max=0,
                is_preview=True,
            )
        )
    except ValueError as e:
        logger.warning("[预览生成] 创建失败: %s", e)
        raise HTTPException(status_code=400, detail=str(e)) from e
    except Exception as e:
        logger.error("[预览生成] 创建失败: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail="创建预览生成任务失败，请稍后重试") from e

    # 入队执行
    try:
        if not safe_enqueue_generation_task(
            task,
            generation_task_repository,
            user_id=user_id,
            log_prefix="[预览生成]",
            log_task_status=True,
        ):
            logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id)
            raise HTTPException(status_code=500, detail="任务入队失败，请稍后重试")
    except UserPendingLimitExceeded:
        raise HTTPException(
            status_code=429,
            detail="您的待处理任务过多，请等待完成后再提交",
        ) from None
    except GlobalQueueFull:
        raise HTTPException(
            status_code=503,
            detail="系统繁忙，请稍后再试",
        ) from None

    return _to_preview_response(task)


@router.get("/preview/{task_id}", response_model=PreviewGenerationTaskResponse)
def get_preview_generation_task(
    task_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    generation_task_repository=Depends(get_generation_task_repository),
    generated_video_repository=Depends(get_generated_video_repository),
) -> PreviewGenerationTaskResponse:
    """查询预览生成任务状态。

    Args:
        task_id: 任务 ID

    Returns:
        预览任务详情（含状态、进度、结果 URL 等）
    """
    use_case = GetGenerationTaskUseCase(generation_task_repository)
    task = use_case.execute(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"预览任务 {task_id} 不存在")

    # 权限校验：任务必须属于当前用户
    task_user_id = getattr(task, "created_by_user_id", "") or ""
    if not task_user_id or task_user_id != authenticated_user.user.id:
        raise HTTPException(status_code=403, detail="无权访问该任务")

    # 校验是否为预览任务
    if not getattr(task, "is_preview", False):
        raise HTTPException(status_code=404, detail=f"预览任务 {task_id} 不存在")

    # 查询生成的视频（取第一个）
    generated_videos = []
    status_val = task.status.value if hasattr(task.status, "value") else str(task.status)
    if status_val == "completed":
        list_use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository)
        generated_videos = list_use_case.execute(task_id)

    return _to_preview_response(task, generated_videos=generated_videos)