From e78c551ab15f3465051112bdcc6495c385fd3e43 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 09:54:18 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(api):=20=E6=96=B0=E5=A2=9E=E6=B8=B2?= =?UTF-8?q?=E6=9F=93=E7=BB=93=E6=9E=9C=E5=86=85=E9=83=A8=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=20-=20=E6=94=AF=E6=8C=81=E6=8C=89=E8=A7=86?= =?UTF-8?q?=E9=A2=91ID/=E4=BB=BB=E5=8A=A1ID=E8=8E=B7=E5=8F=96=E9=A2=84?= =?UTF-8?q?=E7=AD=BE=E5=90=8DURL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 /internal/render/videos/{video_id}/download-url 接口：单个视频下载URL - 新增 /internal/render/tasks/{task_id}/videos 接口：任务下所有视频+下载URL - 内部 API Key 鉴权（X-API-Key），不对外开放 - 下载URL有效期24小时，方便对比工具批量下载 - 支持按 status 筛选（completed/failed 等） - 8个单元测试覆盖核心逻辑 --- apps/api/app/api/router.py | 5 + apps/api/app/api/routes/internal_render.py | 120 +++++++++++++ tests/unit/test_internal_render_download.py | 190 ++++++++++++++++++++ 3 files changed, 315 insertions(+) create mode 100755 apps/api/app/api/routes/internal_render.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/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 == [] -- 2.54.0 From aca20d9bf05822bc26a1058c810f5f4fb3b8bf16 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 10:17:14 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(worker):=20FFmpeg=20=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=93?= =?UTF-8?q?=20hang=20=E4=BD=8F=E5=AF=BC=E8=87=B4=20worker=20=E6=B0=B8?= =?UTF-8?q?=E4=B9=85=E9=98=BB=E5=A1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：新引擎 UnifiedRenderService 所有 FFmpeg 调用通过 run_ffmpeg 执行， 但 subprocess.run 未设置 timeout，FFmpeg hang 住时 worker 线程永久阻塞。 旧引擎 compose_video 有单独的 timeout=3600，但新引擎路径没有。 修复： 1. run_ffmpeg 新增默认超时 1800s（30分钟），支持自定义传参 2. 捕获 TimeoutExpired 并打 error 日志后重新抛出 3. probe_video_info 新增 timeout=15s 超时保护 4. 7个单元测试覆盖超时逻辑 影响范围：所有通过 run_ffmpeg 调用的 FFmpeg 命令 （unified_render_service 所有渲染/混音/合并操作） --- apps/worker/video_processing/ffmpeg_utils.py | 16 ++++ tests/unit/test_ffmpeg_timeout_protection.py | 93 ++++++++++++++++++++ 2 files changed, 109 insertions(+) create mode 100644 tests/unit/test_ffmpeg_timeout_protection.py 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 -- 2.54.0