From 1f8de776e82adb2a865f8fe91a87e74829c1c9bd Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 21:30:54 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20=E6=88=90=E7=89=87=E4=B8=AD?= =?UTF-8?q?=E5=BF=83=E5=90=8E=E7=AB=AF=E5=8D=87=E7=BA=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 成片列表分页接口（支持按项目/状态/复核状态筛选） - 封面图自动生成（视频生成后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 | 5 + apps/api/app/api/routes/videos.py | 180 ++++++++++++ apps/api/app/schemas/video_center.py | 45 +++ apps/worker/video_processing/dedup_helpers.py | 13 + .../video_processing/thumbnail_generator.py | 113 ++++++++ apps/worker/worker_app/celery_app.py | 1 + .../worker/worker_app/tasks/batch_download.py | 110 ++++++++ .../generated_video_repository.py | 57 ++++ packages/application/__init__.py | 6 + packages/application/generated_videos.py | 46 ++++ packages/ports/generated_video_repository.py | 16 ++ tests/render_compare/README_assets.md | 201 ++++++++++++++ tests/render_compare/create_test_asset.py | 259 ++++++++++++++++++ tests/unit/test_video_center_backend.py | 216 +++++++++++++++ 14 files changed, 1268 insertions(+) create mode 100755 apps/api/app/api/routes/videos.py create mode 100755 apps/api/app/schemas/video_center.py mode change 100644 => 100755 apps/worker/video_processing/dedup_helpers.py create mode 100755 apps/worker/video_processing/thumbnail_generator.py create mode 100755 apps/worker/worker_app/tasks/batch_download.py mode change 100644 => 100755 packages/adapters/sqlalchemy_impl/generated_video_repository.py mode change 100644 => 100755 packages/application/generated_videos.py mode change 100644 => 100755 packages/ports/generated_video_repository.py create mode 100755 tests/render_compare/README_assets.md create mode 100755 tests/render_compare/create_test_asset.py create mode 100755 tests/unit/test_video_center_backend.py diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index c0711f12d..9bbabcd16 100755 --- a/apps/api/app/api/router.py +++ b/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", diff --git a/apps/api/app/api/routes/videos.py b/apps/api/app/api/routes/videos.py new file mode 100755 index 000000000..1227ffb0f --- /dev/null +++ b/apps/api/app/api/routes/videos.py @@ -0,0 +1,180 @@ +import logging +import uuid + +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 app.api.routes._helpers import check_project_access + +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, + ) diff --git a/apps/api/app/schemas/video_center.py b/apps/api/app/schemas/video_center.py new file mode 100755 index 000000000..664769d7a --- /dev/null +++ b/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 diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py old mode 100644 new mode 100755 index 61e1ef427..0eb74e136 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/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: diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py new file mode 100755 index 000000000..1533d0e4c --- /dev/null +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -0,0 +1,113 @@ +"""视频缩略图生成工具 — 抽取首帧上传到 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 diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 352775112..7df07170b 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/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", ) diff --git a/apps/worker/worker_app/tasks/batch_download.py b/apps/worker/worker_app/tasks/batch_download.py new file mode 100755 index 000000000..919e45f0c --- /dev/null +++ b/apps/worker/worker_app/tasks/batch_download.py @@ -0,0 +1,110 @@ +"""批量下载任务 — 将多个成片打包为 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 packages.adapters.sqlalchemy_impl.generated_video_repository import ( + SQLAlchemyGeneratedVideoRepository, + ) + from worker_app.db import SessionLocal + + 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") + + import uuid + 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 diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py old mode 100644 new mode 100755 index 31620e6f6..4bef1427a --- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py +++ b/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( diff --git a/packages/application/__init__.py b/packages/application/__init__.py index e0c234713..cd2dccf4f 100755 --- a/packages/application/__init__.py +++ b/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", ] diff --git a/packages/application/generated_videos.py b/packages/application/generated_videos.py old mode 100644 new mode 100755 index 4e638346d..ac2974ed3 --- a/packages/application/generated_videos.py +++ b/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) diff --git a/packages/ports/generated_video_repository.py b/packages/ports/generated_video_repository.py old mode 100644 new mode 100755 index f9aac9c21..481c64cb8 --- a/packages/ports/generated_video_repository.py +++ b/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]: ... diff --git a/tests/render_compare/README_assets.md b/tests/render_compare/README_assets.md new file mode 100755 index 000000000..2e340e434 --- /dev/null +++ b/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. **清理测试数据**：测试完成后记得清理不需要的测试素材，避免占用资源。 diff --git a/tests/render_compare/create_test_asset.py b/tests/render_compare/create_test_asset.py new file mode 100755 index 000000000..c8a65d92e --- /dev/null +++ b/tests/render_compare/create_test_asset.py @@ -0,0 +1,259 @@ +#!/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(f"[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(f" # 在生成任务中使用:") + print(f" # --asset-id $ASSET_ID") + print() + + # 输出 asset_id 到 stdout（方便脚本捕获） + print(asset_id) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_video_center_backend.py b/tests/unit/test_video_center_backend.py new file mode 100755 index 000000000..4f4d2f606 --- /dev/null +++ b/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 -- 2.54.0 From c34cae3c21e657884d1b20a08b3d7054c523ed80 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 09:17:22 +0800 Subject: [PATCH 2/3] =?UTF-8?q?chore:=20=E4=BF=AE=E5=A4=8Dblack/isort?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/api/routes/videos.py | 3 +- .../video_processing/thumbnail_generator.py | 30 ++++++++++++------- .../worker/worker_app/tasks/batch_download.py | 9 ++++-- tests/render_compare/create_test_asset.py | 8 ++--- 4 files changed, 30 insertions(+), 20 deletions(-) diff --git a/apps/api/app/api/routes/videos.py b/apps/api/app/api/routes/videos.py index 1227ffb0f..932c1672b 100755 --- a/apps/api/app/api/routes/videos.py +++ b/apps/api/app/api/routes/videos.py @@ -1,6 +1,7 @@ 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 @@ -14,8 +15,6 @@ from app.schemas.video_center import ( ) from fastapi import APIRouter, Depends, HTTPException, Query -from app.api.routes._helpers import check_project_access - from packages.application import ( GetGeneratedVideoUseCase, GetVideosByIdsUseCase, diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 1533d0e4c..6b9f3ed4f 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -46,11 +46,16 @@ def extract_first_frame( cmd = [ FFMPEG_BIN, "-y", - "-i", video_path, - "-ss", "00:00:01", - "-vframes", "1", - "-vf", scale_filter, - "-q:v", "2", + "-i", + video_path, + "-ss", + "00:00:01", + "-vframes", + "1", + "-vf", + scale_filter, + "-q:v", + "2", output_path, ] @@ -61,11 +66,16 @@ def extract_first_frame( cmd2 = [ FFMPEG_BIN, "-y", - "-i", video_path, - "-ss", "00:00:00", - "-vframes", "1", - "-vf", scale_filter, - "-q:v", "2", + "-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) diff --git a/apps/worker/worker_app/tasks/batch_download.py b/apps/worker/worker_app/tasks/batch_download.py index 919e45f0c..e55aef068 100755 --- a/apps/worker/worker_app/tasks/batch_download.py +++ b/apps/worker/worker_app/tasks/batch_download.py @@ -26,10 +26,11 @@ def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict {"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, ) - from worker_app.db import SessionLocal session = SessionLocal() try: @@ -74,6 +75,7 @@ def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict raise RuntimeError("Batch download zip file is empty") import uuid + zip_storage_key = f"batch-downloads/{uuid.uuid4().hex}/{zip_filename}" download_url = upload_to_oss(str(zip_path), zip_storage_key) @@ -82,7 +84,9 @@ def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict logger.info( "Batch download complete: %d files, %d bytes, url=%s", - file_count, total_size, download_url, + file_count, + total_size, + download_url, ) return { @@ -107,4 +111,5 @@ def _download_video_to_file(url: str, dest_path: str) -> None: # 回退到 HTTP 下载 import urllib.request + urllib.request.urlretrieve(url, dest_path) # nosec B310 diff --git a/tests/render_compare/create_test_asset.py b/tests/render_compare/create_test_asset.py index c8a65d92e..268ae5731 100755 --- a/tests/render_compare/create_test_asset.py +++ b/tests/render_compare/create_test_asset.py @@ -25,9 +25,7 @@ import urllib.request def parse_args(): - parser = argparse.ArgumentParser( - description="创建 ready 状态的测试素材（跳过转码上传）" - ) + parser = argparse.ArgumentParser(description="创建 ready 状态的测试素材（跳过转码上传）") parser.add_argument( "--base-url", default=os.environ.get("API_BASE_URL", "http://localhost:8000"), @@ -213,9 +211,7 @@ def main(): print() # Step 1: 确保默认素材库存在 - library_id = ensure_default_library( - args.base_url, args.token, args.project_id, args.kind - ) + library_id = ensure_default_library(args.base_url, args.token, args.project_id, args.kind) # Step 2: 创建素材 asset = create_asset( -- 2.54.0 From 0347958d7516786eef2fdb57a103f800405b3f4e Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 09:24:50 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8Dflake8=20F811/F541?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/worker_app/tasks/batch_download.py | 3 --- tests/render_compare/create_test_asset.py | 6 +++--- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/apps/worker/worker_app/tasks/batch_download.py b/apps/worker/worker_app/tasks/batch_download.py index e55aef068..3924f8d8b 100755 --- a/apps/worker/worker_app/tasks/batch_download.py +++ b/apps/worker/worker_app/tasks/batch_download.py @@ -73,9 +73,6 @@ def batch_download_videos(self, video_ids: list[str], user_id: str = "") -> dict # 上传 zip 到 OSS if not zip_path.exists() or zip_path.stat().st_size == 0: raise RuntimeError("Batch download zip file is empty") - - import uuid - zip_storage_key = f"batch-downloads/{uuid.uuid4().hex}/{zip_filename}" download_url = upload_to_oss(str(zip_path), zip_storage_key) diff --git a/tests/render_compare/create_test_asset.py b/tests/render_compare/create_test_asset.py index 268ae5731..4998b7355 100755 --- a/tests/render_compare/create_test_asset.py +++ b/tests/render_compare/create_test_asset.py @@ -161,7 +161,7 @@ def create_asset( codec: str | None, ) -> dict: """创建 ready 状态的素材。""" - print(f"[2/2] 创建 ready 状态素材...") + print("[2/2] 创建 ready 状态素材...") body = { "project_id": project_id, "library_id": library_id, @@ -243,8 +243,8 @@ def main(): print() print("调用示例：") print(f" export ASSET_ID={asset_id}") - print(f" # 在生成任务中使用:") - print(f" # --asset-id $ASSET_ID") + print(" # 在生成任务中使用:") + print(" # --asset-id $ASSET_ID") print() # 输出 asset_id 到 stdout（方便脚本捕获） -- 2.54.0