From bbe831f9e0e25a2a5339533ac1d0685ab8e5fc89 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 13 Jul 2026 10:27:25 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(worker):=20render=5Fedit=5Fplan=20?= =?UTF-8?q?=E6=8E=A5=E5=85=A5=20Feature=20Flag=20=E7=81=B0=E5=BA=A6?= =?UTF-8?q?=E6=8E=A7=E5=88=B6=20(#240)=20(#240)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../worker_app/tasks/edit_plan_generation.py | 401 +++++++++++++----- tests/unit/test_edit_plan_worker_failure.py | 2 + 2 files changed, 304 insertions(+), 99 deletions(-) mode change 100644 => 100755 apps/worker/worker_app/tasks/edit_plan_generation.py mode change 100644 => 100755 tests/unit/test_edit_plan_worker_failure.py diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py old mode 100644 new mode 100755 index a89b9e3e9..a0011daf2 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -1,13 +1,18 @@ -"""剪辑计划渲染任务 — Phase 8 任务 2.05. +"""剪辑计划渲染任务 — 支持 Feature Flag 灰度. Celery 任务 worker.render_edit_plan: 1. 加载 EditPlan + EditPlanClips - 2. 下载各片段素材 - 3. 使用 UnifiedRenderService 按时间线+图层渲染 + 2. 根据 Feature Flag 选择渲染引擎（legacy / unified） + 3. 下载各片段素材 + 渲染 4. 上传渲染结果到 OSS 5. 创建 GeneratedVideo 记录 + 查重 6. 更新 EditPlan / EditPlanClip 状态 7. 更新 GenerationTask 进度 + +渲染引擎灰度： + - 走 Feature Flag (render_engine) 控制 + - legacy: VideoComposeService + FFmpeg filter_complex + - unified: UnifiedRenderService 图层架构 """ from __future__ import annotations @@ -63,14 +68,268 @@ def _get_repos(): # ── Celery Task ─────────────────────────────────────────────────────────────── +def _resolve_render_engine(user_id: str) -> str: + """根据 Feature Flag 决定使用哪个渲染引擎。 + + Returns: + "legacy" 或 "unified" + """ + try: + from video_processing.render_engine_resolver import get_render_engine_resolver + + resolver = get_render_engine_resolver() + return resolver.get_engine(user_id=user_id) + except Exception as exc: + logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc) + return "legacy" + + +def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg: str): + """统一的计划失败标记工具。""" + plan = plan_repo.get(plan_id) + if plan and plan.status.value == "rendering": + plan.mark_failed() + plan_repo.update(plan) + if generation_task_id: + gen_task = gen_task_repo.get(generation_task_id) + if gen_task and gen_task.status.value != "failed": + gen_task.status = "failed" + gen_task.error_message = error_msg + gen_task.completed_at = datetime.now(timezone.utc) + gen_task_repo.update(gen_task) + + +def _finalize_render_success( + plan, + plan_repo, + clip_repo, + gen_task_repo, + db, + plan_id: str, + output_url: str, + storage_key: str, + duration: float, + file_size: int, + width: int, + height: int, + rendered_clip_ids: list[str], + failed_clip_ids: list[str], + generation_task_id: str, + output_path: Path, + engine: str, +) -> dict: + """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。""" + # 创建 GeneratedVideo 记录 + 查重 + project_id = plan.project_id or "" + batch_id = plan.config.get("batch_id", "") + mode = plan.config.get("mode", "edit_plan") + if generation_task_id and project_id: + try: + create_video_record_and_dedup( + generation_task_id=generation_task_id, + project_id=project_id, + batch_id=batch_id, + file_url=output_url or "", + file_size=file_size, + duration=duration, + video_path=str(output_path), + mode=mode, + session=db, + width=width, + height=height, + fps=OUTPUT_FPS, + ) + except Exception as dedup_err: + logger.warning("查重失败（不影响渲染结果）: %s", dedup_err) + + # 更新片段状态为 rendered + for clip_id in rendered_clip_ids: + clip = clip_repo.get(clip_id) + if clip and clip.status.value == "ready": + clip.mark_rendered() + clip_repo.update(clip) + + # 更新 EditPlan 状态为 completed + plan.config["rendered_url"] = output_url or "" + plan.config["rendered_storage_key"] = storage_key + plan.mark_completed() + plan_repo.update(plan) + + # 更新 GenerationTask 状态为 completed + if generation_task_id: + gen_task = gen_task_repo.get(generation_task_id) + if gen_task: + gen_task.status = "completed" + gen_task.progress = 100.0 + gen_task.result_count = len(rendered_clip_ids) + gen_task.completed_at = datetime.now(timezone.utc) + gen_task_repo.update(gen_task) + + logger.info( + "剪辑计划渲染完成: plan_id=%s engine=%s rendered=%d failed=%d duration=%.1fs", + plan_id, + engine, + len(rendered_clip_ids), + len(failed_clip_ids), + duration, + ) + + return { + "status": "completed", + "plan_id": plan_id, + "rendered_count": len(rendered_clip_ids), + "failed_count": len(failed_clip_ids), + "output_url": output_url, + "duration": duration, + } + + +def _render_with_unified( + plan, + clips, + asset_path_map: dict[str, Path], + tmpdir_path: Path, + rendered_clip_ids: list[str], + plan_id: str, + generation_task_id: str, + plan_repo, + clip_repo, + gen_task_repo, + db, +) -> dict: + """统一渲染引擎路径（UnifiedRenderService 图层架构）。""" + render_service = UnifiedRenderService( + plan=plan, + clips=clips, + asset_path_map=asset_path_map, + work_dir=tmpdir_path, + output_width=OUTPUT_WIDTH, + output_height=OUTPUT_HEIGHT, + output_fps=int(OUTPUT_FPS), + ) + + try: + render_result = render_service.render() + except Exception as render_err: + logger.error("渲染失败(unified): %s — %s", plan_id, render_err) + _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}") + return {"status": "error", "message": f"渲染失败: {render_err}"} + + output_path = render_result.output_path + + # 上传到 OSS + storage_key = f"rendered/{plan_id}/output.mp4" + output_url = upload_to_oss(output_path, storage_key) + + failed_clip_ids: list[str] = [] + return _finalize_render_success( + plan=plan, + plan_repo=plan_repo, + clip_repo=clip_repo, + gen_task_repo=gen_task_repo, + db=db, + plan_id=plan_id, + output_url=output_url or "", + storage_key=storage_key, + duration=render_result.duration, + file_size=render_result.file_size, + width=render_result.width, + height=render_result.height, + rendered_clip_ids=rendered_clip_ids, + failed_clip_ids=failed_clip_ids, + generation_task_id=generation_task_id, + output_path=output_path, + engine="unified", + ) + + +def _render_with_legacy( + plan, + clips, + rendered_clip_ids: list[str], + failed_clip_ids: list[str], + tmpdir_path: Path, + plan_id: str, + generation_task_id: str, + plan_repo, + clip_repo, + gen_task_repo, + db, +) -> dict: + """旧引擎路径（VideoComposeService + FFmpeg filter_complex）。""" + import os + import subprocess + + from apps.api.app.services.video_compose_service import VideoComposeService + + compose_svc = VideoComposeService(db) + + # 校验合成条件 + validation = compose_svc.validate_compose(plan_id) + if not validation.valid: + error_msg = "; ".join(validation.errors) + logger.error("合成校验失败(legacy): %s — %s", plan_id, error_msg) + _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"合成校验失败: {error_msg}") + return {"status": "error", "message": error_msg} + + # 构建 FFmpeg 命令 + output_dir = os.environ.get("VIDEO_OUTPUT_DIR", str(tmpdir_path)) + output_path = Path(output_dir) / f"{plan_id}.mp4" + compose_cmd = compose_svc.build_compose_command(plan_id, str(output_path)) + + logger.info("执行 FFmpeg (legacy): plan_id=%s", plan_id) + try: + subprocess.run( + compose_cmd.command, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=3600, + ) + except subprocess.CalledProcessError as e: + error_msg = f"FFmpeg 执行失败: {e.stderr[:500]}" + logger.error("FFmpeg 执行失败(legacy): %s — %s", plan_id, error_msg) + _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg) + return {"status": "error", "message": error_msg} + + # 获取文件大小 + file_size = output_path.stat().st_size if output_path.exists() else 0 + duration = compose_cmd.estimated_duration or 0.0 + + # 上传到 OSS + storage_key = f"rendered/{plan_id}/output.mp4" + output_url = upload_to_oss(output_path, storage_key) + + return _finalize_render_success( + plan=plan, + plan_repo=plan_repo, + clip_repo=clip_repo, + gen_task_repo=gen_task_repo, + db=db, + plan_id=plan_id, + output_url=output_url or "", + storage_key=storage_key, + duration=duration, + file_size=file_size, + width=OUTPUT_WIDTH, + height=OUTPUT_HEIGHT, + rendered_clip_ids=rendered_clip_ids, + failed_clip_ids=failed_clip_ids, + generation_task_id=generation_task_id, + output_path=output_path, + engine="legacy", + ) + + @celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2) def render_edit_plan(self, plan_id: str) -> dict: """渲染剪辑计划 流程： 1. 加载 EditPlan + EditPlanClips - 2. 下载各片段素材到临时目录，构建 asset_path_map - 3. 使用 UnifiedRenderService 按时间线+图层渲染 + 2. 根据 Feature Flag 选择渲染引擎（legacy / unified） + 3. 下载素材 + 渲染 4. 上传渲染结果到 OSS 5. 创建 GeneratedVideo 记录 + 查重 6. 更新 EditPlan → completed, EditPlanClips → rendered @@ -79,6 +338,7 @@ def render_edit_plan(self, plan_id: str) -> dict: logger.info("开始渲染剪辑计划: plan_id=%s", plan_id) generation_task_id = "" + engine = "legacy" for repos in _get_repos(): plan_repo, clip_repo, gen_task_repo, db = repos @@ -93,7 +353,12 @@ def render_edit_plan(self, plan_id: str) -> dict: # 获取 generation_task_id（提前读取，确保 except 块可用） generation_task_id = plan.config.get("generation_task_id", "") - # 2. 加载片段列表（按 order 排序） + # 2. 选择渲染引擎（Feature Flag 灰度控制） + user_id = plan.created_by_user_id or "" + engine = _resolve_render_engine(user_id) + logger.info("剪辑计划渲染引擎: plan_id=%s engine=%s user_id=%s", plan_id, engine, user_id) + + # 3. 加载片段列表（按 order 排序） clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000) if not clips: logger.warning("剪辑计划没有片段: %s", plan_id) @@ -174,100 +439,38 @@ def render_edit_plan(self, plan_id: str) -> dict: gen_task_repo.update(gen_task) return {"status": "error", "message": "所有片段素材下载失败"} - # 4. 使用 UnifiedRenderService 渲染 - render_service = UnifiedRenderService( - plan=plan, - clips=clips, - asset_path_map=asset_path_map, - work_dir=tmpdir_path, - output_width=OUTPUT_WIDTH, - output_height=OUTPUT_HEIGHT, - output_fps=int(OUTPUT_FPS), - ) + # 4. 根据引擎选择渲染方式 + if engine == "unified": + result = _render_with_unified( + plan=plan, + clips=clips, + asset_path_map=asset_path_map, + tmpdir_path=tmpdir_path, + rendered_clip_ids=rendered_clip_ids, + plan_id=plan_id, + generation_task_id=generation_task_id, + plan_repo=plan_repo, + clip_repo=clip_repo, + gen_task_repo=gen_task_repo, + db=db, + ) + else: + result = _render_with_legacy( + plan=plan, + clips=clips, + rendered_clip_ids=rendered_clip_ids, + failed_clip_ids=failed_clip_ids, + tmpdir_path=tmpdir_path, + plan_id=plan_id, + generation_task_id=generation_task_id, + plan_repo=plan_repo, + clip_repo=clip_repo, + gen_task_repo=gen_task_repo, + db=db, + ) - try: - render_result = render_service.render() - except Exception as render_err: - logger.error("渲染失败: %s — %s", plan_id, render_err) - plan.mark_failed() - plan_repo.update(plan) - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task: - gen_task.status = "failed" - gen_task.error_message = f"渲染失败: {render_err}" - gen_task.completed_at = datetime.now(timezone.utc) - gen_task_repo.update(gen_task) - return {"status": "error", "message": f"渲染失败: {render_err}"} - - output_path = render_result.output_path - - # 5. 上传到 OSS - storage_key = f"rendered/{plan_id}/output.mp4" - output_url = upload_to_oss(output_path, storage_key) - - # 6. 创建 GeneratedVideo 记录 + 查重 - project_id = plan.project_id or "" - batch_id = plan.config.get("batch_id", "") - mode = plan.config.get("mode", "edit_plan") - if generation_task_id and project_id: - try: - create_video_record_and_dedup( - generation_task_id=generation_task_id, - project_id=project_id, - batch_id=batch_id, - file_url=output_url or "", - file_size=render_result.file_size, - duration=render_result.duration, - video_path=str(output_path), - mode=mode, - session=db, - width=render_result.width, - height=render_result.height, - fps=OUTPUT_FPS, - ) - except Exception as dedup_err: - logger.warning("查重失败（不影响渲染结果）: %s", dedup_err) - - # 7. 更新片段状态为 rendered - for clip_id in rendered_clip_ids: - clip = clip_repo.get(clip_id) - if clip and clip.status.value == "ready": - clip.mark_rendered() - clip_repo.update(clip) - - # 8. 更新 EditPlan 状态为 completed - plan.config["rendered_url"] = output_url or "" - plan.config["rendered_storage_key"] = storage_key - plan.mark_completed() - plan_repo.update(plan) - - # 9. 更新 GenerationTask 状态为 completed - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task: - gen_task.status = "completed" - gen_task.progress = 100.0 - gen_task.result_count = len(rendered_clip_ids) - gen_task.completed_at = datetime.now(timezone.utc) - gen_task_repo.update(gen_task) - - logger.info( - "剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d duration=%.1fs", - plan_id, - len(rendered_clip_ids), - len(failed_clip_ids), - render_result.duration, - ) - - return { - "status": "completed", - "plan_id": plan_id, - "rendered_count": len(rendered_clip_ids), - "failed_count": len(failed_clip_ids), - "output_url": output_url, - "duration": render_result.duration, - } + result["engine"] = engine + return result except Exception as exc: logger.exception("渲染剪辑计划异常: %s", plan_id) diff --git a/tests/unit/test_edit_plan_worker_failure.py b/tests/unit/test_edit_plan_worker_failure.py old mode 100644 new mode 100755 index 137e28767..aad74f916 --- a/tests/unit/test_edit_plan_worker_failure.py +++ b/tests/unit/test_edit_plan_worker_failure.py @@ -63,6 +63,8 @@ class StubEditPlan: template_id: str = "tmpl-001" status: Any = None config: dict = field(default_factory=dict) + project_id: str = "" + created_by_user_id: str = "user-001" def mark_failed(self): self.status = _StubStatus("failed") From a0cac1b75d7930d6aee342243f395ec61b17808a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Mon, 13 Jul 2026 11:07:10 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(worker):=20FFmpeg=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E4=BF=9D=E6=8A=A4=20-=20=E9=98=B2=E6=AD=A2=E6=B8=B2=E6=9F=93ha?= =?UTF-8?q?ng=E4=BD=8F=E5=AF=BC=E8=87=B4worker=E6=B0=B8=E4=B9=85=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=20(#242)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/api/router.py | 5 + apps/api/app/api/routes/internal_render.py | 120 ++++++++++++ apps/worker/video_processing/ffmpeg_utils.py | 16 ++ tests/unit/test_ffmpeg_timeout_protection.py | 93 +++++++++ tests/unit/test_internal_render_download.py | 190 +++++++++++++++++++ 5 files changed, 424 insertions(+) create mode 100755 apps/api/app/api/routes/internal_render.py create mode 100644 tests/unit/test_ffmpeg_timeout_protection.py create mode 100644 tests/unit/test_internal_render_download.py diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index ec8902fe1..5fdac8e03 100755 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -13,6 +13,7 @@ from app.api.routes.generated_videos import router as generated_videos_router from app.api.routes.generation_tasks import router as generation_tasks_router from app.api.routes.health import router as health_check_router from app.api.routes.ingest_jobs import router as ingest_jobs_router +from app.api.routes.internal_render import router as internal_render_router from app.api.routes.jobs import router as jobs_router from app.api.routes.projects import router as projects_router from app.api.routes.recipes import router as recipes_router @@ -156,3 +157,7 @@ api_router.include_router( feature_flags_router, tags=["Internal"], ) +api_router.include_router( + internal_render_router, + tags=["Internal"], +) diff --git a/apps/api/app/api/routes/internal_render.py b/apps/api/app/api/routes/internal_render.py new file mode 100755 index 000000000..9d1cf3da3 --- /dev/null +++ b/apps/api/app/api/routes/internal_render.py @@ -0,0 +1,120 @@ +"""渲染结果内部下载接口。 + +通过内部 API Key 鉴权，为灰度对比工具等内部系统提供渲染结果下载能力。 + +API: + GET /api/v1/internal/render/videos/{video_id}/download-url - 获取单个视频下载URL + GET /api/v1/internal/render/tasks/{task_id}/videos - 获取任务下所有视频及下载URL + +鉴权：X-API-Key header，走内部 API Key 验证 +""" + +from __future__ import annotations + +import logging +from typing import Any + +from app.api.routes.auth import _verify_internal_api_key +from app.core.storage import OSSStorageService, get_storage_service +from app.dependencies import get_generated_video_repository +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/internal/render", tags=["Internal"]) + + +class InternalRenderVideoItem(BaseModel): + """内部渲染视频项。""" + + video_id: str + generation_task_id: str + project_id: str + name: str + file_url: str + file_size: int | None = None + duration: float | None = None + width: int | None = None + height: int | None = None + fps: float | None = None + status: str + download_url: str + + +class InternalRenderTaskVideosResponse(BaseModel): + """任务下所有渲染视频响应。""" + + task_id: str + count: int + videos: list[InternalRenderVideoItem] + + +class InternalRenderDownloadUrlResponse(BaseModel): + """单个视频下载URL响应。""" + + video_id: str + download_url: str + + +def _video_to_item(video: Any, download_url: str) -> InternalRenderVideoItem: + """将 GeneratedVideo 领域对象转为响应项。""" + return InternalRenderVideoItem( + video_id=video.id, + generation_task_id=video.generation_task_id, + project_id=video.project_id, + name=video.name, + file_url=video.file_url, + file_size=getattr(video, "file_size", None), + duration=getattr(video, "duration", None), + width=getattr(video, "width", None), + height=getattr(video, "height", None), + fps=getattr(video, "fps", None), + status=video.status, + download_url=download_url, + ) + + +@router.get("/videos/{video_id}/download-url", response_model=InternalRenderDownloadUrlResponse) +def get_render_video_download_url( + video_id: str, + _: bool = Depends(_verify_internal_api_key), + generated_video_repository: Any = Depends(get_generated_video_repository), + storage_service: OSSStorageService = Depends(get_storage_service), +) -> InternalRenderDownloadUrlResponse: + """获取单个渲染视频的下载URL（预签名）。""" + video = generated_video_repository.get(video_id) + if video is None: + raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found") + + download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400) + logger.info("内部渲染下载URL生成: video_id=%s", video_id) + return InternalRenderDownloadUrlResponse(video_id=video_id, download_url=download_url) + + +@router.get("/tasks/{task_id}/videos", response_model=InternalRenderTaskVideosResponse) +def get_render_task_videos( + task_id: str, + status: str | None = Query(None, description="按状态筛选，如 completed/failed"), + _: bool = Depends(_verify_internal_api_key), + generated_video_repository: Any = Depends(get_generated_video_repository), + storage_service: OSSStorageService = Depends(get_storage_service), +) -> InternalRenderTaskVideosResponse: + """获取生成任务下所有渲染视频及下载URL。""" + videos = generated_video_repository.list_by_generation_task(task_id) + + # 状态筛选 + if status: + videos = [v for v in videos if v.status == status] + + items = [] + for video in videos: + download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400) + items.append(_video_to_item(video, download_url)) + + logger.info("内部渲染任务视频查询: task_id=%s count=%d", task_id, len(items)) + return InternalRenderTaskVideosResponse( + task_id=task_id, + count=len(items), + videos=items, + ) diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 6501dca76..dab6b4b72 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -42,6 +42,10 @@ XFADE_TRANSITION_MAP: dict[str, str] = { DEFAULT_TRANSITION_DURATION = 0.5 +# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致 worker 永久阻塞 +# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖 +DEFAULT_FFMPEG_TIMEOUT = 1800 + # ── FFmpeg 执行 ─────────────────────────────────────────────────────────────── @@ -50,12 +54,14 @@ def run_ffmpeg( command: list[str], *, capture_output: bool = True, + timeout: int | None = DEFAULT_FFMPEG_TIMEOUT, ) -> tuple[str, str]: """执行 FFmpeg 命令。 Args: command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身） capture_output: 是否捕获 stdout/stderr + timeout: 超时时间（秒），默认 1800s（30分钟）；None 表示不设超时（不推荐） Returns: (stdout, stderr) 元组 @@ -63,6 +69,7 @@ def run_ffmpeg( Raises: subprocess.CalledProcessError: 命令执行失败时抛出， 异常信息包含完整 stderr 以便排查。 + subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。 """ try: result = subprocess.run( # nosec B603 @@ -71,8 +78,16 @@ def run_ffmpeg( stdout=subprocess.PIPE if capture_output else None, stderr=subprocess.PIPE if capture_output else None, text=True, + timeout=timeout, ) return (result.stdout or "", result.stderr or "") + except subprocess.TimeoutExpired as e: + logger.error( + "FFmpeg 命令超时 (%ds): command=%s", + timeout or -1, + " ".join(str(c) for c in command[:20]), + ) + raise except subprocess.CalledProcessError as e: # 把完整 stderr 打到日志，方便排查 exit code 183 等问题 stderr_text = (e.stderr or "").strip() @@ -174,6 +189,7 @@ def probe_video_info(video_path: str) -> dict[str, Any]: stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + timeout=15, ) import json diff --git a/tests/unit/test_ffmpeg_timeout_protection.py b/tests/unit/test_ffmpeg_timeout_protection.py new file mode 100644 index 000000000..04ecbac22 --- /dev/null +++ b/tests/unit/test_ffmpeg_timeout_protection.py @@ -0,0 +1,93 @@ +"""FFmpeg 超时保护测试。 + +验证 run_ffmpeg / probe_video_info 的超时保护机制， +防止 FFmpeg hang 住导致 worker 永久阻塞。 +""" + +from __future__ import annotations + +import subprocess +from unittest.mock import MagicMock, patch + +import pytest +from video_processing.ffmpeg_utils import ( + DEFAULT_FFMPEG_TIMEOUT, + probe_video_info, + run_ffmpeg, +) + +# ── run_ffmpeg 超时保护 ────────────────────────────────────────────────────── + + +class TestRunFFmpegTimeout: + """run_ffmpeg 超时保护测试。""" + + def test_default_timeout_is_set(self): + """默认超时应为 1800 秒（30分钟）。""" + assert DEFAULT_FFMPEG_TIMEOUT == 1800 + + def test_timeout_expired_is_raised(self): + """超时未完成时 TimeoutExpired 异常被传播。""" + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg", "test"], timeout=1) + with pytest.raises(subprocess.TimeoutExpired): + run_ffmpeg(["ffmpeg", "test"]) + + def test_custom_timeout(self): + """支持自定义超时时间。""" + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=5) + with pytest.raises(subprocess.TimeoutExpired): + run_ffmpeg(["ffmpeg", "test"], timeout=5) + + def test_none_timeout_disables_protection(self): + """timeout=None 可以禁用超时保护（不推荐）。""" + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_result = MagicMock() + mock_result.stdout = "" + mock_result.stderr = "" + mock_run.return_value = mock_result + run_ffmpeg(["ffmpeg", "test"], timeout=None) + # 验证 timeout=None 被传递 + call_kwargs = mock_run.call_args.kwargs + assert call_kwargs["timeout"] is None + + def test_called_process_error_still_raised(self): + """超时异常不影响原有 CalledProcessError 的抛出。""" + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error msg") + with pytest.raises(subprocess.CalledProcessError): + run_ffmpeg(["ffmpeg", "test"]) + + +# ── probe_video_info 超时保护 ──────────────────────────────────────────────── + + +class TestProbeVideoInfoTimeout: + """probe_video_info 超时保护测试。""" + + def test_probe_uses_timeout(self): + """probe_video_info 调用 ffprobe 时应设置 timeout=15。""" + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=15) + # 超时异常被捕获，返回默认值 + result = probe_video_info("/tmp/test.mp4") + assert result["width"] == 1280 # DEFAULT_OUTPUT_WIDTH + assert result["height"] == 720 # DEFAULT_OUTPUT_HEIGHT + + def test_probe_success(self): + """正常情况应解析 ffprobe JSON 输出。""" + fake_output = """ + { + "streams": [{"width": 1920, "height": 1080, "r_frame_rate": "30/1", "duration": "10.5"}], + "format": {"duration": "10.5"} + } + """ + with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run: + mock_result = MagicMock() + mock_result.stdout = fake_output + mock_run.return_value = mock_result + result = probe_video_info("/tmp/test.mp4") + assert result["width"] == 1920 + assert result["height"] == 1080 + assert abs(result["duration"] - 10.5) < 0.01 diff --git a/tests/unit/test_internal_render_download.py b/tests/unit/test_internal_render_download.py new file mode 100644 index 000000000..c93b1082e --- /dev/null +++ b/tests/unit/test_internal_render_download.py @@ -0,0 +1,190 @@ +"""渲染结果内部下载接口单元测试。 + +测试 internal_render 路由的核心逻辑，mock 掉 repository 和 storage 依赖。 +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest +from app.api.routes.internal_render import ( + InternalRenderDownloadUrlResponse, + InternalRenderTaskVideosResponse, + _video_to_item, + get_render_task_videos, + get_render_video_download_url, +) + +# ── Helpers ──────────────────────────────────────────────────────────────── + + +class MockVideo: + """模拟 GeneratedVideo 领域对象。""" + + def __init__(self, **kwargs): + self.id = kwargs.get("id", "video-1") + self.generation_task_id = kwargs.get("generation_task_id", "task-1") + self.project_id = kwargs.get("project_id", "proj-1") + self.name = kwargs.get("name", "test_video.mp4") + self.file_url = kwargs.get("file_url", "videos/test/output.mp4") + self.file_size = kwargs.get("file_size", 1024000) + self.duration = kwargs.get("duration", 30.5) + self.width = kwargs.get("width", 1080) + self.height = kwargs.get("height", 1920) + self.fps = kwargs.get("fps", 30.0) + self.status = kwargs.get("status", "completed") + + +# ── _video_to_item 测试 ──────────────────────────────────────────────────── + + +class TestVideoToItem: + """测试视频对象转响应项。""" + + def test_basic_conversion(self): + video = MockVideo(id="v1", generation_task_id="t1", status="completed") + item = _video_to_item(video, "https://oss.example.com/download?v1") + assert item.video_id == "v1" + assert item.generation_task_id == "t1" + assert item.status == "completed" + assert item.download_url == "https://oss.example.com/download?v1" + + def test_missing_optional_fields(self): + """缺可选字段时返回 None。""" + video = MockVideo() + # 去掉可选字段 + del video.file_size + del video.duration + item = _video_to_item(video, "https://example.com/dl") + assert item.file_size is None + assert item.duration is None + assert item.width == 1080 # 还在 + + +# ── 路由函数测试 ──────────────────────────────────────────────────────────── + + +class TestGetRenderVideoDownloadUrl: + """测试单个视频下载URL接口。""" + + def test_video_exists(self): + video = MockVideo(id="v-abc", file_url="videos/abc/out.mp4") + mock_repo = MagicMock() + mock_repo.get.return_value = video + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = "https://oss.test/signed?v=abc" + + result = get_render_video_download_url( + video_id="v-abc", + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + + assert isinstance(result, InternalRenderDownloadUrlResponse) + assert result.video_id == "v-abc" + assert result.download_url == "https://oss.test/signed?v=abc" + mock_repo.get.assert_called_once_with("v-abc") + mock_storage.get_download_url.assert_called_once() + + def test_video_not_found_raises_404(self): + from fastapi import HTTPException + + mock_repo = MagicMock() + mock_repo.get.return_value = None + mock_storage = MagicMock() + + with pytest.raises(HTTPException) as exc_info: + get_render_video_download_url( + video_id="nonexistent", + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + assert exc_info.value.status_code == 404 + + def test_download_url_long_expiry(self): + """过期时间应为 24 小时（86400s）。""" + video = MockVideo(id="v1") + mock_repo = MagicMock() + mock_repo.get.return_value = video + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = "https://oss.test/signed" + + get_render_video_download_url( + video_id="v1", + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + + # 验证 expires_seconds=86400 + call_kwargs = mock_storage.get_download_url.call_args + assert call_kwargs.kwargs.get("expires_seconds") == 86400 or call_kwargs[1].get("expires_seconds") == 86400 + + +class TestGetRenderTaskVideos: + """测试任务视频列表接口。""" + + def test_list_multiple_videos(self): + videos = [ + MockVideo(id="v1", status="completed"), + MockVideo(id="v2", status="completed"), + MockVideo(id="v3", status="failed"), + ] + mock_repo = MagicMock() + mock_repo.list_by_generation_task.return_value = videos + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = "https://oss.test/signed" + + result = get_render_task_videos( + task_id="task-1", + status=None, + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + + assert isinstance(result, InternalRenderTaskVideosResponse) + assert result.task_id == "task-1" + assert result.count == 3 + assert len(result.videos) == 3 + + def test_filter_by_status(self): + videos = [ + MockVideo(id="v1", status="completed"), + MockVideo(id="v2", status="completed"), + MockVideo(id="v3", status="failed"), + ] + mock_repo = MagicMock() + mock_repo.list_by_generation_task.return_value = videos + mock_storage = MagicMock() + mock_storage.get_download_url.return_value = "https://oss.test/signed" + + result = get_render_task_videos( + task_id="task-1", + status="completed", + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + + assert result.count == 2 + assert all(v.status == "completed" for v in result.videos) + + def test_empty_task(self): + mock_repo = MagicMock() + mock_repo.list_by_generation_task.return_value = [] + mock_storage = MagicMock() + + result = get_render_task_videos( + task_id="empty-task", + status=None, + _=True, + generated_video_repository=mock_repo, + storage_service=mock_storage, + ) + + assert result.count == 0 + assert result.videos == []