video-gen/video-gen-api/app/api/v3/videos.py

import logging
import time

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.models.api.api_generation_task import ApiGenerationTask
from app.schemas.api_v3.video import (
    ApiVideoCreateRequest,
    ApiVideoCreateResponse,
    ApiVideoStatusResponse,
)
from app.services.api_v3 import auth_service, generation_service, task_service
from app.services.resource_signed_url_service import build_resource_signed_url

logger = logging.getLogger("videogen")

router = APIRouter(prefix="/videos", tags=["api-v3-videos"])


async def _validate_request(
    db: AsyncSession,
    key_context: auth_service.ApiKeyContext,
    req: ApiVideoCreateRequest,
) -> ApiGenerationTask | None:
    """请求层校验：参数、权限、幂等性。

    Returns:
        None = 校验通过，继续创建
        ApiGenerationTask = 幂等请求，返回已有任务
    """
    # 模型权限校验
    allowed_model_names = {m.get("model_name", "") for m in key_context.callable_models}
    if allowed_model_names and req.model not in allowed_model_names:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"无权使用模型 {req.model}",
        )

    # 幂等性检查
    if req.idempotency_key:
        result = await db.execute(
            select(ApiGenerationTask).where(
                ApiGenerationTask.api_key_id == key_context.api_key.id,
                ApiGenerationTask.external_idempotency_key == req.idempotency_key,
                ApiGenerationTask.deleted_at.is_(None),
            ).limit(1)
        )
        existing_task = result.scalar_one_or_none()
        if existing_task:
            logger.info(
                "Idempotent request: returning existing task %s for key=%s",
                existing_task.id, req.idempotency_key,
            )
            return existing_task

    return None


def _map_status(internal_status: str) -> str:
    """将内部状态映射为 API 状态。"""
    status_map = {
        "pending": "queued",
        "queued": "queued",
        "generating": "running",
        "processing": "running",
        "completed": "succeeded",
        "failed": "failed",
        "timeout": "expired",
    }
    return status_map.get(internal_status, internal_status)


@router.post(
    "",
    response_model=ApiVideoCreateResponse,
    summary="创建视频生成任务",
)
async def create_video(
    req: ApiVideoCreateRequest,
    key_context: auth_service.ApiKeyContext = Depends(auth_service.get_api_key_dependency),
    db: AsyncSession = Depends(get_db),
) -> ApiVideoCreateResponse:
    """创建视频生成任务（异步）。

    幂等性说明：如果 idempotency_key 已存在，直接返回已有任务 ID（不会重复创建）。
    """
    try:
        # 路由层校验：权限、幂等性
        existing_task = await _validate_request(db, key_context, req)
        if existing_task:
            logger.info(
                "Idempotent request: returning existing task %s for key=%s",
                existing_task.id, req.idempotency_key,
            )
            return ApiVideoCreateResponse(id=f"zc-{existing_task.id}")

        # 调用服务层创建任务
        result = await generation_service.submit_video_generation(
            db=db,
            key=key_context.api_key,
            callable_models=key_context.callable_models,
            req=req,
        )
        return ApiVideoCreateResponse(id=f"zc-{result.id}")

    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("API video creation failed")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建视频任务失败: {str(exc)[:200]}",
        )


@router.get(
    "/{task_id}",
    response_model=ApiVideoStatusResponse,
    summary="查询视频任务状态",
)
async def get_video_status(
    task_id: str,
    key_context: auth_service.ApiKeyContext = Depends(auth_service.get_api_key_dependency),
    db: AsyncSession = Depends(get_db),
) -> ApiVideoStatusResponse:
    """查询视频任务状态。"""
    # 去掉 zc- 前缀
    if task_id.startswith("zc-"):
        task_id = task_id[3:]
    task = await task_service.get_task(db, task_id, key_context.api_key.id)
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"任务 {task_id} 不存在或不属于当前 API Key",
        )

    now = int(time.time())
    # 构建 content（成功时返回完整视频URL，包含 BASE_URL）
    content = None
    if task.status == "completed" and task.video_url:
        from app.schemas.api_v3.video import ApiVideoContent
        from app.config import settings
        # 拼接完整 URL
        video_url = build_resource_signed_url(task.video_url)
        if video_url and not video_url.startswith(("http://", "https://")):
            base = settings.BASE_URL.rstrip("/")
            if video_url.startswith("/"):
                video_url = f"{base}{video_url}"
            else:
                video_url = f"{base}/{video_url}"
        content = ApiVideoContent(video_url=video_url)

    return ApiVideoStatusResponse(
        id=f"zc-{task.id}",
        model=task.model_name,
        status=_map_status(task.status),
        created_at=int(task.created_at.timestamp()) if task.created_at else now,
        updated_at=int(task.updated_at.timestamp()) if task.updated_at else now,
        content=content,
        duration=task.duration,
        ratio=task.aspect_ratio,
        resolution=task.resolution,
        error=task.error_message if task.status in ("failed", "timeout") else None,
    )