fix: 修复flake8 F811/F541问题

- 成片列表分页接口（支持按项目/状态/复核状态筛选）
- 封面图自动生成（视频生成后ffmpeg抽首帧+OSS上传）
- 复核状态流转接口（pending_review/approved/rejected）
- 批量下载异步任务（打包zip+OSS上传+状态查询）
- 18个新增单测全绿，20个已有单测保持通过

新增文件:
- apps/api/app/api/routes/videos.py - 成片中心API
- apps/api/app/schemas/video_center.py - Schema定义
- apps/worker/video_processing/thumbnail_generator.py - 缩略图生成
- apps/worker/worker_app/tasks/batch_download.py - 批量下载任务
- tests/unit/test_video_center_backend.py - 单元测试

修改文件:
- packages/ports/generated_video_repository.py - 新增4个接口方法
- packages/adapters/sqlalchemy_impl/generated_video_repository.py - 实现4个方法
- packages/application/generated_videos.py - 新增3个UseCase
- packages/application/__init__.py - 导出新类
- apps/api/app/api/router.py - 注册videos路由
- apps/worker/video_processing/dedup_helpers.py - 集成缩略图生成
- apps/worker/worker_app/celery_app.py - 注册batch_download任务

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,
+    )

apps/api/app/schemas/video_center.py

@@ -0,0 +1,45 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+VideoReviewStatus = Literal["pending_review", "approved", "rejected"]
+
+
+class VideoItemResponse(BaseModel):
+    id: str
+    project_id: str
+    generation_task_id: str
+    name: str
+    file_url: str
+    file_size: int
+    duration: float
+    thumbnail_url: str | None = None
+    width: int
+    height: int
+    fps: float
+    status: str = "completed"
+    review_status: str = "pending_review"
+    generation_params: dict = Field(default_factory=dict)
+    download_url: str | None = None
+    generated_at: str = ""
+
+
+class ListVideosResponse(BaseModel):
+    items: list[VideoItemResponse]
+    total: int
+    page: int
+    page_size: int
+
+
+class UpdateVideoReviewRequest(BaseModel):
+    review_status: VideoReviewStatus
+
+
+class BatchDownloadRequest(BaseModel):
+    video_ids: list[str]
+
+
+class BatchDownloadResponse(BaseModel):
+    job_id: str
+    status: str = "pending"
+    download_url: str | None = None

apps/worker/video_processing/dedup_helpers.py

@@ -75,6 +75,19 @@ def create_video_record_and_dedup(
         video_repo = SQLAlchemyGeneratedVideoRepository(session)
         video_repo.create(generated_video)
 
+        # 生成封面缩略图
+        thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg"
+        try:
+            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+
+            thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key)
+            if thumbnail_url:
+                generated_video.thumbnail_url = thumbnail_url
+                video_repo.update_thumbnail(video_id, thumbnail_url)
+                logger.info("Thumbnail generated for video %s: %s", video_id, thumbnail_url)
+        except Exception as thumb_err:
+            logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err)
+
         # 计算视频指纹
         deduplicator = VideoDeduplicator()
         try:

apps/worker/video_processing/thumbnail_generator.py

@@ -0,0 +1,123 @@
+"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+def extract_first_frame(
+    video_path: str,
+    output_path: str | None = None,
+    *,
+    width: int = 640,
+    height: int = -1,
+    timeout: int = 30,
+) -> str:
+    """抽取视频第一帧作为封面图。
+
+    Args:
+        video_path: 视频文件路径
+        output_path: 输出图片路径，不传则用临时文件
+        width: 输出宽度（默认 640，-1 表示按比例缩放）
+        height: 输出高度（默认 -1，按比例缩放）
+        timeout: 超时时间（秒）
+
+    Returns:
+        生成的缩略图文件路径
+
+    Raises:
+        subprocess.CalledProcessError: ffmpeg 执行失败
+    """
+    from video_processing.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
+
+    if output_path is None:
+        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+        tmp.close()
+        output_path = tmp.name
+
+    # -ss 00:00:01 取第1秒帧（避免首帧黑屏）
+    # -vframes 1 只取一帧
+    # -q:v 2 jpeg 高质量
+    scale_filter = f"scale={width}:{height}:force_original_aspect_ratio=decrease"
+    cmd = [
+        FFMPEG_BIN,
+        "-y",
+        "-i",
+        video_path,
+        "-ss",
+        "00:00:01",
+        "-vframes",
+        "1",
+        "-vf",
+        scale_filter,
+        "-q:v",
+        "2",
+        output_path,
+    ]
+
+    try:
+        run_ffmpeg(cmd, capture_output=True, timeout=timeout)
+    except Exception:
+        # 短视频可能没有第1秒，退回到第0帧
+        cmd2 = [
+            FFMPEG_BIN,
+            "-y",
+            "-i",
+            video_path,
+            "-ss",
+            "00:00:00",
+            "-vframes",
+            "1",
+            "-vf",
+            scale_filter,
+            "-q:v",
+            "2",
+            output_path,
+        ]
+        run_ffmpeg(cmd2, capture_output=True, timeout=timeout)
+
+    if not Path(output_path).exists() or Path(output_path).stat().st_size == 0:
+        raise RuntimeError(f"Thumbnail generation failed: {output_path}")
+
+    return output_path
+
+
+def generate_and_upload_thumbnail(
+    video_path: str,
+    storage_key: str,
+) -> str | None:
+    """生成缩略图并上传到 OSS，返回 URL。
+
+    Args:
+        video_path: 本地视频路径
+        storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg）
+
+    Returns:
+        上传成功返回 URL，失败返回 None
+    """
+    thumbnail_path = None
+    try:
+        thumbnail_path = extract_first_frame(video_path)
+    except Exception as e:
+        logger.warning("Failed to extract thumbnail from %s: %s", video_path, e)
+        return None
+
+    try:
+        from video_processing.oss_helpers import upload_to_oss
+
+        url = upload_to_oss(thumbnail_path, storage_key)
+        return url
+    except Exception as e:
+        logger.warning("Failed to upload thumbnail to OSS: %s", e)
+        return None
+    finally:
+        # 清理临时文件
+        if thumbnail_path:
+            try:
+                Path(thumbnail_path).unlink(missing_ok=True)
+            except Exception:
+                pass

apps/worker/worker_app/celery_app.py

@@ -16,5 +16,6 @@ celery_app.conf.imports = (
     "worker_app.tasks.tts_synthesis",
     "worker_app.tasks.edit_plan_generation",
     "worker_app.tasks.compose_video",
+    "worker_app.tasks.batch_download",
     "apps.worker.video_processing.dedup",
 )

apps/worker/worker_app/tasks/batch_download.py

@@ -0,0 +1,112 @@
+"""批量下载任务 — 将多个成片打包为 zip 上传到 OSS。"""
+
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+import uuid
+import zipfile
+from pathlib import Path
+
+from worker_app.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, name="worker.batch_download_videos", max_retries=1)
+def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict:
+    """批量下载视频并打包为 zip。
+
+    Args:
+        video_ids: 视频 ID 列表
+        user_id: 发起用户 ID
+
+    Returns:
+        {"download_url": "...", "file_count": N, "total_size": total_bytes}
+    """
+    from video_processing.oss_helpers import download_asset, upload_to_oss
+    from worker_app.db import SessionLocal
+
+    from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+        SQLAlchemyGeneratedVideoRepository,
+    )
+
+    session = SessionLocal()
+    try:
+        repo = SQLAlchemyGeneratedVideoRepository(session)
+        videos = repo.get_by_ids(video_ids)
+    finally:
+        session.close()
+
+    if not videos:
+        raise ValueError("No videos found for batch download")
+
+    # 创建临时工作目录
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = Path(tmpdir)
+        zip_filename = f"videos-{len(videos)}-{video_ids[0][:8]}.zip"
+        zip_path = tmpdir_path / zip_filename
+
+        # 逐个下载视频并加入 zip
+        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_STORED) as zf:
+            for idx, video in enumerate(videos, 1):
+                logger.info("Batch download: downloading %d/%d %s", idx, len(videos), video.id)
+                try:
+                    # 下载视频到临时文件
+                    local_name = f"{idx:03d}_{video.name}"
+                    local_path = tmpdir_path / local_name
+
+                    # 使用 oss_helpers 的 download_asset，或者直接从 URL 下载
+                    if video.file_url:
+                        _download_video_to_file(video.file_url, str(local_path))
+
+                    if local_path.exists() and local_path.stat().st_size > 0:
+                        zf.write(str(local_path), arcname=local_name)
+                        local_path.unlink(missing_ok=True)
+                    else:
+                        logger.warning("Video %s download failed, skipping", video.id)
+                except Exception as e:
+                    logger.warning("Failed to download video %s: %s", video.id, e)
+                    continue
+
+        # 上传 zip 到 OSS
+        if not zip_path.exists() or zip_path.stat().st_size == 0:
+            raise RuntimeError("Batch download zip file is empty")
+        zip_storage_key = f"batch-downloads/{uuid.uuid4().hex}/{zip_filename}"
+        download_url = upload_to_oss(str(zip_path), zip_storage_key)
+
+        total_size = zip_path.stat().st_size
+        file_count = len(zipfile.ZipFile(str(zip_path), "r").namelist())
+
+        logger.info(
+            "Batch download complete: %d files, %d bytes, url=%s",
+            file_count,
+            total_size,
+            download_url,
+        )
+
+        return {
+            "download_url": download_url,
+            "file_count": file_count,
+            "total_size": total_size,
+            "video_count": len(videos),
+        }
+
+
+def _download_video_to_file(url: str, dest_path: str) -> None:
+    """下载视频文件到本地路径。优先用 OSS SDK 走内网，回退到 HTTP 下载。"""
+    from video_processing.oss_helpers import download_asset
+
+    try:
+        # 尝试走 OSS 下载（如果是 OSS URL 的话）
+        success = download_asset(url, dest_path)
+        if success:
+            return
+    except Exception:
+        pass
+
+    # 回退到 HTTP 下载
+    import urllib.request
+
+    urllib.request.urlretrieve(url, dest_path)  # nosec B310

packages/adapters/sqlalchemy_impl/generated_video_repository.py

@@ -100,6 +100,63 @@ class SQLAlchemyGeneratedVideoRepository:
         )
         return [self._to_domain(model) for model in models]
 
+    def list_paginated(
+        self,
+        *,
+        project_id: str | None = None,
+        status: str | None = None,
+        review_status: str | None = None,
+        page: int = 1,
+        page_size: int = 20,
+    ) -> tuple[list[GeneratedVideo], int]:
+        """分页查询成片列表，支持按项目、状态、复核状态筛选。"""
+        query = self.session.query(GeneratedVideoModel)
+
+        if project_id:
+            query = query.filter(GeneratedVideoModel.project_id == project_id)
+        if status:
+            query = query.filter(GeneratedVideoModel.status == status)
+        if review_status:
+            query = query.filter(GeneratedVideoModel.review_status == review_status)
+
+        total = query.count()
+
+        models = (
+            query.order_by(GeneratedVideoModel.generated_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+            .all()
+        )
+
+        return [self._to_domain(model) for model in models], total
+
+    def update_review_status(self, video_id: str, review_status: str) -> GeneratedVideo | None:
+        """更新成片复核状态。"""
+        model = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id == video_id).first()
+        if model is None:
+            return None
+        model.review_status = review_status
+        self.session.add(model)
+        self.session.commit()
+        return self._to_domain(model)
+
+    def update_thumbnail(self, video_id: str, thumbnail_url: str) -> bool:
+        """更新成片封面图URL。"""
+        model = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id == video_id).first()
+        if model is None:
+            return False
+        model.thumbnail_url = thumbnail_url
+        self.session.add(model)
+        self.session.commit()
+        return True
+
+    def get_by_ids(self, video_ids: list[str]) -> list[GeneratedVideo]:
+        """批量获取成片记录。"""
+        if not video_ids:
+            return []
+        models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id.in_(video_ids)).all()
+        return [self._to_domain(model) for model in models]
+
     @staticmethod
     def _to_domain(model: GeneratedVideoModel) -> GeneratedVideo:
         return GeneratedVideo(

packages/application/__init__.py

@@ -21,8 +21,11 @@ from .duplication import (
 from .generated_videos import (
     GetGeneratedVideoDownloadUrlUseCase,
     GetGeneratedVideoUseCase,
+    GetVideosByIdsUseCase,
     ListGeneratedVideosByTaskUseCase,
+    ListGeneratedVideosPaginatedUseCase,
     ListGeneratedVideosUseCase,
+    UpdateVideoReviewStatusUseCase,
 )
 from .generation_tasks import (
     CreateGenerationTaskCommand,
@@ -76,6 +79,7 @@ __all__ = [
     "GetDuplicationDetailUseCase",
     "GetGeneratedVideoDownloadUrlUseCase",
     "GetGeneratedVideoUseCase",
+    "GetVideosByIdsUseCase",
     "GetJobStatisticsUseCase",
     "GetJobUseCase",
     "GetProjectUseCase",
@@ -83,6 +87,7 @@ __all__ = [
     "ListAssetsUseCase",
     "ListDuplicationRecordsUseCase",
     "ListGeneratedVideosByTaskUseCase",
+    "ListGeneratedVideosPaginatedUseCase",
     "ListGeneratedVideosUseCase",
     "ListJobsUseCase",
     "ListProjectsUseCase",
@@ -95,6 +100,7 @@ __all__ = [
     "SubmitJobUseCase",
     "UpdateJobProgressCommand",
     "UpdateJobProgressUseCase",
+    "UpdateVideoReviewStatusUseCase",
     "UploadForDuplicationCommand",
     "UploadForDuplicationUseCase",
 ]

packages/application/generated_videos.py

@@ -14,6 +14,32 @@ class ListGeneratedVideosUseCase:
         return self.generated_video_repository.list_by_project(project_id.strip())
 
 
+class ListGeneratedVideosPaginatedUseCase:
+    def __init__(self, generated_video_repository: GeneratedVideoRepository):
+        self.generated_video_repository = generated_video_repository
+
+    def execute(
+        self,
+        *,
+        project_id: str | None = None,
+        status: str | None = None,
+        review_status: str | None = None,
+        page: int = 1,
+        page_size: int = 20,
+    ) -> tuple[list[GeneratedVideo], int]:
+        if page < 1:
+            page = 1
+        if page_size < 1 or page_size > 100:
+            page_size = 20
+        return self.generated_video_repository.list_paginated(
+            project_id=project_id,
+            status=status,
+            review_status=review_status,
+            page=page,
+            page_size=page_size,
+        )
+
+
 class GetGeneratedVideoUseCase:
     def __init__(self, generated_video_repository: GeneratedVideoRepository):
         self.generated_video_repository = generated_video_repository
@@ -41,3 +67,23 @@ class GetGeneratedVideoDownloadUrlUseCase:
         if item is None:
             return None
         return item.file_url
+
+
+class UpdateVideoReviewStatusUseCase:
+    def __init__(self, generated_video_repository: GeneratedVideoRepository):
+        self.generated_video_repository = generated_video_repository
+
+    def execute(self, video_id: str, review_status: str) -> GeneratedVideo | None:
+        if not video_id.strip():
+            raise ValueError("video_id 不能为空")
+        if review_status not in ("pending_review", "approved", "rejected"):
+            raise ValueError(f"无效的 review_status: {review_status}")
+        return self.generated_video_repository.update_review_status(video_id.strip(), review_status)
+
+
+class GetVideosByIdsUseCase:
+    def __init__(self, generated_video_repository: GeneratedVideoRepository):
+        self.generated_video_repository = generated_video_repository
+
+    def execute(self, video_ids: list[str]) -> list[GeneratedVideo]:
+        return self.generated_video_repository.get_by_ids(video_ids)

packages/ports/generated_video_repository.py

@@ -15,3 +15,19 @@ class GeneratedVideoRepository(Protocol):
     def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]: ...
 
     def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]: ...
+
+    def list_paginated(
+        self,
+        *,
+        project_id: str | None = None,
+        status: str | None = None,
+        review_status: str | None = None,
+        page: int = 1,
+        page_size: int = 20,
+    ) -> tuple[list[GeneratedVideo], int]: ...
+
+    def update_review_status(self, video_id: str, review_status: str) -> GeneratedVideo | None: ...
+
+    def update_thumbnail(self, video_id: str, thumbnail_url: str) -> bool: ...
+
+    def get_by_ids(self, video_ids: list[str]) -> list[GeneratedVideo]: ...

tests/render_compare/README_assets.md

@@ -0,0 +1,201 @@
+# 测试素材创建工具
+
+`create_test_asset.py` 是一个自动化测试辅助工具，用于快速创建 `ready` 状态的视频素材，跳过正常的上传和转码流程，直接指定已存在于 OSS 的文件来生成可用素材。
+
+## 适用场景
+
+- 渲染对比测试：快速创建测试素材用于生成任务
+- 性能测试：批量创建素材模拟真实场景
+- 开发调试：无需真实上传文件即可测试素材相关功能
+
+## 前置条件
+
+1. API 服务正在运行
+2. 有有效的登录 token
+3. 指定的 `storage_key` 对应的文件已存在于 OSS 中
+4. 用户对指定项目有访问权限
+
+## 使用方法
+
+### 基本用法
+
+```bash
+# 设置环境变量（可选）
+export API_BASE_URL=http://localhost:8000
+export API_TOKEN=your_token_here
+
+# 创建测试素材
+python tests/render_compare/create_test_asset.py \
+    --project-id proj_xxx \
+    --name "测试素材-30s" \
+    --storage-key "assets/test/sample_30s.mp4" \
+    --duration 30 \
+    --file-size 10485760
+```
+
+### 完整参数示例
+
+```bash
+python tests/render_compare/create_test_asset.py \
+    --base-url http://localhost:8000 \
+    --token eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9... \
+    --project-id proj_a1b2c3d4e5f6 \
+    --name "1080p测试视频-60s" \
+    --storage-key "test-assets/1080p_60fps_60s.mp4" \
+    --duration 60 \
+    --width 1920 \
+    --height 1080 \
+    --fps 60 \
+    --mime-type "video/mp4" \
+    --file-size 52428800 \
+    --codec "h264" \
+    --kind video
+```
+
+### 在脚本中捕获 asset_id
+
+```bash
+# 最后一行输出为 asset_id，方便脚本捕获
+ASSET_ID=$(python tests/render_compare/create_test_asset.py \
+    --project-id proj_xxx \
+    --name "测试素材" \
+    --storage-key "test/video.mp4" \
+    2>&1 | tail -1)
+
+echo "创建的素材 ID: $ASSET_ID"
+```
+
+## 参数说明
+
+| 参数 | 环境变量 | 必填 | 默认值 | 说明 |
+|------|----------|------|--------|------|
+| `--base-url` | `API_BASE_URL` | 否 | `http://localhost:8000` | API 服务地址 |
+| `--token` | `API_TOKEN` | 是 | - | 登录认证 token |
+| `--project-id` | - | 是 | - | 项目 ID |
+| `--name` | - | 是 | - | 素材名称 |
+| `--storage-key` | - | 是 | - | OSS storage_key（文件需已存在） |
+| `--duration` | - | 否 | `30` | 视频时长（秒） |
+| `--width` | - | 否 | `1280` | 视频宽度 |
+| `--height` | - | 否 | `720` | 视频高度 |
+| `--fps` | - | 否 | `25` | 帧率 |
+| `--mime-type` | - | 否 | `video/mp4` | MIME 类型 |
+| `--file-size` | - | 否 | `0` | 文件大小（字节） |
+| `--codec` | - | 否 | - | 视频编码 |
+| `--kind` | - | 否 | `video` | 素材库类型 (video/voice/image) |
+
+## API 调用流程
+
+脚本会依次调用以下 API：
+
+### 1. 确保默认素材库存在
+
+```
+POST /api/v1/asset-libraries/ensure-default
+Content-Type: application/json
+Authorization: Bearer {token}
+
+{
+  "project_id": "proj_xxx",
+  "kind": "video"
+}
+```
+
+- 如果项目下已有对应类型的素材库，直接返回第一个
+- 如果不存在，自动创建默认名称的素材库
+
+### 2. 创建素材
+
+```
+POST /api/v1/assets
+Content-Type: application/json
+Authorization: Bearer {token}
+
+{
+  "project_id": "proj_xxx",
+  "library_id": "lib_xxx",
+  "name": "测试素材",
+  "storage_key": "assets/test/video.mp4",
+  "mime_type": "video/mp4",
+  "file_size": 10485760,
+  "duration": 30,
+  "width": 1280,
+  "height": 720,
+  "fps": 25,
+  "status": "ready",
+  "classification_status": "pending",
+  "metadata": {}
+}
+```
+
+**关键点**：
+- `status: "ready"` 直接跳过转码流程，立即可用
+- `storage_key` 必须对应 OSS 中真实存在的文件，否则播放会失败
+- `uploaded_by_user_id` 由 API 自动设置为当前登录用户
+
+## 输出说明
+
+脚本输出分为三部分：
+
+1. **参数回显**：确认输入的参数是否正确
+2. **执行日志**：显示每一步的执行情况
+3. **结果输出**：
+   - 素材详细信息（asset_id、状态、文件 URL 等）
+   - 调用示例
+   - 最后一行为纯 asset_id，方便脚本捕获
+
+## 错误排查
+
+### 常见错误
+
+#### 401 Unauthorized
+- 检查 token 是否正确且未过期
+- 确认 Authorization header 格式为 `Bearer {token}`
+
+#### 403 Forbidden
+- 确认用户对该项目有访问权限
+- 检查项目 ID 是否正确
+
+#### 404 Project not found
+- 项目 ID 错误或项目不存在
+
+#### 422 Validation Error
+- 检查参数格式是否正确
+- 查看响应中的 detail 字段了解具体错误
+
+### 调试技巧
+
+脚本在 API 失败时会打印完整的响应内容，包括：
+- HTTP 状态码
+- 错误原因
+- 响应 body 详情
+
+如果遇到问题，请检查：
+1. API 服务是否正常运行
+2. base-url 是否正确（注意端口号）
+3. token 是否有效
+4. storage_key 对应的文件是否存在于 OSS
+
+## 与渲染对比测试配合使用
+
+```bash
+# 1. 创建测试素材
+ASSET_ID=$(python tests/render_compare/create_test_asset.py \
+    --project-id proj_xxx \
+    --name "渲染对比测试素材" \
+    --storage-key "test-assets/base_1080p_30s.mp4" \
+    --duration 30 --width 1920 --height 1080 \
+    2>&1 | tail -1)
+
+# 2. 使用该素材运行渲染对比测试
+python tests/render_compare/runner.py \
+    --project-id proj_xxx \
+    --asset-id $ASSET_ID \
+    --scenario quality_test
+```
+
+## 注意事项
+
+1. **storage_key 必须真实存在**：脚本不会上传文件，只是创建数据库记录。如果 OSS 中没有对应文件，素材虽然状态是 ready，但无法正常播放。
+2. **素材参数要准确**：duration、width、height、fps 等参数应与实际文件一致，否则可能导致后续渲染或分析出现偏差。
+3. **权限检查**：确保 token 对应用户有项目的素材创建权限。
+4. **清理测试数据**：测试完成后记得清理不需要的测试素材，避免占用资源。

tests/render_compare/create_test_asset.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""
+自动化测试用：一键创建 ready 状态的视频素材。
+
+跳过转码和上传流程，直接指定 storage_key 创建可用素材，
+供渲染对比测试等场景快速生成测试素材。
+
+用法示例：
+    python create_test_asset.py \
+        --base-url http://localhost:8000 \
+        --token xxx \
+        --project-id proj_xxx \
+        --name "测试素材-30s" \
+        --storage-key "assets/test/video.mp4" \
+        --duration 30 \
+        --file-size 10485760
+"""
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="创建 ready 状态的测试素材（跳过转码上传）")
+    parser.add_argument(
+        "--base-url",
+        default=os.environ.get("API_BASE_URL", "http://localhost:8000"),
+        help="API 地址，默认 http://localhost:8000（或环境变量 API_BASE_URL）",
+    )
+    parser.add_argument(
+        "--token",
+        default=os.environ.get("API_TOKEN", ""),
+        help="登录 token（或环境变量 API_TOKEN）",
+    )
+    parser.add_argument(
+        "--project-id",
+        required=True,
+        help="项目 ID",
+    )
+    parser.add_argument(
+        "--name",
+        required=True,
+        help="素材名称",
+    )
+    parser.add_argument(
+        "--storage-key",
+        required=True,
+        help="OSS storage_key（文件必须已存在于 OSS）",
+    )
+    parser.add_argument(
+        "--duration",
+        type=float,
+        default=30.0,
+        help="视频时长（秒），默认 30",
+    )
+    parser.add_argument(
+        "--width",
+        type=int,
+        default=1280,
+        help="视频宽度，默认 1280",
+    )
+    parser.add_argument(
+        "--height",
+        type=int,
+        default=720,
+        help="视频高度，默认 720",
+    )
+    parser.add_argument(
+        "--fps",
+        type=float,
+        default=25.0,
+        help="帧率，默认 25",
+    )
+    parser.add_argument(
+        "--mime-type",
+        default="video/mp4",
+        help="MIME 类型，默认 video/mp4",
+    )
+    parser.add_argument(
+        "--file-size",
+        type=int,
+        default=0,
+        help="文件大小（字节），默认 0",
+    )
+    parser.add_argument(
+        "--codec",
+        default=None,
+        help="视频编码，可选",
+    )
+    parser.add_argument(
+        "--kind",
+        default="video",
+        choices=["video", "voice", "image"],
+        help="素材库类型，默认 video",
+    )
+    return parser.parse_args()
+
+
+def api_request(base_url: str, token: str, method: str, path: str, body: dict | None = None) -> dict:
+    """发送 API 请求，返回 JSON 响应。"""
+    url = f"{base_url.rstrip('/')}{path}"
+    data = json.dumps(body).encode("utf-8") if body else None
+
+    headers = {
+        "Content-Type": "application/json",
+    }
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    req = urllib.request.Request(url, data=data, method=method, headers=headers)
+
+    try:
+        with urllib.request.urlopen(req) as resp:
+            resp_body = resp.read().decode("utf-8")
+            return json.loads(resp_body) if resp_body else {}
+    except urllib.error.HTTPError as e:
+        error_body = e.read().decode("utf-8", errors="replace")
+        print(f"[ERROR] API 请求失败: {method} {url}", file=sys.stderr)
+        print(f"  HTTP {e.code}: {e.reason}", file=sys.stderr)
+        print(f"  响应内容: {error_body}", file=sys.stderr)
+        sys.exit(1)
+    except urllib.error.URLError as e:
+        print(f"[ERROR] 网络错误: {method} {url}", file=sys.stderr)
+        print(f"  原因: {e.reason}", file=sys.stderr)
+        sys.exit(1)
+
+
+def ensure_default_library(base_url: str, token: str, project_id: str, kind: str) -> str:
+    """确保项目有默认素材库，返回 library_id。"""
+    print(f"[1/2] 确保默认 {kind} 素材库存在...")
+    result = api_request(
+        base_url,
+        token,
+        "POST",
+        "/api/v1/asset-libraries/ensure-default",
+        body={"project_id": project_id, "kind": kind},
+    )
+    library_id = result.get("id")
+    library_name = result.get("name")
+    print(f"      素材库: {library_name} (id: {library_id})")
+    return library_id
+
+
+def create_asset(
+    base_url: str,
+    token: str,
+    project_id: str,
+    library_id: str,
+    name: str,
+    storage_key: str,
+    mime_type: str,
+    file_size: int,
+    duration: float,
+    width: int,
+    height: int,
+    fps: float,
+    codec: str | None,
+) -> dict:
+    """创建 ready 状态的素材。"""
+    print("[2/2] 创建 ready 状态素材...")
+    body = {
+        "project_id": project_id,
+        "library_id": library_id,
+        "name": name,
+        "storage_key": storage_key,
+        "mime_type": mime_type,
+        "file_size": file_size,
+        "duration": duration,
+        "width": width,
+        "height": height,
+        "fps": fps,
+        "status": "ready",
+        "classification_status": "pending",
+        "metadata": {},
+    }
+    if codec:
+        body["codec"] = codec
+
+    result = api_request(
+        base_url,
+        token,
+        "POST",
+        "/api/v1/assets",
+        body=body,
+    )
+    return result
+
+
+def main():
+    args = parse_args()
+
+    if not args.token:
+        print("[ERROR] 请通过 --token 参数或 API_TOKEN 环境变量提供登录 token", file=sys.stderr)
+        sys.exit(1)
+
+    print("=" * 60)
+    print("创建测试素材工具")
+    print("=" * 60)
+    print(f"  API 地址: {args.base_url}")
+    print(f"  项目 ID:  {args.project_id}")
+    print(f"  素材名称: {args.name}")
+    print(f"  storage_key: {args.storage_key}")
+    print(f"  分辨率:   {args.width}x{args.height} @ {args.fps}fps")
+    print(f"  时长:     {args.duration}s")
+    print(f"  文件大小: {args.file_size} bytes")
+    print("=" * 60)
+    print()
+
+    # Step 1: 确保默认素材库存在
+    library_id = ensure_default_library(args.base_url, args.token, args.project_id, args.kind)
+
+    # Step 2: 创建素材
+    asset = create_asset(
+        base_url=args.base_url,
+        token=args.token,
+        project_id=args.project_id,
+        library_id=library_id,
+        name=args.name,
+        storage_key=args.storage_key,
+        mime_type=args.mime_type,
+        file_size=args.file_size,
+        duration=args.duration,
+        width=args.width,
+        height=args.height,
+        fps=args.fps,
+        codec=args.codec,
+    )
+
+    asset_id = asset.get("id")
+    print()
+    print("=" * 60)
+    print("✓ 素材创建成功！")
+    print("=" * 60)
+    print(f"  asset_id:   {asset_id}")
+    print(f"  状态:       {asset.get('status')}")
+    print(f"  素材库 ID:  {asset.get('library_id')}")
+    print(f"  文件 URL:   {asset.get('file_url', 'N/A')}")
+    print("=" * 60)
+    print()
+    print("调用示例：")
+    print(f"  export ASSET_ID={asset_id}")
+    print("  # 在生成任务中使用:")
+    print("  #   --asset-id $ASSET_ID")
+    print()
+
+    # 输出 asset_id 到 stdout（方便脚本捕获）
+    print(asset_id)
+
+
+if __name__ == "__main__":
+    main()

tests/unit/test_video_center_backend.py

@@ -0,0 +1,216 @@
+"""成片中心新功能单元测试。
+
+覆盖：分页列表、复核状态更新、缩略图更新、批量获取、use case。
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.application.generated_videos import (
+    GetVideosByIdsUseCase,
+    ListGeneratedVideosPaginatedUseCase,
+    UpdateVideoReviewStatusUseCase,
+)
+from packages.domain import GeneratedVideo
+
+
+def _repository():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return SQLAlchemyGeneratedVideoRepository(session)
+
+
+def _create_video(repo, project_id="proj-1", status="completed", review_status="pending_review", idx=1):
+    video = GeneratedVideo.create(
+        project_id=project_id,
+        generation_task_id=f"task-{idx}",
+        name=f"video-{idx}.mp4",
+        file_url=f"generated/video-{idx}.mp4",
+        file_size=1024 * idx,
+        duration=10.0 * idx,
+        width=1280,
+        height=720,
+        fps=25.0,
+    )
+    video.status = status
+    video.review_status = review_status
+    repo.create(video)
+    return video
+
+
+class TestGeneratedVideoRepository:
+    """GeneratedVideoRepository 新方法测试。"""
+
+    def test_list_paginated_default(self):
+        repo = _repository()
+        for i in range(5):
+            _create_video(repo, idx=i)
+
+        items, total = repo.list_paginated(page=1, page_size=3)
+        assert total == 5
+        assert len(items) == 3
+        # 按 generated_at 倒序，最新的在前
+        assert items[0].name == "video-4.mp4"
+
+    def test_list_paginated_by_project(self):
+        repo = _repository()
+        _create_video(repo, project_id="proj-a", idx=1)
+        _create_video(repo, project_id="proj-a", idx=2)
+        _create_video(repo, project_id="proj-b", idx=3)
+
+        items, total = repo.list_paginated(project_id="proj-a")
+        assert total == 2
+        assert all(i.project_id == "proj-a" for i in items)
+
+    def test_list_paginated_by_status(self):
+        repo = _repository()
+        _create_video(repo, status="completed", idx=1)
+        _create_video(repo, status="completed", idx=2)
+        _create_video(repo, status="failed", idx=3)
+
+        items, total = repo.list_paginated(status="completed")
+        assert total == 2
+        assert all(i.status == "completed" for i in items)
+
+    def test_list_paginated_by_review_status(self):
+        repo = _repository()
+        _create_video(repo, review_status="pending_review", idx=1)
+        _create_video(repo, review_status="approved", idx=2)
+        _create_video(repo, review_status="rejected", idx=3)
+
+        items, total = repo.list_paginated(review_status="approved")
+        assert total == 1
+        assert items[0].review_status == "approved"
+
+    def test_list_paginated_multi_filter(self):
+        repo = _repository()
+        _create_video(repo, project_id="p1", status="completed", review_status="approved", idx=1)
+        _create_video(repo, project_id="p1", status="completed", review_status="pending_review", idx=2)
+        _create_video(repo, project_id="p2", status="completed", review_status="approved", idx=3)
+
+        items, total = repo.list_paginated(project_id="p1", review_status="approved")
+        assert total == 1
+        assert items[0].project_id == "p1"
+        assert items[0].review_status == "approved"
+
+    def test_update_review_status(self):
+        repo = _repository()
+        video = _create_video(repo, idx=1)
+
+        result = repo.update_review_status(video.id, "approved")
+        assert result is not None
+        assert result.review_status == "approved"
+
+        # 验证持久化
+        saved = repo.get(video.id)
+        assert saved.review_status == "approved"
+
+    def test_update_review_status_not_found(self):
+        repo = _repository()
+        result = repo.update_review_status("nonexistent", "approved")
+        assert result is None
+
+    def test_update_thumbnail(self):
+        repo = _repository()
+        video = _create_video(repo, idx=1)
+        assert video.thumbnail_url is None
+
+        ok = repo.update_thumbnail(video.id, "https://oss/thumb.jpg")
+        assert ok is True
+
+        saved = repo.get(video.id)
+        assert saved.thumbnail_url == "https://oss/thumb.jpg"
+
+    def test_update_thumbnail_not_found(self):
+        repo = _repository()
+        ok = repo.update_thumbnail("nonexistent", "https://oss/thumb.jpg")
+        assert ok is False
+
+    def test_get_by_ids(self):
+        repo = _repository()
+        v1 = _create_video(repo, idx=1)
+        v2 = _create_video(repo, idx=2)
+        v3 = _create_video(repo, idx=3)
+
+        result = repo.get_by_ids([v1.id, v3.id])
+        assert len(result) == 2
+        ids = {v.id for v in result}
+        assert v1.id in ids
+        assert v3.id in ids
+
+    def test_get_by_ids_empty(self):
+        repo = _repository()
+        result = repo.get_by_ids([])
+        assert result == []
+
+
+class TestGeneratedVideoUseCases:
+    """Use case 层测试。"""
+
+    def test_list_paginated_use_case(self):
+        repo = _repository()
+        for i in range(10):
+            _create_video(repo, idx=i)
+
+        use_case = ListGeneratedVideosPaginatedUseCase(repo)
+        items, total = use_case.execute(page=2, page_size=3)
+        assert total == 10
+        assert len(items) == 3
+
+    def test_list_paginated_use_case_page_clamp(self):
+        repo = _repository()
+        use_case = ListGeneratedVideosPaginatedUseCase(repo)
+        # page < 1 应该被修正为 1
+        items, total = use_case.execute(page=0, page_size=20)
+        assert total == 0
+
+    def test_list_paginated_use_case_page_size_clamp(self):
+        repo = _repository()
+        use_case = ListGeneratedVideosPaginatedUseCase(repo)
+        # page_size > 100 应该被修正为 20
+        for i in range(30):
+            _create_video(repo, idx=i)
+        items, total = use_case.execute(page=1, page_size=200)
+        assert total == 30
+        assert len(items) == 20  # clamp 到默认 20
+
+    def test_update_review_status_use_case(self):
+        repo = _repository()
+        video = _create_video(repo, idx=1)
+
+        use_case = UpdateVideoReviewStatusUseCase(repo)
+        result = use_case.execute(video.id, "approved")
+        assert result is not None
+        assert result.review_status == "approved"
+
+    def test_update_review_status_use_case_invalid_status(self):
+        repo = _repository()
+        video = _create_video(repo, idx=1)
+
+        use_case = UpdateVideoReviewStatusUseCase(repo)
+        with pytest.raises(ValueError, match="无效的 review_status"):
+            use_case.execute(video.id, "invalid_status")
+
+    def test_update_review_status_use_case_empty_id(self):
+        repo = _repository()
+        use_case = UpdateVideoReviewStatusUseCase(repo)
+        with pytest.raises(ValueError, match="video_id 不能为空"):
+            use_case.execute("", "approved")
+
+    def test_get_by_ids_use_case(self):
+        repo = _repository()
+        v1 = _create_video(repo, idx=1)
+        v2 = _create_video(repo, idx=2)
+
+        use_case = GetVideosByIdsUseCase(repo)
+        result = use_case.execute([v1.id, v2.id])
+        assert len(result) == 2