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
         ],

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'(?<!:)//', '/', primary_video_url)
+
+            primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
         logger.info(
             "获取预览视频URL用于封面生成: plan_id=%s url=%s",
             plan_id,
@@ -182,6 +185,28 @@ 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:

apps/worker/video_processing/render_adapter.py

@@ -74,6 +74,9 @@ 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 +571,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 +621,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(

apps/worker/video_processing/thumbnail_generator.py

@@ -155,3 +155,134 @@ 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

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,

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"(?<!:)//", "/", video_url)
+
+    # 先用 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",
+                video_url,
+            ],
+            capture_output=True,
+            text=True,
+            timeout=15,
+        )
+        if probe_result.returncode == 0 and probe_result.stdout.strip():
+            duration = float(probe_result.stdout.strip())
+    except Exception as e:
+        logger.warning("FFprobe 远程视频时长失败，使用默认值: %s", e)
+
+    ratios = [i / (num_frames + 1) for i in range(1, 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 = 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",
+                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("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"(?<!:)//", "/", primary_video_url)
 
-        from packages.shared.mediakit_client import get_mediakit_client
-
-        client = get_mediakit_client()
-        if client.is_available:
-            # 先检查视频 URL 是否可访问，避免 MediaKit 下载失败后超时
-            try:
-                head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
-                if head_resp.status_code != 200:
-                    logger.error(
-                        "封面视频URL不可访问: plan_id=%s url=%s status=%d",
-                        plan_id,
-                        primary_video_url,
-                        head_resp.status_code,
-                    )
-                    raise RuntimeError(
-                        f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。"
-                        f"请重新生成预览视频后再试。"
-                    )
-            except http_requests.RequestException as e:
-                logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
-                raise RuntimeError(
-                    f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。" f"请重新生成预览视频后再试。"
-                ) from e
-
-            try:
-                logger.info("调用 MediaKit 抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
-                frames = client.extract_frames(
-                    video_url=primary_video_url,
-                    strategy="TimeInterval",  # 按固定时间间隔，稳定性好，不易 OOM
-                    max_frames=3,  # 减少到 3 帧，平衡速度和质量
-                    poll_interval=2.0,  # 缩短轮询间隔
-                    max_poll_attempts=60,  # 120秒超时
+        # 先检查视频 URL 是否可访问
+        try:
+            head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
+            if head_resp.status_code != 200:
+                logger.error(
+                    "封面视频URL不可访问: plan_id=%s url=%s status=%d",
+                    plan_id,
+                    primary_video_url,
+                    head_resp.status_code,
                 )
+                raise RuntimeError(
+                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。" f"请重新生成预览视频后再试。"
+                )
+        except http_requests.RequestException as e:
+            logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
+            raise RuntimeError(
+                f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。请重新生成预览视频后再试。"
+            ) from e
 
-                if frames and len(frames) > 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 是否可访问。")
 
 
 # ── 公共入口 ────────────────────────────────────────────────────────────────

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"],

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

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"],

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):