Compare commits

...

2 Commits

Author SHA1 Message Date
CI Bot aca20d9bf0 fix(worker): FFmpeg 超时保护 - 防止渲染 hang 住导致 worker 永久阻塞
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 3m28s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 6m18s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m0s
根因:新引擎 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 所有渲染/混音/合并操作)
2026-07-13 11:05:13 +08:00
CI Bot e78c551ab1 feat(api): 新增渲染结果内部下载接口 - 支持按视频ID/任务ID获取预签名URL
- 新增 /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个单元测试覆盖核心逻辑
2026-07-13 11:05:13 +08:00
5 changed files with 424 additions and 0 deletions
+5
View File
@@ -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"],
)
+120
View File
@@ -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,
)
@@ -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
@@ -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
+190
View File
@@ -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 == []