From bbf3006d0cfbc0e7ea7f8b9cc0969f5e4957f580 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 13 Aug 2026 19:54:46 +0800 Subject: [PATCH 1/6] perf: pre-extract cover frames during render + FFmpeg fallback - Add extract_cover_candidates() and extract_and_upload_cover_frames() to thumbnail_generator.py: extract 3 frames at 25%/50%/75% of video - Add cover_candidates field to RenderAdapterResult - Call frame extraction after thumbnail gen in render_adapter._do_render() - Write cover_candidates to plan.config in edit_plan_generation worker - Check pre-stored frames in generation_cover.py before calling AI service - Replace MediaKit remote API with FFmpeg HTTP range-request in _call_ai_cover_service (no full video download needed) - Fix CoverTemplateResponse ValidationError when config is None - Add unit tests for all new code paths - Update existing tests to match new FFmpeg-based behavior --- apps/api/app/api/routes/cover_templates.py | 2 +- apps/api/app/api/routes/generation_cover.py | 21 ++ .../worker/video_processing/render_adapter.py | 21 ++ .../video_processing/thumbnail_generator.py | 119 ++++++ .../worker_app/tasks/edit_plan_generation.py | 10 + packages/shared/ai_service.py | 211 +++++++---- tests/unit/test_cover_frame_pre_extract.py | 342 ++++++++++++++++++ tests/unit/test_mediakit_cover.py | 123 +++---- tests/unit/test_shared_ai_service.py | 4 +- 9 files changed, 710 insertions(+), 143 deletions(-) create mode 100644 tests/unit/test_cover_frame_pre_extract.py diff --git a/apps/api/app/api/routes/cover_templates.py b/apps/api/app/api/routes/cover_templates.py index f42cbfda2..a8782cc8c 100644 --- a/apps/api/app/api/routes/cover_templates.py +++ b/apps/api/app/api/routes/cover_templates.py @@ -55,7 +55,7 @@ def list_cover_templates( thumbnail_url=t.thumbnail_url, is_system=t.is_system, created_at=t.created_at, - config=t.config, + config=t.config or {}, ) for t in items ], diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index 033de871f..ddc63df4e 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -182,6 +182,27 @@ def generate_cover( detail=f"获取预览视频URL失败: {e}", ) from e + # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回） + cover_candidates = (plan.config or {}).get("cover_candidates", []) + if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"): + logger.info( + "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d", + plan_id, len(cover_candidates), + ) + first_frame = cover_candidates[0] + cover_data = { + "type": "ai_frame", + "image_url": first_frame.get("image_url", ""), + "frame_time": first_frame.get("frame_time", 0.0), + "confidence": 0.9, + } + if cover_data["image_url"]: + current_config = dict(plan.config) if plan.config else {} + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) + from packages.shared.ai_service import run_generate_cover try: diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index 8697d943e..cb799bcca 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -74,6 +74,7 @@ class RenderAdapterResult: failed_clip_ids: list[str] = None # 失败的 clip id 列表 error_message: str = "" error_detail: str = "" # 详细错误信息（如 ffmpeg stderr），用于排查 + cover_candidates: list[dict] | None = None # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}] def __post_init__(self): if self.rendered_clip_ids is None: @@ -568,6 +569,25 @@ class RenderAdapter: thumb_err, ) + # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程） + cover_candidates = None + try: + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + cover_candidates = extract_and_upload_cover_frames( + str(result.output_path), plan_id, num_frames=3 + ) + if cover_candidates: + logger.info( + "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d", + plan_id, len(cover_candidates), + ) + except Exception as cover_err: + logger.warning( + "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s", + plan_id, cover_err, + ) + self._report_progress(progress_cb, 100.0, "渲染完成") logger.info( @@ -599,6 +619,7 @@ class RenderAdapter: clip_count=len(clips), rendered_clip_ids=final_rendered_ids, failed_clip_ids=final_failed_ids, + cover_candidates=cover_candidates, ) def render_from_memory( diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index eecdc2244..2730ab092 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -155,3 +155,122 @@ def generate_and_upload_thumbnail( Path(thumbnail_path).unlink(missing_ok=True) except Exception: pass + + +def extract_cover_candidates( + video_path: str, + num_frames: int = 3, + *, + width: int = 640, + timeout: int = 30, +) -> list[dict]: + """在视频时长 25%/50%/75% 处各抽一帧，返回候选帧信息列表。 + + Args: + video_path: 视频文件路径 + num_frames: 抽帧数量（默认 3） + width: 输出宽度 + timeout: 单帧超时（秒） + + Returns: + [{"local_path": "...", "frame_time": 5.0}, ...] + """ + from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg + + try: + duration = probe_duration(video_path) + except Exception: + duration = 0.0 + + if duration <= 0: + duration = 5.0 # fallback + + # 计算抽帧时间点：25%, 50%, 75% + ratios = [] + for i in range(1, num_frames + 1): + ratios.append(i / (num_frames + 1)) + + results = [] + for idx, ratio in enumerate(ratios): + frame_time = max(0.5, duration * ratio) + tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp.close() + output_path = tmp.name + + try: + seek_str = _format_seek_time(frame_time) + scale_filter = f"scale={width}:-1:force_original_aspect_ratio=decrease,format=yuvj420p" + cmd = [ + FFMPEG_BIN, "-y", + "-ss", seek_str, + "-i", video_path, + "-vframes", "1", + "-vf", scale_filter, + "-q:v", "2", + output_path, + ] + run_ffmpeg(cmd, capture_output=True, timeout=timeout) + + if Path(output_path).exists() and Path(output_path).stat().st_size > 0: + results.append({ + "local_path": output_path, + "frame_time": round(frame_time, 2), + }) + else: + Path(output_path).unlink(missing_ok=True) + except Exception as e: + logger.warning("封面候选帧抽取失败 ratio=%.2f: %s", ratio, e) + Path(output_path).unlink(missing_ok=True) + + return results + + +def extract_and_upload_cover_frames( + video_path: str, + plan_id: str, + num_frames: int = 3, +) -> list[dict]: + """抽取封面候选帧并上传到 OSS。 + + Args: + video_path: 本地视频路径 + plan_id: 剪辑计划 ID（用于 OSS 路径） + num_frames: 抽帧数量 + + Returns: + [{"image_url": "https://...", "frame_time": 5.0, "storage_key": "covers/xxx/frame_0.jpg"}, ...] + """ + candidates = extract_cover_candidates(video_path, num_frames=num_frames) + if not candidates: + logger.warning("封面候选帧抽取为空: plan_id=%s", plan_id) + return [] + + results = [] + for idx, cand in enumerate(candidates): + local_path = cand["local_path"] + frame_time = cand["frame_time"] + storage_key = f"covers/{plan_id}/frame_{idx}.jpg" + + try: + from video_processing.oss_helpers import upload_to_oss + + url = upload_to_oss(local_path, storage_key) + if url: + results.append({ + "image_url": url, + "frame_time": frame_time, + "storage_key": storage_key, + }) + logger.info( + "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f", + plan_id, idx, frame_time, + ) + except Exception as e: + logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e) + finally: + try: + Path(local_path).unlink(missing_ok=True) + except Exception: + pass + + return results diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py index 729a202ce..0d740847c 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -256,6 +256,16 @@ def _render_with_unified( rendered_clip_ids = result.rendered_clip_ids or [] failed_clip_ids = result.failed_clip_ids or [] + # 将封面候选帧写入 plan.config（供封面 API 直接使用，跳过 MediaKit 抽帧） + if result.cover_candidates: + plan_config = plan.config or {} + plan_config["cover_candidates"] = result.cover_candidates + plan.config = plan_config + logger.info( + "封面候选帧已写入 plan.config: plan_id=%s count=%d", + plan_id, len(result.cover_candidates), + ) + return _finalize_render_success( plan=plan, plan_repo=plan_repo, diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index b5e7f0ff3..efe7f7c0a 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -354,6 +354,81 @@ def _transfer_cover_frame_to_storage(frame_url: str, plan_id: str) -> str: return frame_url +def _extract_frames_with_ffmpeg( + video_url: str, + num_frames: int = 3, + timeout: int = 30, +) -> list[dict]: + """用 FFmpeg 从远程视频 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）。 + + Args: + video_url: 视频 URL + num_frames: 抽帧数量 + timeout: 单帧超时（秒） + + Returns: + [{"local_path": "...", "frame_time": 5.0}, ...] + """ + import re as _re + import tempfile + from pathlib import Path as _Path + + from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg + + video_url = _re.sub(r"(? 0: + results.append({"local_path": output_path, "frame_time": round(frame_time, 2)}) + else: + _Path(output_path).unlink(missing_ok=True) + except Exception as e: + logger.warning("FFmpeg 远程抽帧失败 ratio=%.2f: %s", ratio, e) + _Path(output_path).unlink(missing_ok=True) + + return results + + def _call_ai_cover_service( plan_id: str, asset_ids: List[str], @@ -363,7 +438,10 @@ def _call_ai_cover_service( ) -> Dict[str, Any]: """调用 AI 封面生成服务. - 当 cover_type 为 ai_frame 或 ai_regenerate 时，调用 MediaKit 视频截帧。 + 优先级： + 1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理 + 2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频） + 失败时抛出 RuntimeError。 Args: @@ -371,7 +449,7 @@ def _call_ai_cover_service( asset_ids: 素材 ID 列表 cover_type: 封面类型 frame_time: 手动选帧时间点 - primary_video_url: 主视频的可访问 URL（用于 MediaKit 抽帧） + primary_video_url: 主视频的可访问 URL """ if cover_type == "upload": return { @@ -394,79 +472,82 @@ def _call_ai_cover_service( "frame_time": frame_time, } - # ai_frame / ai_regenerate - 尝试调用 MediaKit + # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧 if primary_video_url: - # 规范化 URL：合并路径中的双斜杠（保留协议头 ://） - # 历史数据中 project_id 为空时 OSS key 会出现 projects//tasks/ 路径 import re as _re - primary_video_url = _re.sub(r"(? 0: - # 选择第一帧（SceneChange 策略的第一帧通常是最佳画面） - best_frame = frames[0] - image_url = best_frame.get("image_url", "") - timestamp = best_frame.get("timestamp", 0.0) + # 使用 FFmpeg 从 URL 流式 seek 抽帧 + try: + logger.info("FFmpeg 远程抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80]) + frames = _extract_frames_with_ffmpeg(primary_video_url, num_frames=3) - if image_url: - logger.info( - "MediaKit 抽帧成功: plan_id=%s frame_time=%.2f url=%s", - plan_id, - timestamp, - image_url[:80], - ) - # MediaKit 返回的 URL 是临时内部 URL，浏览器无法直接访问 - # 需要下载到本地并重新上传到 OSS，返回公开可访问的 URL - public_url = _transfer_cover_frame_to_storage(image_url, plan_id) - return { - "type": "ai_frame", - "image_url": public_url, - "frame_time": round(timestamp, 1), - "confidence": 0.85, - } - else: - logger.warning("MediaKit 返回的帧无 image_url") + if frames: + best_frame = frames[0] + local_path = best_frame["local_path"] + frame_time_val = best_frame["frame_time"] - except Exception as e: - logger.exception("MediaKit 抽帧失败: %s", str(e)) + # 上传到 OSS + try: + import uuid + from pathlib import Path - # 封面生成失败 - 不再降级到 stub，直接报错 + from packages.shared.storage import get_shared_storage_service + + storage = get_shared_storage_service() + cover_key = f"covers/{plan_id}/ffmpeg_frame_{uuid.uuid4().hex[:8]}.jpg" + storage.upload_file( + file_or_path=local_path, + storage_key=cover_key, + content_type="image/jpeg", + ) + public_url = storage.get_url(cover_key) + + logger.info( + "FFmpeg 抽帧成功: plan_id=%s frame_time=%.2f url=%s", + plan_id, frame_time_val, public_url[:80], + ) + + return { + "type": "ai_frame", + "image_url": public_url, + "frame_time": round(frame_time_val, 1), + "confidence": 0.85, + } + finally: + # 清理所有临时文件 + for frame in frames: + try: + Path(frame["local_path"]).unlink(missing_ok=True) + except Exception: + pass + + except RuntimeError: + raise + except Exception as e: + logger.exception("FFmpeg 远程抽帧失败: %s", str(e)) + + # 封面生成失败 raise RuntimeError( - f"封面生成失败: plan_id={plan_id}, MediaKit 不可用或抽帧失败。" f"请检查 primary_video_url 是否可访问。" + f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。" ) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py new file mode 100644 index 000000000..0cd51b329 --- /dev/null +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -0,0 +1,342 @@ +"""Tests for cover frame pre-extraction during rendering. + +Tests: +- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75% +- extract_and_upload_cover_frames: extraction + OSS upload +- RenderAdapterResult.cover_candidates field +- generation_cover route uses pre-stored candidates +- ai_service FFmpeg fallback +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, Mock, call, patch + +import pytest + + +class TestExtractCoverCandidates: + """extract_cover_candidates 测试.""" + + @patch("video_processing.ffmpeg_utils.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0) + def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run): + """在 25%/50%/75% 处抽取 3 帧.""" + import tempfile + from video_processing.thumbnail_generator import extract_cover_candidates + + # Create temp files that look like they were created + def fake_run(cmd, **kwargs): + # Find the output path (last arg) + output_path = cmd[-1] + Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100) + return ("", "") + + mock_run.side_effect = fake_run + + with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: + tmp.write(b"fake video") + video_path = tmp.name + + try: + results = extract_cover_candidates(video_path, num_frames=3) + assert len(results) == 3 + + # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0 + assert results[0]["frame_time"] == 5.0 + assert results[1]["frame_time"] == 10.0 + assert results[2]["frame_time"] == 15.0 + + # Check local paths exist + for r in results: + assert Path(r["local_path"]).exists() + + # Clean up + for r in results: + Path(r["local_path"]).unlink(missing_ok=True) + finally: + Path(video_path).unlink(missing_ok=True) + + @patch("video_processing.ffmpeg_utils.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0) + def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run): + """FFmpeg 失败时跳过该帧，继续抽取其他帧.""" + import tempfile + from video_processing.thumbnail_generator import extract_cover_candidates + + call_count = 0 + + def fake_run(cmd, **kwargs): + nonlocal call_count + call_count += 1 + output_path = cmd[-1] + if call_count == 2: + # Second frame fails - don't create file + raise RuntimeError("ffmpeg error") + Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) + return ("", "") + + mock_run.side_effect = fake_run + + with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: + tmp.write(b"fake video") + video_path = tmp.name + + try: + results = extract_cover_candidates(video_path, num_frames=3) + # Should get 2 frames (1st and 3rd), 2nd failed + assert len(results) == 2 + finally: + Path(video_path).unlink(missing_ok=True) + for r in results: + Path(r["local_path"]).unlink(missing_ok=True) + + @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed")) + def test_fallback_duration_when_probe_fails(self, mock_probe): + """probe 失败时使用默认时长.""" + import tempfile + from video_processing.thumbnail_generator import extract_cover_candidates + + # Mock run_ffmpeg to create output files + def fake_run(cmd, **kwargs): + output_path = cmd[-1] + Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) + return ("", "") + + with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run): + with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: + tmp.write(b"fake") + video_path = tmp.name + + try: + results = extract_cover_candidates(video_path, num_frames=3) + assert len(results) == 3 + # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75 + assert results[0]["frame_time"] == 1.25 + assert results[1]["frame_time"] == 2.5 + assert results[2]["frame_time"] == 3.75 + finally: + Path(video_path).unlink(missing_ok=True) + for r in results: + Path(r["local_path"]).unlink(missing_ok=True) + + +class TestExtractAndUploadCoverFrames: + """extract_and_upload_cover_frames 测试.""" + + @patch("video_processing.oss_helpers.upload_to_oss") + @patch("video_processing.thumbnail_generator.extract_cover_candidates") + def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload): + """上传帧到 OSS 并返回正确格式.""" + import tempfile + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + # Create actual temp files + tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp1.write(b"\xff\xd8" + b"\x00" * 50) + tmp1.close() + tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp2.write(b"\xff\xd8" + b"\x00" * 50) + tmp2.close() + + mock_extract.return_value = [ + {"local_path": tmp1.name, "frame_time": 5.0}, + {"local_path": tmp2.name, "frame_time": 10.0}, + ] + mock_upload.side_effect = [ + "https://oss.example.com/covers/plan1/frame_0.jpg", + "https://oss.example.com/covers/plan1/frame_1.jpg", + ] + + results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") + + assert len(results) == 2 + assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg" + assert results[0]["frame_time"] == 5.0 + assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg" + + assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg" + assert results[1]["frame_time"] == 10.0 + + @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[]) + def test_returns_empty_when_no_candidates(self, mock_extract): + """没有候选帧时返回空列表.""" + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") + assert results == [] + + @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error")) + @patch("video_processing.thumbnail_generator.extract_cover_candidates") + def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload): + """上传失败时跳过该帧.""" + import tempfile + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp1.write(b"\xff\xd8" + b"\x00" * 50) + tmp1.close() + + mock_extract.return_value = [ + {"local_path": tmp1.name, "frame_time": 5.0}, + ] + + results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") + assert results == [] + + +class TestRenderAdapterResultCoverCandidates: + """RenderAdapterResult 的 cover_candidates 字段.""" + + def test_default_none(self): + """默认为 None.""" + from video_processing.render_adapter import RenderAdapterResult + + result = RenderAdapterResult(success=True) + assert result.cover_candidates is None + + def test_can_set_candidates(self): + """可以设置候选帧列表.""" + from video_processing.render_adapter import RenderAdapterResult + + candidates = [ + {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"}, + ] + result = RenderAdapterResult(success=True, cover_candidates=candidates) + assert len(result.cover_candidates) == 1 + assert result.cover_candidates[0]["frame_time"] == 5.0 + + +class TestAICoverServiceFFmpegFallback: + """AI 封面服务 FFmpeg 兜底测试.""" + + @patch("packages.shared.ai_service.http_requests.head") + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head): + """FFmpeg 兜底抽帧成功.""" + import tempfile + + mock_head.return_value.status_code = 200 + + tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp.write(b"\xff\xd8" + b"\x00" * 50) + tmp.close() + + mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}] + + # Mock storage + with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn: + mock_storage = Mock() + mock_storage.upload_file = Mock() + mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg" + mock_storage_fn.return_value = mock_storage + + from packages.shared.ai_service import _call_ai_cover_service + + result = _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) + + assert result["type"] == "ai_frame" + assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg" + assert result["frame_time"] == 5.0 + assert result["confidence"] == 0.85 + + Path(tmp.name).unlink(missing_ok=True) + + @patch("packages.shared.ai_service.http_requests.head") + def test_ffmpeg_no_video_url_raises(self, mock_head): + """没有视频 URL 时抛出 RuntimeError.""" + from packages.shared.ai_service import _call_ai_cover_service + + with pytest.raises(RuntimeError, match="无法从视频抽帧"): + _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="ai_frame", + primary_video_url=None, + ) + + @patch("packages.shared.ai_service.http_requests.head") + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head): + """FFmpeg 抽帧为空时抛出 RuntimeError.""" + mock_head.return_value.status_code = 200 + mock_ffmpeg.return_value = [] + + from packages.shared.ai_service import _call_ai_cover_service + + with pytest.raises(RuntimeError, match="无法从视频抽帧"): + _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) + + def test_upload_type_returns_immediately(self): + """upload 类型直接返回.""" + from packages.shared.ai_service import _call_ai_cover_service + + result = _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="upload", + primary_video_url="https://example.com/video.mp4", + ) + assert result["type"] == "upload" + + def test_manual_type_returns_immediately(self): + """manual 类型直接返回.""" + from packages.shared.ai_service import _call_ai_cover_service + + result = _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="manual", + frame_time=5.0, + primary_video_url="https://example.com/video.mp4", + ) + assert result["type"] == "manual" + assert result["frame_time"] == 5.0 + + @patch("packages.shared.ai_service.http_requests.head") + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head): + """视频 URL 不可访问时抛出 RuntimeError.""" + mock_head.return_value.status_code = 404 + + from packages.shared.ai_service import _call_ai_cover_service + + with pytest.raises(RuntimeError, match="预览视频URL不可访问"): + _call_ai_cover_service( + plan_id="plan1", + asset_ids=["a1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) + + +class TestCoverTemplatesFix: + """CoverTemplateResponse config=None 修复测试.""" + + def test_config_none_becomes_empty_dict(self): + """config=None 时 CoverTemplateResponse 不报 ValidationError.""" + from datetime import datetime + + from app.schemas.cover_template import CoverTemplateResponse + + # This should not raise + resp = CoverTemplateResponse( + id="1", + name="test", + thumbnail_url="", + is_system=True, + created_at=datetime.now(), + config={}, + ) + assert resp.config == {} diff --git a/tests/unit/test_mediakit_cover.py b/tests/unit/test_mediakit_cover.py index 287e2c0ef..4b063e4f4 100755 --- a/tests/unit/test_mediakit_cover.py +++ b/tests/unit/test_mediakit_cover.py @@ -3,6 +3,7 @@ 测试 #1208: AI封面接入MediaKit视频截帧 """ +from pathlib import Path from unittest.mock import MagicMock, Mock, patch import pytest @@ -140,47 +141,49 @@ class TestMediaKitClient: class TestAICoverService: - """AI 封面服务测试.""" + """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。""" @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_with_mediakit_success(self, mock_get_client, mock_head): - """MediaKit 抽帧成功.""" - # Mock HEAD request to verify URL is accessible + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_call_ai_cover_with_ffmpeg_success(self, mock_ffmpeg, mock_head): + """FFmpeg 本地抽帧成功.""" + import tempfile + mock_head.return_value.status_code = 200 - mock_client = Mock() - mock_client.is_available = True - mock_client.extract_frames.return_value = [{"image_url": "https://example.com/frame.jpg", "timestamp": 3.5}] - mock_get_client.return_value = mock_client + tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp.write(b"\xff\xd8" + b"\x00" * 50) + tmp.close() - from packages.shared.ai_service import _call_ai_cover_service + mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 3.5}] - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) + with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn: + mock_storage = Mock() + mock_storage.upload_file = Mock() + mock_storage.get_url.return_value = "https://example.com/frame.jpg" + mock_storage_fn.return_value = mock_storage - assert result["type"] == "ai_frame" - assert result["image_url"] == "https://example.com/frame.jpg" - assert result["frame_time"] == 3.5 - assert result["confidence"] == 0.85 + from packages.shared.ai_service import _call_ai_cover_service - mock_client.extract_frames.assert_called_once() + result = _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) + + assert result["type"] == "ai_frame" + assert result["image_url"] == "https://example.com/frame.jpg" + assert result["frame_time"] == 3.5 + assert result["confidence"] == 0.85 + + Path(tmp.name).unlink(missing_ok=True) @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_video_url_unreachable(self, mock_get_client, mock_head): + def test_call_ai_cover_video_url_unreachable(self, mock_head): """视频 URL 不可访问时抛出 RuntimeError.""" - # Mock HEAD request to return 404 mock_head.return_value.status_code = 404 - mock_client = Mock() - mock_client.is_available = True - mock_get_client.return_value = mock_client - from packages.shared.ai_service import _call_ai_cover_service with pytest.raises(RuntimeError, match="预览视频URL不可访问"): @@ -192,17 +195,14 @@ class TestAICoverService: ) @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_url_double_slash_normalized(self, mock_get_client, mock_head): - """URL 路径中的双斜杠应被规范化，避免 MediaKit 404.""" + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_call_ai_cover_url_double_slash_normalized(self, mock_ffmpeg, mock_head): + """URL 路径中的双斜杠应被规范化.""" dirty_url = "https://oss.example.com/generated/projects//tasks/abc123/rendered.mp4" clean_url = "https://oss.example.com/generated/projects/tasks/abc123/rendered.mp4" mock_head.return_value.status_code = 200 - mock_client = Mock() - mock_client.is_available = True - mock_client.extract_frames.return_value = [] - mock_get_client.return_value = mock_client + mock_ffmpeg.return_value = [] from packages.shared.ai_service import _call_ai_cover_service @@ -219,20 +219,15 @@ class TestAICoverService: assert mock_head.call_args[0][0] == clean_url @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_with_mediakit_failure_raises(self, mock_get_client, mock_head): - """MediaKit 失败时抛出 RuntimeError（不再降级到 stub）.""" - # Mock HEAD request to return 200 (URL is accessible, but MediaKit fails) + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_call_ai_cover_ffmpeg_failure_raises(self, mock_ffmpeg, mock_head): + """FFmpeg 抽帧失败时抛出 RuntimeError.""" mock_head.return_value.status_code = 200 - - mock_client = Mock() - mock_client.is_available = True - mock_client.extract_frames.side_effect = Exception("API error") - mock_get_client.return_value = mock_client + mock_ffmpeg.side_effect = Exception("ffmpeg error") from packages.shared.ai_service import _call_ai_cover_service - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): _call_ai_cover_service( plan_id="plan-123", asset_ids=["asset-1"], @@ -241,11 +236,10 @@ class TestAICoverService: ) def test_call_ai_cover_without_video_url_raises(self): - """没有视频 URL 时抛出 RuntimeError（不再降级到 stub）.""" - + """没有视频 URL 时抛出 RuntimeError.""" from packages.shared.ai_service import _call_ai_cover_service - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): _call_ai_cover_service( plan_id="plan-123", asset_ids=["asset-1"], @@ -282,37 +276,16 @@ class TestAICoverService: assert result["type"] == "manual" assert result["frame_time"] == 5.0 - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_mediakit_not_available_raises(self, mock_get_client): - """MediaKit 未配置时抛出 RuntimeError（不再降级到 stub）.""" - mock_client = Mock() - mock_client.is_available = False - mock_get_client.return_value = mock_client - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="MediaKit"): - _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_empty_frames_raises(self, mock_get_client, mock_head): - """MediaKit 返回空帧列表时抛出 RuntimeError（不再降级）.""" - mock_head.return_value.status_code = 200 # URL accessible - - mock_client = Mock() - mock_client.is_available = True - mock_client.extract_frames.return_value = [] - mock_get_client.return_value = mock_client + @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") + def test_call_ai_cover_empty_frames_raises(self, mock_ffmpeg, mock_head): + """FFmpeg 返回空帧列表时抛出 RuntimeError.""" + mock_head.return_value.status_code = 200 + mock_ffmpeg.return_value = [] from packages.shared.ai_service import _call_ai_cover_service - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): _call_ai_cover_service( plan_id="plan-123", asset_ids=["asset-1"], diff --git a/tests/unit/test_shared_ai_service.py b/tests/unit/test_shared_ai_service.py index b9da45a95..f6b8ef35e 100755 --- a/tests/unit/test_shared_ai_service.py +++ b/tests/unit/test_shared_ai_service.py @@ -436,12 +436,12 @@ class TestAiCoverService: def test_cover_type_ai_frame_raises_without_mediakit(self): """ai_frame mode raises RuntimeError when MediaKit is unavailable.""" - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): _call_ai_cover_service("plan1", ["a1"], "ai_frame") def test_cover_type_ai_regenerate_raises_without_mediakit(self): """ai_regenerate mode raises RuntimeError when MediaKit is unavailable.""" - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): _call_ai_cover_service("plan1", ["a1"], "ai_regenerate") def test_cover_type_manual_still_works(self): -- 2.54.0 From 7ebff0dd62797a5c7157bbd47f43442527f6c96f Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 13 Aug 2026 20:01:42 +0800 Subject: [PATCH 2/6] style: apply black + isort + ruff formatting fixes --- apps/api/app/api/routes/cover_templates.py | 1 - apps/api/app/api/routes/generation_cover.py | 10 ++-- .../worker/video_processing/render_adapter.py | 14 +++--- .../video_processing/thumbnail_generator.py | 46 +++++++++++------- .../worker_app/tasks/edit_plan_generation.py | 3 +- packages/shared/ai_service.py | 48 ++++++++++++------- tests/unit/test_cover_frame_pre_extract.py | 5 ++ 7 files changed, 82 insertions(+), 45 deletions(-) diff --git a/apps/api/app/api/routes/cover_templates.py b/apps/api/app/api/routes/cover_templates.py index a8782cc8c..5c6d8402c 100644 --- a/apps/api/app/api/routes/cover_templates.py +++ b/apps/api/app/api/routes/cover_templates.py @@ -7,7 +7,6 @@ API: DELETE /api/v1/cover-templates/{id} - 删除自定义模板（系统模板不可删） """ - import logging from typing import Any diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index ddc63df4e..c9bd2bf4d 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -86,7 +86,9 @@ def generate_cover( # 第二步：如果还没有，通过 generation_task_id 查找预览任务的产物 if not rendered_storage_key: generation_task_id = (plan.config or {}).get("generation_task_id", "") - logger.info("[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id) + logger.info( + "[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id + ) if generation_task_id: try: gen_task_repo = SQLAlchemyGenerationTaskRepository(db) @@ -170,7 +172,8 @@ def generate_cover( # MediaKit 的 HTTP 客户端会规范化 URL 导致 404 if primary_video_url: import re as _re - primary_video_url = _re.sub(r'(? 0: - results.append({ - "local_path": output_path, - "frame_time": round(frame_time, 2), - }) + results.append( + { + "local_path": output_path, + "frame_time": round(frame_time, 2), + } + ) else: Path(output_path).unlink(missing_ok=True) except Exception as e: @@ -256,14 +264,18 @@ def extract_and_upload_cover_frames( url = upload_to_oss(local_path, storage_key) if url: - results.append({ - "image_url": url, - "frame_time": frame_time, - "storage_key": storage_key, - }) + results.append( + { + "image_url": url, + "frame_time": frame_time, + "storage_key": storage_key, + } + ) logger.info( "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f", - plan_id, idx, frame_time, + plan_id, + idx, + frame_time, ) except Exception as e: logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e) diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py index 0d740847c..58155e4ab 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -263,7 +263,8 @@ def _render_with_unified( plan.config = plan_config logger.info( "封面候选帧已写入 plan.config: plan_id=%s count=%d", - plan_id, len(result.cover_candidates), + plan_id, + len(result.cover_candidates), ) return _finalize_render_success( diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index efe7f7c0a..2ae55042e 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -379,18 +379,25 @@ def _extract_frames_with_ffmpeg( # 先用 ffprobe 获取视频时长 import subprocess as _subprocess + from packages.shared.ffmpeg_utils import FFPROBE_BIN duration = 30.0 # 默认假设 30 秒 try: probe_result = _subprocess.run( [ - FFPROBE_BIN, "-v", "error", - "-show_entries", "format=duration", - "-of", "default=noprint_wrappers=1:nokey=1", + FFPROBE_BIN, + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", video_url, ], - capture_output=True, text=True, timeout=15, + capture_output=True, + text=True, + timeout=15, ) if probe_result.returncode == 0 and probe_result.stdout.strip(): duration = float(probe_result.stdout.strip()) @@ -400,7 +407,7 @@ def _extract_frames_with_ffmpeg( ratios = [i / (num_frames + 1) for i in range(1, num_frames + 1)] results = [] - for idx, ratio in enumerate(ratios): + for _idx, ratio in enumerate(ratios): frame_time = max(0.5, duration * ratio) tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) tmp.close() @@ -409,11 +416,16 @@ def _extract_frames_with_ffmpeg( try: seek_str = f"{int(frame_time // 3600):02d}:{int((frame_time % 3600) // 60):02d}:{frame_time % 60:05.2f}" cmd = [ - FFMPEG_BIN, "-y", - "-ss", seek_str, - "-i", video_url, - "-vframes", "1", - "-q:v", "2", + FFMPEG_BIN, + "-y", + "-ss", + seek_str, + "-i", + video_url, + "-vframes", + "1", + "-q:v", + "2", output_path, ] run_ffmpeg(cmd, capture_output=True, timeout=timeout) @@ -475,6 +487,7 @@ def _call_ai_cover_service( # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧 if primary_video_url: import re as _re + primary_video_url = _re.sub(r"(? Date: Thu, 13 Aug 2026 20:15:15 +0800 Subject: [PATCH 3/6] fix: update error match pattern in test_config_schemas --- tests/unit/test_config_schemas_and_ai_endpoints.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_config_schemas_and_ai_endpoints.py b/tests/unit/test_config_schemas_and_ai_endpoints.py index 7a08f5e09..daf84e5eb 100644 --- a/tests/unit/test_config_schemas_and_ai_endpoints.py +++ b/tests/unit/test_config_schemas_and_ai_endpoints.py @@ -237,7 +237,7 @@ class TestAIRunTasks: from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover - with pytest.raises(RuntimeError, match="MediaKit"): + with pytest.raises(RuntimeError, match="无法从视频抽帧"): run_generate_cover( plan_id="plan-001", asset_ids=["asset-1"], -- 2.54.0 From 768203a794556ba9facf041398f54f0ad931b2bf Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 13 Aug 2026 20:27:42 +0800 Subject: [PATCH 4/6] test: add diff coverage tests for _extract_frames_with_ffmpeg and pre-stored cover path --- tests/unit/test_batch_download.py | 15 ++- tests/unit/test_cover_frame_pre_extract.py | 139 +++++++++++++++++++++ 2 files changed, 153 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_batch_download.py b/tests/unit/test_batch_download.py index 58c669bce..345917875 100755 --- a/tests/unit/test_batch_download.py +++ b/tests/unit/test_batch_download.py @@ -7,6 +7,7 @@ and _download_video_to_file helper. from __future__ import annotations import io +import sys import zipfile from pathlib import Path from unittest.mock import MagicMock, patch @@ -244,7 +245,19 @@ def test_batch_download_single_video(): def test_batch_download_session_closed(): - """DB session is always closed (via finally block).""" + """DB session is always closed (via finally block). + + NOTE: Some prior tests in the full suite replace sys.modules['worker_app.db'] + with a MagicMock, causing 'from worker_app.db import SessionLocal' to get a + mock. We ensure the real module is restored before testing. + """ + # Restore real worker_app.db module if it was mocked by a prior test + import types + if not isinstance(sys.modules.get("worker_app.db"), types.ModuleType): + # Remove the mock and re-import the real module + sys.modules.pop("worker_app.db", None) + import worker_app.db # noqa: F401 — re-import real module + videos = [_FakeVideo("v1", "v.mp4")] session = MagicMock() diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index bef51877e..8033bbfbe 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -345,3 +345,142 @@ class TestCoverTemplatesFix: config={}, ) assert resp.config == {} + + +class TestExtractFramesWithFFmpeg: + """_extract_frames_with_ffmpeg 单元测试.""" + + def test_extracts_frames_with_correct_seek_times(self): + """抽帧时间点正确计算.""" + import subprocess + import tempfile + + # Mock ffprobe to return duration + mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="") + with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc: + # First call is ffprobe, rest are ffmpeg + call_count = 0 + + def side_effect(cmd, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + # ffprobe call + return mock_probe_result + else: + # ffmpeg call - create output file + output_path = cmd[-1] + from pathlib import Path + + Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) + return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") + + mock_subproc.side_effect = side_effect + + from packages.shared.ai_service import _extract_frames_with_ffmpeg + + results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3) + + assert len(results) == 3 + # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0 + assert results[0]["frame_time"] == 5.0 + assert results[1]["frame_time"] == 10.0 + assert results[2]["frame_time"] == 15.0 + + # Clean up + for r in results: + from pathlib import Path + + Path(r["local_path"]).unlink(missing_ok=True) + + def test_handles_ffmpeg_failure(self): + """FFmpeg 失败时跳过该帧.""" + import subprocess + import tempfile + from pathlib import Path + + mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="") + + call_count = 0 + + def side_effect(cmd, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + return mock_probe_result + output_path = cmd[-1] + if call_count == 2: + # First frame succeeds + Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) + return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") + else: + # Other frames fail + raise subprocess.CalledProcessError(1, cmd) + + with patch("subprocess.run", side_effect=side_effect): + from packages.shared.ai_service import _extract_frames_with_ffmpeg + + results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3) + assert len(results) == 1 + Path(results[0]["local_path"]).unlink(missing_ok=True) + + +class TestGenerationCoverPreStored: + """generation_cover.py 预存帧逻辑测试.""" + + def test_pre_stored_candidates_used_when_available(self): + """有预存帧时直接使用，不调用 AI 服务.""" + from unittest.mock import patch + + # Mock the dependencies + mock_plan = MagicMock() + mock_plan.config = { + "cover_candidates": [ + { + "image_url": "https://oss.example.com/covers/p1/frame_0.jpg", + "frame_time": 5.0, + "storage_key": "covers/p1/frame_0.jpg", + }, + { + "image_url": "https://oss.example.com/covers/p1/frame_1.jpg", + "frame_time": 10.0, + "storage_key": "covers/p1/frame_1.jpg", + }, + ], + "rendered_storage_key": "rendered/p1/video.mp4", + } + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + mock_body = MagicMock() + mock_body.asset_ids = ["a1"] + mock_body.cover_type = "ai_frame" + mock_body.frame_time = None + + with ( + patch("app.api.routes.generation_cover.get_editor_services") as mock_services, + patch("app.api.routes.generation_cover.get_db_session"), + patch("app.api.routes.generation_cover.get_current_user"), + patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"), + patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, + ): + + mock_services.return_value = (MagicMock(), mock_plan_svc) + mock_normalize.side_effect = lambda c: c + + from app.api.routes.generation_cover import generate_cover, GenerateCoverRequest + + result = generate_cover( + body=mock_body, + template_id="t1", + plan_id="p1", + services=(MagicMock(), mock_plan_svc), + db=MagicMock(), + current_user=MagicMock(), + ) + + assert result.plan_id == "p1" + assert result.cover["type"] == "ai_frame" + assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg" + assert result.cover["frame_time"] == 5.0 -- 2.54.0 From 823f3fb8a32268d446c1b036c39b675b7e7c0454 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 13 Aug 2026 20:34:05 +0800 Subject: [PATCH 5/6] test: revert test_batch_download.py to develop original (no changes needed) --- tests/unit/test_batch_download.py | 15 +-------------- 1 file changed, 1 insertion(+), 14 deletions(-) diff --git a/tests/unit/test_batch_download.py b/tests/unit/test_batch_download.py index 345917875..58c669bce 100755 --- a/tests/unit/test_batch_download.py +++ b/tests/unit/test_batch_download.py @@ -7,7 +7,6 @@ and _download_video_to_file helper. from __future__ import annotations import io -import sys import zipfile from pathlib import Path from unittest.mock import MagicMock, patch @@ -245,19 +244,7 @@ def test_batch_download_single_video(): def test_batch_download_session_closed(): - """DB session is always closed (via finally block). - - NOTE: Some prior tests in the full suite replace sys.modules['worker_app.db'] - with a MagicMock, causing 'from worker_app.db import SessionLocal' to get a - mock. We ensure the real module is restored before testing. - """ - # Restore real worker_app.db module if it was mocked by a prior test - import types - if not isinstance(sys.modules.get("worker_app.db"), types.ModuleType): - # Remove the mock and re-import the real module - sys.modules.pop("worker_app.db", None) - import worker_app.db # noqa: F401 — re-import real module - + """DB session is always closed (via finally block).""" videos = [_FakeVideo("v1", "v.mp4")] session = MagicMock() -- 2.54.0 From 306e6486416ed2d0ed80d6696eef44618822512a Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 13 Aug 2026 12:38:27 +0000 Subject: [PATCH 6/6] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_cover_frame_pre_extract.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 8033bbfbe..3f462ec14 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -469,7 +469,7 @@ class TestGenerationCoverPreStored: mock_services.return_value = (MagicMock(), mock_plan_svc) mock_normalize.side_effect = lambda c: c - from app.api.routes.generation_cover import generate_cover, GenerateCoverRequest + from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover result = generate_cover( body=mock_body, -- 2.54.0