feat: 成片中心后端升级（封面生成/复核/批量下载） (#287)

feat: 成片中心后端升级（封面生成/复核/批量下载）

apps/api/app/api/router.py

@@ -19,6 +19,7 @@ from app.api.routes.templates import router as templates_router
 from app.api.routes.titles import router as titles_router
 from app.api.routes.tts import router as tts_router
 from app.api.routes.upload import router as upload_router
+from app.api.routes.videos import router as videos_router
 from app.api.routes.voice_clones import router as voice_clones_router
 from app.api.routes.voices import router as voices_router
 from fastapi import APIRouter
@@ -99,6 +100,10 @@ api_router.include_router(
     prefix="/voice-clones",
     tags=["VoiceClone"],
 )
+api_router.include_router(
+    videos_router,
+    tags=["VideoCenter"],
+)
 api_router.include_router(
     duplication_router,
     prefix="/duplication",

apps/api/app/api/routes/videos.py

@@ -0,0 +1,179 @@
+import logging
+import uuid
+
+from app.api.routes._helpers import check_project_access
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.celery_app import celery_app
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_generated_video_repository
+from app.schemas.video_center import (
+    BatchDownloadRequest,
+    BatchDownloadResponse,
+    ListVideosResponse,
+    UpdateVideoReviewRequest,
+    VideoItemResponse,
+)
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from packages.application import (
+    GetGeneratedVideoUseCase,
+    GetVideosByIdsUseCase,
+    ListGeneratedVideosPaginatedUseCase,
+    UpdateVideoReviewStatusUseCase,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+def _to_video_response(item, storage: OSSStorageService | None = None) -> VideoItemResponse:
+    download_url = None
+    if storage and item.file_url:
+        try:
+            download_url = storage.get_download_url(item.file_url)
+        except Exception:
+            download_url = item.file_url
+    return VideoItemResponse(
+        id=item.id,
+        project_id=item.project_id,
+        generation_task_id=item.generation_task_id,
+        name=item.name,
+        file_url=item.file_url,
+        file_size=item.file_size,
+        duration=item.duration,
+        thumbnail_url=item.thumbnail_url,
+        width=item.width,
+        height=item.height,
+        fps=item.fps,
+        status=item.status,
+        review_status=item.review_status,
+        generation_params=item.generation_params,
+        download_url=download_url,
+        generated_at=item.generated_at.isoformat() if hasattr(item, "generated_at") and item.generated_at else "",
+    )
+
+
+@router.get("/videos", response_model=ListVideosResponse)
+def list_videos(
+    project_id: str | None = Query(None, description="项目ID，不传则返回所有项目"),
+    status: str | None = Query(None, description="按状态筛选"),
+    review_status: str | None = Query(None, description="按复核状态筛选"),
+    page: int = Query(1, ge=1, description="页码"),
+    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
+    repo=Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """成片列表，支持分页、按项目/状态/复核状态筛选。"""
+    use_case = ListGeneratedVideosPaginatedUseCase(repo)
+    items, total = use_case.execute(
+        project_id=project_id,
+        status=status,
+        review_status=review_status,
+        page=page,
+        page_size=page_size,
+    )
+    return ListVideosResponse(
+        items=[_to_video_response(item, storage) for item in items],
+        total=total,
+        page=page,
+        page_size=page_size,
+    )
+
+
+@router.get("/videos/{video_id}", response_model=VideoItemResponse)
+def get_video(
+    video_id: str,
+    repo=Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """获取单个成片详情。"""
+    use_case = GetGeneratedVideoUseCase(repo)
+    item = use_case.execute(video_id)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Video not found")
+    return _to_video_response(item, storage)
+
+
+@router.patch("/videos/{video_id}/review", response_model=VideoItemResponse)
+def update_video_review_status(
+    video_id: str,
+    request: UpdateVideoReviewRequest,
+    repo=Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """更新成片复核状态：pending_review / approved / rejected。"""
+    use_case = UpdateVideoReviewStatusUseCase(repo)
+    item = use_case.execute(video_id, request.review_status)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Video not found")
+    logger.info("Video %s review status updated to %s by user %s", video_id, request.review_status, current_user.user_id)
+    return _to_video_response(item, storage)
+
+
+@router.post("/videos/batch-download", response_model=BatchDownloadResponse)
+def batch_download_videos(
+    request: BatchDownloadRequest,
+    repo=Depends(get_generated_video_repository),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """批量下载成片，异步打包 zip。
+    
+    传入 video_ids 列表，创建一个批量下载任务，任务完成后返回 zip 下载链接。
+    """
+    if not request.video_ids:
+        raise HTTPException(status_code=400, detail="video_ids cannot be empty")
+    if len(request.video_ids) > 50:
+        raise HTTPException(status_code=400, detail="Maximum 50 videos per batch download")
+
+    # 校验视频都存在
+    use_case = GetVideosByIdsUseCase(repo)
+    videos = use_case.execute(request.video_ids)
+    if len(videos) != len(request.video_ids):
+        raise HTTPException(status_code=404, detail="Some videos not found")
+
+    # 发送 celery 任务
+    task = celery_app.send_task(
+        "worker.batch_download_videos",
+        args=[request.video_ids, current_user.user_id],
+    )
+
+    logger.info("Batch download job created: %s, videos=%d", task.id, len(request.video_ids))
+    return BatchDownloadResponse(job_id=task.id, status="pending")
+
+
+@router.get("/videos/batch-download/{job_id}", response_model=BatchDownloadResponse)
+def get_batch_download_status(
+    job_id: str,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """查询批量下载任务状态。"""
+    from celery.result import AsyncResult
+
+    task = AsyncResult(job_id, app=celery_app)
+    
+    status_map = {
+        "PENDING": "pending",
+        "STARTED": "running",
+        "SUCCESS": "success",
+        "FAILURE": "failed",
+        "RETRY": "pending",
+        "REVOKED": "cancelled",
+    }
+    api_status = status_map.get(task.state, "pending")
+
+    download_url = None
+    if task.state == "SUCCESS" and task.result:
+        if isinstance(task.result, dict):
+            download_url = task.result.get("download_url")
+        elif isinstance(task.result, str):
+            download_url = task.result
+
+    return BatchDownloadResponse(
+        job_id=job_id,
+        status=api_status,
+        download_url=download_url,
+    )