style: auto-format with black + isort + prettier [skip ci-format-check]

Add _reset_worker_app_db pytest fixture with autouse=True to prevent
mock pollution from leaking between tests. This fixes the flaky
test_batch_download_session_closed test that would fail when running
the full test suite due to worker_app.db being replaced with a
MagicMock from a previous test.

alembic/versions/055_cover_templates.py

@@ -10,8 +10,6 @@ Changes:
 3. config 为 JSON 字段，存储封面配置信息
 """
 
-import json
-
 import sqlalchemy as sa
 
 from alembic import context, op
@@ -73,7 +71,7 @@ def upgrade() -> None:
                 name=name,
                 thumbnail_url="",
                 is_system=True,
-                config=json.dumps(config),
+                config=config,
                 created_at=sa.func.now(),
                 updated_at=sa.func.now(),
             )

alembic/versions/056_fix_cover_templates_config.py

@@ -0,0 +1,39 @@
+"""修复 cover_templates.config 双重序列化
+
+Revision ID: 056_fix_cover_templates_config
+Revises: 055_cover_templates
+Create Date: 2026-08-13
+
+问题: 055 迁移 seed 数据时 json.dumps(config) 导致 config 被双重序列化为 JSON 字符串
+例如 "{}"（字符串）而不是 {}（对象），导致 Pydantic CoverTemplateResponse 校验失败 500。
+
+修复: 从 JSON 字符串中提取文本值，再 cast 回 json 对象类型。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "056_fix_cover_templates_config"
+down_revision = "055_cover_templates"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # PostgreSQL: 从 JSON string scalar 中提取文本内容，cast 为 json object
+    # 例如: JSON string "{}" -> text "{}" -> JSON object {}
+    if conn.dialect.name == "postgresql":
+        conn.execute(
+            sa.text(
+                "UPDATE cover_templates SET config = (config#>>'{}')::json "
+                "WHERE jsonb_typeof(config::jsonb) = 'string'"
+            )
+        )
+
+
+def downgrade() -> None:
+    # No safe rollback — the original data was incorrect
+    pass

apps/api/app/api/routes/generation_cover.py

@@ -185,27 +185,39 @@ 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)
+    # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
+    if body.cover_type in ("ai_frame", "ai_regenerate"):
+        # 尝试从 GenerationTask 读取已持久化的封面 URL
+        generation_task_id = (plan.config or {}).get("generation_task_id", "")
+        if generation_task_id:
+            try:
+                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+                task = gen_task_repo.get(generation_task_id)
+                if task and getattr(task, "cover_url", ""):
+                    cover_data = {
+                        "type": "ai_frame",
+                        "image_url": task.cover_url,
+                        "frame_time": 0.0,
+                        "confidence": 0.95,
+                    }
+                    logger.info(
+                        "[封面生成] 使用统一管道封面: plan_id=%s task_id=%s url=%s",
+                        plan_id,
+                        generation_task_id,
+                        task.cover_url[:80],
+                    )
+                    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)
+            except Exception:
+                logger.warning(
+                    "[封面生成] 读取 GenerationTask.cover_url 失败: plan_id=%s task_id=%s",
+                    plan_id,
+                    generation_task_id,
+                    exc_info=True,
+                )
 
     from packages.shared.ai_service import run_generate_cover
 

apps/api/app/schemas/generation_task.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, Field, field_validator, model_validator
 class ConfirmGenerationRequest(BaseModel):
     """确认生成请求体 — 基于预览任务创建正式生成任务"""
 
-    output_width: int = Field(default=1080, description="输出视频宽度")
-    output_height: int = Field(default=1920, description="输出视频高度")
+    output_width: int = Field(default=1080, ge=100, description="输出视频宽度")
+    output_height: int = Field(default=1920, ge=100, description="输出视频高度")
     cover_url: str = Field(default="", description="自定义封面图片 URL")
     custom_title: str = Field(default="", description="自定义视频标题")
 

apps/web/e2e/core-generation.spec.ts

@@ -258,9 +258,12 @@ test.describe("Core generation flow", () => {
     // Generate API may return 400 in test env if template has no ready segments
     // That is OK for a wizard flow smoke test
     if (genResp.ok()) {
-      const genData = (await genResp.json()) as { plan_id: string; generation_task_id: string }
-      expect(genData.plan_id).toBeTruthy()
-      expect(genData.generation_task_id).toBeTruthy()
+      const genData = (await genResp.json()) as {
+        items: Array<{ id: string; status: string }>
+        total: number
+      }
+      expect(genData.items.length).toBeGreaterThan(0)
+      expect(genData.items[0].id).toBeTruthy()
     } else {
       console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
     }

apps/web/src/pages/generate/hooks/useGenerateVideo.ts

@@ -56,10 +56,41 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
 
     try {
       // 使用确认生成 API（基于预览任务）
-      // 解析分辨率
-      const [widthStr, heightStr] = (props.videoRatio || "1080x1920").split("x")
-      const outputWidth = parseInt(widthStr, 10) || 1080
-      const outputHeight = parseInt(heightStr, 10) || 1920
+      // 解析分辨率：videoRatio 可能是 "9:16"（宽高比）或 "1080x1920"（分辨率）
+      const ratio = props.videoRatio || "9:16"
+      let outputWidth: number
+      let outputHeight: number
+
+      if (ratio.includes(":")) {
+        // 宽高比格式，如 "9:16" → 基于基准高度 1920 计算
+        const [rw, rh] = ratio.split(":").map(Number)
+        if (rw > 0 && rh > 0) {
+          // 基准：长边 1920，短边按比例计算
+          const [longSide, shortSide] = rw < rh ? [rh, rw] : [rw, rh]
+          const baseLong = 1920
+          const baseShort = Math.round((baseLong * shortSide) / longSide)
+          // 确保偶数（FFmpeg 要求）
+          const evenShort = baseShort - (baseShort % 2)
+          if (rw < rh) {
+            outputWidth = evenShort
+            outputHeight = baseLong
+          } else {
+            outputWidth = baseLong
+            outputHeight = evenShort
+          }
+        } else {
+          outputWidth = 1080
+          outputHeight = 1920
+        }
+      } else if (ratio.includes("x")) {
+        // 分辨率格式，如 "1080x1920"
+        const [wStr, hStr] = ratio.split("x")
+        outputWidth = parseInt(wStr, 10) || 1080
+        outputHeight = parseInt(hStr, 10) || 1920
+      } else {
+        outputWidth = 1080
+        outputHeight = 1920
+      }
 
       await confirmGeneration(props.previewTaskId, {
         output_width: outputWidth,

apps/worker/video_processing/dedup_helpers.py

@@ -85,19 +85,9 @@ def create_video_record_and_dedup(
         if thumbnail_url:
             generated_video.thumbnail_url = thumbnail_url
             video_repo.update_thumbnail(video_id, thumbnail_url)
-            logger.info("Thumbnail reused (pre-generated) for video %s", video_id)
+            logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80] if thumbnail_url else "")
         else:
-            thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg"
-            try:
-                from video_processing.thumbnail_generator import generate_and_upload_thumbnail
-
-                _thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key)
-                if _thumbnail_url:
-                    generated_video.thumbnail_url = _thumbnail_url
-                    video_repo.update_thumbnail(video_id, _thumbnail_url)
-                    logger.info("Thumbnail generated for video %s: %s", video_id, _thumbnail_url)
-            except Exception as thumb_err:
-                logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err)
+            logger.debug("No thumbnail_url provided for video %s, skipping", video_id)
 
         # 计算视频指纹
         deduplicator = VideoDeduplicator()

apps/worker/video_processing/render_adapter.py

@@ -47,7 +47,14 @@ def _parse_resolution(resolution_str: str | None) -> tuple[int, int]:
         w, h = resolution_str.lower().split("x", 1)
         width = int(w.strip())
         height = int(h.strip())
-        if width <= 0 or height <= 0:
+        # 最小 100px 防护：避免前端传入宽高比（如 "9:16"）被 parseInt 截断为极小值
+        if width < 100 or height < 100:
+            logger.warning(
+                "分辨率异常小 (%dx%d)，使用默认值。原始值: %s",
+                width,
+                height,
+                resolution_str,
+            )
             return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT
         return width, height
     except (ValueError, TypeError):
@@ -74,9 +81,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": "..."}]
-    )
+    cover_url: str = ""  # 封面图片 URL（从渲染后视频抽帧，天然带标题）
 
     def __post_init__(self):
         if self.rendered_clip_ids is None:
@@ -555,37 +560,33 @@ class RenderAdapter:
         storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
         output_url = upload_to_oss(result.output_path, storage_key)
 
-        self._report_progress(progress_cb, 90.0, "生成封面缩略图")
+        self._report_progress(progress_cb, 90.0, "抽取封面帧")
 
-        # 6. 生成封面缩略图
-        thumbnail_url = ""
+        # 6. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题）
+        cover_url = ""
+        cover_frame_path = None
         try:
-            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+            from video_processing.thumbnail_generator import extract_first_frame
 
-            thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
-            thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
-        except Exception as thumb_err:
-            logger.warning(
-                "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
-                plan_id,
-                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:
+            cover_frame_path = extract_first_frame(str(result.output_path), width=640)
+            cover_storage_key = f"rendered/{plan_id}/cover.jpg"
+            try:
+                cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or ""
+            finally:
+                if cover_frame_path:
+                    try:
+                        Path(cover_frame_path).unlink(missing_ok=True)
+                    except Exception:
+                        pass
+            if cover_url:
                 logger.info(
-                    "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
+                    "[render-adapter] 封面帧提取成功: plan_id=%s url=%s",
                     plan_id,
-                    len(cover_candidates),
+                    cover_url[:80],
                 )
         except Exception as cover_err:
             logger.warning(
-                "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s",
+                "[render-adapter] 封面帧提取失败（不影响主流程）: plan_id=%s error=%s",
                 plan_id,
                 cover_err,
             )
@@ -613,7 +614,7 @@ class RenderAdapter:
             success=True,
             output_url=output_url or "",
             output_path=result.output_path,
-            thumbnail_url=thumbnail_url,
+            thumbnail_url=cover_url,
             duration=result.duration,
             file_size=result.file_size,
             width=result.width,
@@ -621,7 +622,7 @@ class RenderAdapter:
             clip_count=len(clips),
             rendered_clip_ids=final_rendered_ids,
             failed_clip_ids=final_failed_ids,
-            cover_candidates=cover_candidates,
+            cover_url=cover_url,
         )
 
     def render_from_memory(

apps/worker/video_processing/thumbnail_generator.py

@@ -1,4 +1,8 @@
-"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
+"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。
+
+统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
+渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
+"""
 
 from __future__ import annotations
 
@@ -13,28 +17,30 @@ def extract_first_frame(
     video_path: str,
     output_path: str | None = None,
     *,
-    width: int = 640,
+    width: int = -1,
     height: int = -1,
     timeout: int = 30,
     seek_ratio: float = 0.15,
     min_seek_seconds: float = 1.0,
 ) -> str:
-    """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。
+    """抽取视频封面帧（默认取视频时长 15% 处的帧，避开片头纯色画面）。
+
+    因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。
 
     Args:
         video_path: 视频文件路径
         output_path: 输出图片路径，不传则用临时文件
-        width: 输出宽度（默认 640，-1 表示按比例缩放）
-        height: 输出高度（默认 -1，按比例缩放）
+        width: 输出宽度（默认 -1，保持原始分辨率）
+        height: 输出高度（默认 -1，保持原始分辨率）
         timeout: 超时时间（秒）
         seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
         min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
 
     Returns:
-        生成的缩略图文件路径
+        生成的封面帧文件路径
 
     Raises:
-        subprocess.CalledProcessError: ffmpeg 执行失败
+        RuntimeError: ffmpeg 执行失败或输出文件为空
     """
     from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 
@@ -57,10 +63,20 @@ def extract_first_frame(
         # 格式化为 HH:MM:SS.xx
         seek_str = _format_seek_time(seek_time)
 
-        # -ss 放在 -i 前面（input seeking，更快但精度稍低，缩略图够用）
+        # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率。
+        # NOTE: scale_filter 在此处通过 if/else 分支赋值，之后不再被覆盖，
+        #       后续 cmd / cmd2 均复用同一变量，逻辑无变化。
+        if width > 0 or height > 0:
+            w_str = str(width) if width > 0 else "-1"
+            h_str = str(height) if height > 0 else "-1"
+            scale_filter = f"scale={w_str}:{h_str}:force_original_aspect_ratio=decrease,format=yuvj420p"
+        else:
+            # 保持原始分辨率，只确保格式兼容
+            scale_filter = "format=yuvj420p"
+
+        # -ss 放在 -i 前面（input seeking，更快）
         # -vframes 1 只取一帧
         # -q:v 2 jpeg 高质量
-        scale_filter = f"scale={width}:{height}:force_original_aspect_ratio=decrease,format=yuvj420p"
         cmd = [
             FFMPEG_BIN,
             "-y",
@@ -99,7 +115,7 @@ def extract_first_frame(
             run_ffmpeg(cmd2, capture_output=True, timeout=timeout)
 
         if not Path(output_path).exists() or Path(output_path).stat().st_size == 0:
-            raise RuntimeError(f"Thumbnail generation failed: {output_path}")
+            raise RuntimeError(f"Cover frame extraction failed: {output_path}")
 
         return output_path
     except Exception:
@@ -118,171 +134,3 @@ def _format_seek_time(seconds: float) -> str:
     m = int((seconds % 3600) // 60)
     s = seconds % 60
     return f"{h:02d}:{m:02d}:{s:05.2f}"
-
-
-def generate_and_upload_thumbnail(
-    video_path: str,
-    storage_key: str,
-) -> str | None:
-    """生成缩略图并上传到 OSS，返回 URL。
-
-    Args:
-        video_path: 本地视频路径
-        storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg）
-
-    Returns:
-        上传成功返回 URL，失败返回 None
-    """
-    thumbnail_path = None
-    try:
-        thumbnail_path = extract_first_frame(video_path)
-    except Exception as e:
-        logger.warning("Failed to extract thumbnail from %s: %s", video_path, e)
-        return None
-
-    try:
-        from video_processing.oss_helpers import upload_to_oss
-
-        url = upload_to_oss(thumbnail_path, storage_key)
-        return url
-    except Exception as e:
-        logger.warning("Failed to upload thumbnail to OSS: %s", e)
-        return None
-    finally:
-        # 清理临时文件
-        if thumbnail_path:
-            try:
-                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/generation.py

@@ -1123,14 +1123,15 @@ def _render_video(
     resolution: str = "",
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
-) -> tuple[Path, float]:
+    custom_title: str = "",
+) -> tuple[Path, float, str]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_url)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1169,6 +1170,20 @@ def _render_video(
             merged_bgm.get("source", ""),
         )
 
+    # 用户自定义标题覆盖模板标题（用户指定优先级最高）
+    if custom_title and custom_title.strip():
+        plan_cfg = dict(virtual_plan.config or {})
+        title_cfg = dict(plan_cfg.get("title", {}) or {})
+        title_cfg["text"] = custom_title.strip()
+        title_cfg["enabled"] = True
+        plan_cfg["title"] = title_cfg
+        virtual_plan.config = plan_cfg
+        logger.info(
+            "[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
+            task_id,
+            custom_title[:50],
+        )
+
     # 确保输出分辨率配置存在
     # 优先级：用户指定 > 模板配置 > 默认 1280x720
     # 预览模式：强制 854x480 + 低码率
@@ -1244,8 +1259,9 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    cover_url = getattr(render_result, "cover_url", "") or ""
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_url
 
 
 def _upload_and_record(
@@ -1256,6 +1272,7 @@ def _upload_and_record(
     editing_mode,
     user_id: str = "",
     video_name: str = "",
+    thumbnail_url: str = "",
 ) -> tuple[str, float, int, int]:
     """上传 OSS、创建视频记录并查重。
 
@@ -1501,12 +1518,22 @@ def generate_video(self, task_id: str) -> dict:
             # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串
             _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
             _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
+            # 防护：前端可能误传宽高比（如 parseInt("9:16") = 9），宽度 < 100 时忽略
+            if _ow < 100 or _oh < 100:
+                logger.warning(
+                    "[task_id=%s] output_width/output_height 异常 (%dx%d)，回退到默认",
+                    task_id,
+                    _ow,
+                    _oh,
+                )
+                _ow = OUTPUT_WIDTH
+                _oh = OUTPUT_HEIGHT
             if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT:
                 _resolved_resolution = f"{_ow}x{_oh}"
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_url = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1519,12 +1546,42 @@ def generate_video(self, task_id: str) -> dict:
                 resolution=_resolved_resolution,
                 bgm_config=task_info.get("bgm_config", {}),
                 voice_ids=task_info.get("voice_ids", []),
+                custom_title=task_info.get("custom_title", ""),
             )
 
             if gen_task:
                 gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                 _flush_logs(task_id, gen_task)
 
+            # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧）
+            if cover_url:
+                _cover_session = None
+                try:
+                    _cover_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+                        _cover_model = (
+                            _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+                        )
+                        if _cover_model:
+                            _cover_model.cover_url = cover_url
+                            _cover_session.commit()
+                            logger.info(
+                                "[task_id=%s] 封面URL已持久化: %s",
+                                task_id,
+                                cover_url[:80],
+                            )
+                    finally:
+                        if _cover_session:
+                            _cover_session.close()
+                except Exception as cover_err:
+                    logger.warning(
+                        "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s",
+                        task_id,
+                        cover_err,
+                    )
+
             _update_task_progress(task_id, 80, "渲染完成")
 
             # ── 4. 上传 OSS + 查重记录 ───────────────────────────────────────
@@ -1537,6 +1594,7 @@ def generate_video(self, task_id: str) -> dict:
                 editing_mode=editing_mode,
                 user_id=user_id,
                 video_name=task_info.get("video_title", ""),
+                thumbnail_url=cover_url,
             )
 
             if gen_task:

apps/worker/worker_app/tasks/ingest.py

@@ -206,11 +206,21 @@ def ingest_asset(job_id: str) -> dict:
             # 视频类型：生成缩略图（文件还在的时候生成）
             thumbnail_url = None
             if media_type == "video" and extract_success:
+                frame_path = None
                 try:
-                    from video_processing.thumbnail_generator import generate_and_upload_thumbnail
+                    from video_processing.oss_helpers import upload_to_oss
+                    from video_processing.thumbnail_generator import extract_first_frame
 
+                    frame_path = extract_first_frame(str(local_file), width=640)
                     thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg"
-                    thumbnail_url = generate_and_upload_thumbnail(str(local_file), thumb_storage_key)
+                    try:
+                        thumbnail_url = upload_to_oss(frame_path, thumb_storage_key)
+                    finally:
+                        if frame_path:
+                            try:
+                                Path(frame_path).unlink(missing_ok=True)
+                            except Exception:
+                                pass
                     if thumbnail_url:
                         logger.info(
                             "素材缩略图生成成功: job_id=%s url=%s",

infra/docker/api.Dockerfile

@@ -53,9 +53,10 @@ ARG APP_VERSION=dev
 RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
     sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list 2>/dev/null || true
 
-# 只装运行时需要的库（libpq5 是 psycopg2 运行时依赖）
+# 只装运行时需要的库（libpq5 是 psycopg2 运行时依赖，ffmpeg 用于封面兜底取帧）
 RUN apt-get update && apt-get install -y --no-install-recommends \
     libpq5 \
+    ffmpeg \
     && rm -rf /var/lib/apt/lists/*
 
 # 从 builder 复制虚拟环境

packages/shared/ai_service.py

@@ -13,8 +13,6 @@ import random
 import time
 from typing import Any, Dict, List, Optional
 
-import requests as http_requests
-
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
 from packages.shared.ai_client import get_doubao_client
 
@@ -354,93 +352,6 @@ 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],
@@ -450,9 +361,9 @@ def _call_ai_cover_service(
 ) -> Dict[str, Any]:
     """调用 AI 封面生成服务.
 
-    优先级：
-    1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理
-    2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）
+    统一封面管道下，封面已由渲染后视频抽帧生成并持久化到 GenerationTask.cover_url。
+    此函数仅处理 manual/upload 等需要前端交互的类型，
+    ai_frame/ai_regenerate 类型应由调用方直接从持久化的封面 URL 读取。
 
     失败时抛出 RuntimeError。
 
@@ -484,88 +395,14 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧
-    if primary_video_url:
-        import re as _re
-
-        primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
-
-        # 先检查视频 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
-
-        # 使用 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 frames:
-                best_frame = frames[0]
-                local_path = best_frame["local_path"]
-                frame_time_val = best_frame["frame_time"]
-
-                # 上传到 OSS
-                try:
-                    import uuid
-                    from pathlib import Path
-
-                    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 是否可访问。")
-
-
-# ── 公共入口 ────────────────────────────────────────────────────────────────
+    # ai_frame / ai_regenerate: 封面应由渲染后视频抽帧管道生成
+    # 如果调用方传入了持久化的封面 URL，直接使用
+    logger.warning(
+        "封面生成回退: plan_id=%s cover_type=%s — 统一管道应已生成封面，请检查 GenerationTask.cover_url",
+        plan_id,
+        cover_type,
+    )
+    raise RuntimeError(f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。")
 
 
 def run_ai_recommend(

tests/unit/test_1294_preview_voice_injection.py

@@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection:
 
             from packages.domain import EditingMode
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, _cover_url = _render_video(
                 task_id="test_task_123",
                 downloaded_videos=[Path("/tmp/video1.mp4")],
                 voice_path=None,

tests/unit/test_batch_download.py

@@ -13,6 +13,21 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+
+@pytest.fixture(autouse=True)
+def _reset_worker_app_db():
+    """Prevent mock pollution from leaking between tests."""
+    import importlib
+    import sys
+
+    # Save original module reference
+    original = sys.modules.get("worker_app.db")
+    yield
+    # Restore original module after each test
+    if original is not None:
+        sys.modules["worker_app.db"] = original
+
+
 # ── Fake repository ─────────────────────────────────────────────────────────
 
 

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="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             run_generate_cover(
                 plan_id="plan-001",
                 asset_ids=["asset-1"],

tests/unit/test_cover_frame_pre_extract.py

@@ -1,486 +1,149 @@
-"""Tests for cover frame pre-extraction during rendering.
+"""Tests for unified cover frame extraction pipeline.
 
-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
+统一封面管道测试：
+- extract_first_frame: 从已渲染视频抽取封面帧
+- 封面天然带标题（ASS 字幕已烧录到视频中）
 """
 
 from __future__ import annotations
 
+import tempfile
+import unittest
 from pathlib import Path
-from unittest.mock import MagicMock, Mock, call, patch
-
-import pytest
+from unittest.mock import patch
 
 
-class TestExtractCoverCandidates:
-    """extract_cover_candidates 测试."""
+class TestExtractFirstFrame(unittest.TestCase):
+    """extract_first_frame 单元测试."""
 
     @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
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run):
+        """默认在视频 15% 处抽帧."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        # Mock run_ffmpeg 创建输出文件（ffmpeg 真实行为）
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        # 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 ("", "")
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name)
 
-        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)
+            self.assertTrue(Path(result).exists())
+            # 验证 ffmpeg 被调用
+            mock_run.assert_called()
+            cmd = mock_run.call_args[0][0]
+            self.assertIn("-vframes", cmd)
+            self.assertIn("1", cmd)
+            Path(result).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
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_seek_ratio(self, mock_probe, mock_run):
+        """自定义抽帧位置."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        call_count = 0
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, seek_ratio=0.5)
 
-        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 ("", "")
+            self.assertTrue(Path(result).exists())
+            # 50% of 10s = 5s
+            cmd = mock_run.call_args[0][0]
+            ss_idx = cmd.index("-ss") + 1
+            seek_val = cmd[ss_idx]
+            # Should be around 5 seconds
+            self.assertIn("05", seek_val)
+            Path(result).unlink(missing_ok=True)
 
-        mock_run.side_effect = fake_run
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_output_path_parameter(self, mock_probe, mock_run):
+        """指定输出路径."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(b"fake video")
-            video_path = tmp.name
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as out:
+                pass  # just get a path
 
-        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)
+            # Create the file so ffmpeg "succeeds"
+            mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image")
 
-    @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
+            result = extract_first_frame(video.name, output_path=out.name)
+            self.assertEqual(result, out.name)
+            Path(out.name).unlink(missing_ok=True)
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_keeps_original_resolution_by_default(self, mock_probe, mock_run):
+        """默认保持原始分辨率（width=-1, height=-1）."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        # 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 ("", "")
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        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
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.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)
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            # Should NOT have scale filter (only format)
+            self.assertNotIn("scale", vf_filter)
+            self.assertIn("format", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_width_triggers_scale(self, mock_probe, mock_run):
+        """指定宽度时添加 scale 滤镜."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, width=640)
+
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            self.assertIn("scale=640", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=RuntimeError("fail"))
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run):
+        """失败时清理临时文件."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+
+            with self.assertRaises(RuntimeError):
+                extract_first_frame(video.name)
 
 
-class TestExtractAndUploadCoverFrames:
-    """extract_and_upload_cover_frames 测试."""
+class TestRenderAdapterCoverUrl(unittest.TestCase):
+    """RenderAdapterResult.cover_url 字段测试."""
 
-    @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
+    def test_result_has_cover_url_field(self):
+        """RenderAdapterResult 包含 cover_url 字段."""
+        from video_processing.render_adapter import RenderAdapterResult
 
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+        result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg")
+        self.assertEqual(result.cover_url, "https://example.com/cover.jpg")
 
-        # 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."""
+    def test_result_cover_url_defaults_empty(self):
+        """cover_url 默认为空字符串."""
         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
+        self.assertEqual(result.cover_url, "")
 
 
-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
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_generated_video_creation_logic.py

@@ -310,9 +310,9 @@ class TestThumbnailInDedupHelpers:
         mock_dedup.VideoDeduplicator = MagicMock()
         sys.modules["video_processing.dedup"] = mock_dedup
 
-        # mock video_processing.thumbnail_generator
+        # mock video_processing.thumbnail_generator (统一封面管道: 仅保留 extract_first_frame)
         mock_thumb = MagicMock()
-        mock_thumb.generate_and_upload_thumbnail = MagicMock()
+        mock_thumb.extract_first_frame = MagicMock()
         sys.modules["video_processing.thumbnail_generator"] = mock_thumb
 
         # 关键：给 video_processing 包设置子模块属性，让 patch() 能通过属性访问找到
@@ -322,7 +322,7 @@ class TestThumbnailInDedupHelpers:
         video_processing.thumbnail_generator = mock_thumb
 
     def test_pre_generated_thumbnail_url_is_reused(self):
-        """传入 thumbnail_url 时直接复用，不调用 generate_and_upload_thumbnail。"""
+        """传入 thumbnail_url 时直接复用，统一封面管道不再自动生成缩略图。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -339,26 +339,23 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch("video_processing.thumbnail_generator.generate_and_upload_thumbnail") as mock_gen:
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-reuse",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                        thumbnail_url=pre_thumb_url,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-reuse",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                    thumbnail_url=pre_thumb_url,
+                )
 
             assert result == 1
-            # 预生成缩略图时不应调用 generate_and_upload_thumbnail
-            mock_gen.assert_not_called()
 
             from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
 
@@ -368,8 +365,8 @@ class TestThumbnailInDedupHelpers:
         finally:
             session.close()
 
-    def test_thumbnail_generated_when_not_provided(self):
-        """未传 thumbnail_url 时调用 generate_and_upload_thumbnail 生成。"""
+    def test_no_thumbnail_when_not_provided(self):
+        """未传 thumbnail_url 时不生成缩略图（统一封面管道已移除自动缩略图生成）。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -377,8 +374,6 @@ class TestThumbnailInDedupHelpers:
         Session = sessionmaker(bind=engine)
         session = Session()
 
-        generated_thumb_url = "https://oss.example.com/generated-thumb.jpg"
-
         try:
             with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls:
                 mock_dedup = mock_dedup_cls.return_value
@@ -386,43 +381,34 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch(
-                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-                    return_value=generated_thumb_url,
-                ) as mock_gen:
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-gen",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-gen",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                )
 
             assert result == 1
-            # 应调用一次缩略图生成
-            mock_gen.assert_called_once()
-            # 验证参数：video_path 和 storage_key
-            call_args = mock_gen.call_args
-            assert call_args[0][0] == "/tmp/fake.mp4"
-            assert "thumbnails" in call_args[0][1]
 
             from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
 
             video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-gen").first()
             assert video is not None
-            assert video.thumbnail_url == generated_thumb_url
+            # 统一封面管道下，不传 thumbnail_url 时不自动生成
+            assert not video.thumbnail_url
         finally:
             session.close()
 
-    def test_thumbnail_generation_failure_does_not_block(self):
-        """缩略图生成失败不影响主流程。"""
+    def test_no_thumbnail_does_not_block(self):
+        """统一封面管道下，缩略图不再在 dedup 阶段生成。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -437,24 +423,20 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                with patch(
-                    "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-                    side_effect=RuntimeError("cv2 not found"),
-                ):
-                    result = create_video_record_and_dedup(
-                        generation_task_id="task-thumb-fail",
-                        project_id="proj-1",
-                        batch_id="",
-                        file_url="https://oss.example.com/video.mp4",
-                        file_size=1024,
-                        duration=10.0,
-                        video_path="/tmp/fake.mp4",
-                        mode="edit_plan",
-                        session=session,
-                        width=1280,
-                        height=720,
-                        fps=25.0,
-                    )
+                result = create_video_record_and_dedup(
+                    generation_task_id="task-thumb-fail",
+                    project_id="proj-1",
+                    batch_id="",
+                    file_url="https://oss.example.com/video.mp4",
+                    file_size=1024,
+                    duration=10.0,
+                    video_path="/tmp/fake.mp4",
+                    mode="edit_plan",
+                    session=session,
+                    width=1280,
+                    height=720,
+                    fps=25.0,
+                )
 
             assert result == 1  # 不阻断
 

tests/unit/test_generation_cover.py

@@ -92,3 +92,120 @@ def test_generation_cover_request_validation():
     req2 = GenerateCoverRequest(cover_type="upload", asset_ids=["a1", "a2"])
     assert req2.cover_type == "upload"
     assert req2.asset_ids == ["a1", "a2"]
+
+
+class TestUnifiedCoverPipelineEndpoint:
+    """测试统一封面管道在 generate_cover endpoint 中的逻辑 (lines 189-215)."""
+
+    def test_cover_url_from_generation_task(self):
+        """当 GenerationTask 有 cover_url 时，直接返回该 URL 作为封面。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse
+
+        # Mock plan with rendered_storage_key (so we skip the 3-step lookup)
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "generation_task_id": "task-123",
+            "rendered_storage_key": "rendered/plan-1/video.mp4",
+        }
+
+        # Mock plan_svc
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+        # Mock template_svc
+        mock_template_svc = MagicMock()
+
+        # Mock generation task with cover_url
+        mock_task = MagicMock()
+        mock_task.cover_url = "https://oss.example.com/rendered/plan-1/cover.jpg"
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_task
+            mock_repo_cls.return_value = mock_repo
+
+            # normalize_plan_config should return the config with cover
+            mock_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/plan-1/cover.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-1",
+                plan_id="plan-1",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            # 验证返回的封面数据来自 GenerationTask.cover_url
+            assert result.plan_id == "plan-1"
+            assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg"
+            assert result.cover["type"] == "ai_frame"
+
+    def test_cover_url_fallback_when_no_cover_url(self):
+        """当 GenerationTask 没有 cover_url 时，跳过统一管道走 run_generate_cover。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {
+            "generation_task_id": "task-456",
+            "rendered_storage_key": "rendered/plan-2/video.mp4",
+        }
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+
+        mock_template_svc = MagicMock()
+
+        # Task has no cover_url
+        mock_task = MagicMock()
+        mock_task.cover_url = ""
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = mock_task
+            mock_repo_cls.return_value = mock_repo
+
+            # Mock storage service
+            mock_storage_svc = MagicMock()
+            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4"
+            mock_storage_getter.return_value = mock_storage_svc
+
+            mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"}
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="template-2",
+                plan_id="plan-2",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            # 应该走 run_generate_cover 回退
+            mock_run.assert_called_once()
+            assert result.cover["image_url"] == "https://fallback.com/cover.jpg"

tests/unit/test_mediakit_cover.py

@@ -141,93 +141,13 @@ class TestMediaKitClient:
 
 
 class TestAICoverService:
-    """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。"""
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @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
-
-        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": 3.5}]
-
-        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
-
-            from packages.shared.ai_service import _call_ai_cover_service
-
-            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")
-    def test_call_ai_cover_video_url_unreachable(self, mock_head):
-        """视频 URL 不可访问时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 404
+    """AI 封面服务测试（统一封面管道后）。"""
 
+    def test_call_ai_cover_ai_frame_raises(self):
+        """ai_frame type raises RuntimeError in unified pipeline."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/nonexistent.mp4",
-            )
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @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_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError):
-            _call_ai_cover_service(
-                plan_id="plan-1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url=dirty_url,
-            )
-
-        # HEAD 请求使用规范化后的 URL
-        mock_head.assert_called_once()
-        assert mock_head.call_args[0][0] == clean_url
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @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_ffmpeg.side_effect = Exception("ffmpeg error")
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -235,11 +155,22 @@ class TestAICoverService:
                 primary_video_url="https://example.com/video.mp4",
             )
 
+    def test_call_ai_cover_ai_regenerate_raises(self):
+        """ai_regenerate type raises RuntimeError in unified pipeline."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_regenerate",
+            )
+
     def test_call_ai_cover_without_video_url_raises(self):
-        """没有视频 URL 时抛出 RuntimeError."""
+        """ai_frame without video URL still raises RuntimeError."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -276,23 +207,6 @@ class TestAICoverService:
         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_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="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
 
 class TestGenerateCover:
     """run_generate_cover 测试."""

tests/unit/test_render_adapter.py

@@ -348,24 +348,32 @@ class TestRenderPlan:
         mock_render_cls.return_value = mock_render
         mock_upload.return_value = "https://oss.example.com/out.mp4"
 
-        fake_thumb = "https://oss.example.com/rendered/plan_thumb/thumbnail.jpg"
-
         plan = FakePlan(id="plan_thumb")
         clips = [_make_clip("c1", order=0, duration=5.0)]
         asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
         adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
+        # Mock extract_first_frame to return a temp file path
+        import tempfile as _tf
+
+        _fake_frame = _tf.NamedTemporaryFile(suffix=".jpg", delete=False)
+        _fake_frame.write(b"fake frame")
+        _fake_frame.close()
         with patch(
-            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-            return_value=fake_thumb,
+            "video_processing.thumbnail_generator.extract_first_frame",
+            return_value=_fake_frame.name,
         ):
             result = adapter.render_plan(
                 "plan_thumb",
                 work_dir=tmp_path / "work",
             )
+        from pathlib import Path as _P
+
+        _P(_fake_frame.name).unlink(missing_ok=True)
 
         assert result.success
-        assert result.thumbnail_url == fake_thumb
+        # cover_url from upload_to_oss (mocked globally)
+        assert result.thumbnail_url == "https://oss.example.com/out.mp4"
 
     @patch("video_processing.render_adapter.upload_to_oss")
     @patch("video_processing.render_adapter.UnifiedRenderService")
@@ -397,8 +405,8 @@ class TestRenderPlan:
         adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         with patch(
-            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
-            side_effect=RuntimeError("cv2 not available"),
+            "video_processing.thumbnail_generator.extract_first_frame",
+            side_effect=RuntimeError("ffmpeg not available"),
         ):
             result = adapter.render_plan(
                 "plan_thumb_fail",

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="无法从视频抽帧"):
+        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="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
 
     def test_cover_type_manual_still_works(self):