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

"""渲染结果内部下载接口。

通过内部 API Key 鉴权，为灰度对比工具等内部系统提供渲染结果下载能力。

API:
  GET  /api/v1/internal/render/videos/{video_id}/download-url  - 获取单个视频下载URL
  GET  /api/v1/internal/render/tasks/{task_id}/videos          - 获取任务下所有视频及下载URL

鉴权：X-API-Key header，走内部 API Key 验证
"""

from __future__ import annotations

import logging
from typing import Any

from app.api.routes.auth import _verify_internal_api_key
from app.core.storage import OSSStorageService, get_storage_service
from app.dependencies import get_generated_video_repository
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/internal/render", tags=["Internal"])


class InternalRenderVideoItem(BaseModel):
    """内部渲染视频项。"""

    video_id: str
    generation_task_id: str
    project_id: str
    name: str
    file_url: str
    file_size: int | None = None
    duration: float | None = None
    width: int | None = None
    height: int | None = None
    fps: float | None = None
    status: str
    download_url: str


class InternalRenderTaskVideosResponse(BaseModel):
    """任务下所有渲染视频响应。"""

    task_id: str
    count: int
    videos: list[InternalRenderVideoItem]


class InternalRenderDownloadUrlResponse(BaseModel):
    """单个视频下载URL响应。"""

    video_id: str
    download_url: str


def _video_to_item(video: Any, download_url: str) -> InternalRenderVideoItem:
    """将 GeneratedVideo 领域对象转为响应项。"""
    return InternalRenderVideoItem(
        video_id=video.id,
        generation_task_id=video.generation_task_id,
        project_id=video.project_id,
        name=video.name,
        file_url=video.file_url,
        file_size=getattr(video, "file_size", None),
        duration=getattr(video, "duration", None),
        width=getattr(video, "width", None),
        height=getattr(video, "height", None),
        fps=getattr(video, "fps", None),
        status=video.status,
        download_url=download_url,
    )


@router.get("/videos/{video_id}/download-url", response_model=InternalRenderDownloadUrlResponse)
def get_render_video_download_url(
    video_id: str,
    _: bool = Depends(_verify_internal_api_key),
    generated_video_repository: Any = Depends(get_generated_video_repository),
    storage_service: OSSStorageService = Depends(get_storage_service),
) -> InternalRenderDownloadUrlResponse:
    """获取单个渲染视频的下载URL（预签名）。"""
    video = generated_video_repository.get(video_id)
    if video is None:
        raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found")

    download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
    logger.info("内部渲染下载URL生成: video_id=%s", video_id)
    return InternalRenderDownloadUrlResponse(video_id=video_id, download_url=download_url)


@router.get("/tasks/{task_id}/videos", response_model=InternalRenderTaskVideosResponse)
def get_render_task_videos(
    task_id: str,
    status: str | None = Query(None, description="按状态筛选，如 completed/failed"),
    _: bool = Depends(_verify_internal_api_key),
    generated_video_repository: Any = Depends(get_generated_video_repository),
    storage_service: OSSStorageService = Depends(get_storage_service),
) -> InternalRenderTaskVideosResponse:
    """获取生成任务下所有渲染视频及下载URL。"""
    videos = generated_video_repository.list_by_generation_task(task_id)

    # 状态筛选
    if status:
        videos = [v for v in videos if v.status == status]

    items = []
    for video in videos:
        download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400)
        items.append(_video_to_item(video, download_url))

    logger.info("内部渲染任务视频查询: task_id=%s count=%d", task_id, len(items))
    return InternalRenderTaskVideosResponse(
        task_id=task_id,
        count=len(items),
        videos=items,
    )