diff --git a/apps/api/app/api/routes/cover_templates.py b/apps/api/app/api/routes/cover_templates.py index f42cbfda2..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 @@ -55,7 +54,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..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'(? 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..58155e4ab 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -256,6 +256,17 @@ 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..2ae55042e 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -354,6 +354,93 @@ 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 +450,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 +461,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,80 +484,85 @@ 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，直接报错 - raise RuntimeError( - f"封面生成失败: plan_id={plan_id}, MediaKit 不可用或抽帧失败。" f"请检查 primary_video_url 是否可访问。" - ) + 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}，无法从视频抽帧。请检查 primary_video_url 是否可访问。") # ── 公共入口 ──────────────────────────────────────────────────────────────── 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"], 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..3f462ec14 --- /dev/null +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -0,0 +1,486 @@ +"""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 == {} + + +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 GenerateCoverRequest, generate_cover + + 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 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):