fix: add ffmpeg to API runtime image for cover fallback

PR #1360 introduced FFmpeg-based frame extraction as fallback for cover
generation, but the API Docker image did not include FFmpeg binary.
Worker image already has it via worker-base-runtime, but API image only
had libpq5 in its runtime stage.

This caused subprocess calls to fail with FileNotFoundError when the
cover generation tried to use _extract_frames_with_ffmpeg fallback.

Fix: add ffmpeg to apt-get install in runtime stage of api.Dockerfile.

alembic/versions/055_cover_templates.py

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

alembic/versions/056_fix_cover_templates_config.py

@@ -1,39 +0,0 @@
-"""修复 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

@@ -31,8 +31,6 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=["Generation"])
 
 
-
-
 # ── Schemas ──────────────────────────────────────────────────────────────
 
 
@@ -107,6 +105,10 @@ def generate_cover(
                             generation_task_id,
                             rendered_storage_key[:80],
                         )
+                        logger.info(
+                            plan_id,
+                            generation_task_id,
+                        )
             except Exception:
                 logger.warning(
                     "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
@@ -114,33 +116,6 @@ def generate_cover(
                     exc_info=True,
                 )
 
-    # 第 2.5 步：通过 plan_id 作为 source_edit_plan_id 查找关联的已完成预览任务
-    if not rendered_storage_key:
-        try:
-            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-            logger.info("[封面生成] 步骤2.5: 通过 source_edit_plan_id 查找: plan_id=%s", plan_id)
-            preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-            for pt in preview_tasks:
-                if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False):
-                    video_repo = get_generated_video_repository(db)
-                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
-                    videos = use_case.execute(pt.id)
-                    if videos:
-                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
-                        logger.info(
-                            "[封面生成] ✅ 步骤2.5找到视频: plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            pt.id,
-                            rendered_storage_key[:80],
-                        )
-                        break
-        except Exception:
-            logger.warning(
-                "封面生成: 通过 source_edit_plan_id 查找预览任务失败: plan_id=%s",
-                plan_id,
-                exc_info=True,
-            )
-
     # 第三步：按 user + template 查找最近的已完成预览任务（兜底）
     if not rendered_storage_key:
         try:
@@ -210,103 +185,28 @@ def generate_cover(
             detail=f"获取预览视频URL失败: {e}",
         ) from e
 
-    # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
-    # 多步查找 cover_url，和查找视频 URL 一样的 fallback 逻辑
-    if body.cover_type in ("ai_frame", "ai_regenerate"):
-        cover_url_from_task = None
-        gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
-
-        # 步骤 A：通过 generation_task_id 直接查找
-        generation_task_id = (plan.config or {}).get("generation_task_id", "")
-        if generation_task_id:
-            try:
-                task = gen_task_repo.get(generation_task_id)
-                if task and getattr(task, "cover_url", ""):
-                    cover_url_from_task = task.cover_url
-                    logger.info(
-                        "[封面生成] 统一管道封面(步骤A-direct): plan_id=%s task_id=%s url=%s",
-                        plan_id,
-                        generation_task_id,
-                        cover_url_from_task[:80],
-                    )
-            except Exception:
-                logger.warning(
-                    "[封面生成] 步骤A读取 cover_url 失败: plan_id=%s task_id=%s",
-                    plan_id,
-                    generation_task_id,
-                    exc_info=True,
-                )
-
-        # 步骤 B：通过 source_edit_plan_id 查找关联预览任务的 cover_url
-        if not cover_url_from_task:
-            try:
-                preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
-                for pt in preview_tasks:
-                    if getattr(pt, "status", "") == "completed" and getattr(pt, "cover_url", ""):
-                        cover_url_from_task = pt.cover_url
-                        logger.info(
-                            "[封面生成] 统一管道封面(步骤B-source_plan): plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            pt.id,
-                            cover_url_from_task[:80],
-                        )
-                        break
-            except Exception:
-                logger.warning(
-                    "[封面生成] 步骤B查找 cover_url 失败: plan_id=%s",
-                    plan_id,
-                    exc_info=True,
-                )
-
-        # 步骤 C：通过 user+template 查找最近的已完成预览任务的 cover_url
-        if not cover_url_from_task:
-            try:
-                preview_tasks = gen_task_repo.list_latest_completed_preview(
-                    user_id=str(current_user.user.id),
-                    template_id=template_id,
-                )
-                for pt in preview_tasks:
-                    if getattr(pt, "cover_url", ""):
-                        cover_url_from_task = pt.cover_url
-                        logger.info(
-                            "[封面生成] 统一管道封面(步骤C-user+template): plan_id=%s task_id=%s url=%s",
-                            plan_id,
-                            pt.id,
-                            cover_url_from_task[:80],
-                        )
-                        break
-            except Exception:
-                logger.warning(
-                    "[封面生成] 步骤C查找 cover_url 失败: plan_id=%s template_id=%s",
-                    plan_id,
-                    template_id,
-                    exc_info=True,
-                )
-
-        if cover_url_from_task:
-            # 标题已在预览视频渲染时烧录（ASS字幕），封面帧自然包含标题
-            cover_data = {
-                "type": "ai_frame",
-                "image_url": cover_url_from_task,
-                "frame_time": 0.0,
-                "confidence": 0.95,
-            }
+    # 优先使用渲染时预抽的封面候选帧（跳过 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)
 
-        logger.warning(
-            "[封面生成] 统一管道未找到 cover_url: plan_id=%s",
-            plan_id,
-        )
-        # ai_frame/ai_regenerate 类型必须从渲染管道获取，不再回退到 AI 服务
-        raise HTTPException(
-            status_code=400,
-            detail="封面尚未生成，请先重新生成预览视频以触发封面自动提取",
-        )
-
     from packages.shared.ai_service import run_generate_cover
 
     try:

apps/api/app/api/routes/generation_preview.py

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import json
 import logging
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -268,20 +267,6 @@ def create_preview_generation_task(
     # 从模板读取 editing_mode / mode 作为 strategy_id（渲染 pipeline 的 mode 参数）
     strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
 
-    # 处理标题配置：如果有标题文本，序列化到 custom_title 字段传递给 worker
-    title_config = request.title_config or {}
-    title_text = (title_config.get("text") or "").strip()
-    custom_title_value = ""
-    if title_text:
-        # 将标题文本和样式配置序列化为 JSON 存入 custom_title
-        # Worker 端会解析 JSON 获取完整标题配置
-        custom_title_value = json.dumps(title_config, ensure_ascii=False)
-        logger.info(
-            "[预览生成] 标题配置: text=%s, config_keys=%s",
-            title_text[:30],
-            list(title_config.keys()),
-        )
-
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
 
     try:
@@ -305,7 +290,6 @@ def create_preview_generation_task(
                 auto_retry_enabled=False,
                 auto_retry_max=0,
                 is_preview=True,
-                custom_title=custom_title_value,
             )
         )
     except ValueError as e:

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, ge=100, description="输出视频宽度")
-    output_height: int = Field(default=1920, ge=100, description="输出视频高度")
+    output_width: int = Field(default=1080, description="输出视频宽度")
+    output_height: int = Field(default=1920, description="输出视频高度")
     cover_url: str = Field(default="", description="自定义封面图片 URL")
     custom_title: str = Field(default="", description="自定义视频标题")
 
@@ -181,10 +181,6 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
         default="",
         description="关联的编辑计划ID（可选），用于确认生成时复用预览产物",
     )
-    title_config: dict = Field(
-        default_factory=dict,
-        description="标题配置（可选），渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow",
-    )
 
     @model_validator(mode="after")
     def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":

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

@@ -200,7 +200,15 @@ test.describe("Core generation flow", () => {
     await expect(page.getByRole("heading", { name: /选择配音/ })).toBeVisible({ timeout: 15000 })
     await page.getByRole("button", { name: "下一步" }).click()
 
-    // Step 4: title（新顺序：标题在预览之前）
+    // Step 4: preview — 需要先生成预览视频，才能进入下一步
+    await expect(page.getByRole("heading", { name: /生成预览/ })).toBeVisible({ timeout: 15000 })
+    // 点击"生成预览"按钮触发预览生成
+    await page.locator(".xx-preview-generate-btn").click()
+    // 等待预览生成完成（后端渲染，可能需要较长时间）
+    await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
+    await page.getByRole("button", { name: "下一步" }).click()
+
+    // Step 5: title
     await expect(page.getByRole("heading", { name: /选择标题/ })).toBeVisible({ timeout: 15000 })
     // 等待组件完全渲染
     await page.waitForTimeout(2000)
@@ -214,14 +222,6 @@ test.describe("Core generation flow", () => {
     await titleInput.fill(titleText)
     await page.getByRole("button", { name: "下一步" }).click()
 
-    // Step 5: preview — 需要先生成预览视频，才能进入下一步
-    await expect(page.getByRole("heading", { name: /生成预览/ })).toBeVisible({ timeout: 15000 })
-    // 点击"生成预览"按钮触发预览生成
-    await page.locator(".xx-preview-generate-btn").click()
-    // 等待预览生成完成（后端渲染，可能需要较长时间）
-    await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
-    await page.getByRole("button", { name: "下一步" }).click()
-
     // Step 6: cover (默认 AI 智能选帧模式，直接下一步)
     await expect(page.getByRole("heading", { name: /选择封面/ })).toBeVisible({ timeout: 15000 })
     await page.getByRole("button", { name: "下一步" }).click()
@@ -258,12 +258,9 @@ 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 {
-        items: Array<{ id: string; status: string }>
-        total: number
-      }
-      expect(genData.items.length).toBeGreaterThan(0)
-      expect(genData.items[0].id).toBeTruthy()
+      const genData = (await genResp.json()) as { plan_id: string; generation_task_id: string }
+      expect(genData.plan_id).toBeTruthy()
+      expect(genData.generation_task_id).toBeTruthy()
     } else {
       console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
     }

apps/web/src/api/generation/types.ts

@@ -13,17 +13,6 @@ export interface CreatePreviewRequest {
   video_title?: string
   duration?: number
   video_ratio?: string
-  /* 标题烧录配置（可选，传入后 ASS 渲染标题到预览视频中） */
-  title_config?: {
-    text?: string
-    font?: string
-    font_size?: number
-    font_color?: string
-    position?: string
-    bold?: boolean
-    stroke?: boolean
-    shadow?: boolean
-  }
   bgm_config?: {
     enabled: boolean
     preset_id?: string

apps/web/src/pages/generate/GeneratePage.tsx

@@ -1,6 +1,6 @@
 /**
  * 智能剪辑页面 — V22 多预览 + 配音前置
- * 7 步向导：选择模板 → 选择素材 → 选择配音 → 选择标题 → 生成预览 → 选择封面 → 确认生成
+ * 7 步向导：选择模板 → 选择素材 → 选择配音 → 生成预览 → 选择标题 → 选择封面 → 确认生成
  * 左右布局：左侧 generate-form + 右侧 generate-preview
  * 主组件仅保留整体布局与事件编排
  * 状态管理 → hooks/useGenerateFormState
@@ -12,10 +12,7 @@
 import React, { useState, useMemo } from "react"
 import { Modal, message } from "antd"
 import { useNavigate } from "react-router-dom"
-import { useQuery } from "@tanstack/react-query"
 import type { VoiceClone } from "@/api/voice-clone"
-import { getAssetsByKind } from "@/api/assets/assets"
-import type { AssetItem } from "@/api/assets/types"
 import { useCloneProgress } from "@/hooks/useCloneProgress"
 import CloneModal from "@/components/voice/CloneModal"
 import GenerateHeader from "./components/GenerateHeader"
@@ -27,7 +24,7 @@ import GenerateStepActions from "./components/GenerateStepActions"
 import { useGenerateFormState } from "./hooks/useGenerateFormState"
 import { useStepNavigation } from "./hooks/useStepNavigation"
 import { useGenerateVideo } from "./hooks/useGenerateVideo"
-import { useStep5Preview } from "./hooks/useStep5Preview"
+import { useStep4Preview } from "./hooks/useStep4Preview"
 import "./generate.css"
 
 const GeneratePage: React.FC = () => {
@@ -74,20 +71,6 @@ const GeneratePage: React.FC = () => {
     setPreviewModalOpen,
   } = formState
 
-  /* ── 查询视频素材，用于 Step4 标题预览背景 ── */
-  const { data: videoAssets = [] } = useQuery({
-    queryKey: ["generate-video-assets"],
-    queryFn: () => getAssetsByKind("video", { limit: 50 }),
-  })
-
-  // 获取第一个选中素材的 URL
-  const sourceVideoUrl = useMemo(() => {
-    const firstId = selectedMaterials[0]
-    if (!firstId) return undefined
-    const asset = videoAssets.find((a: AssetItem) => a.id === firstId)
-    return asset?.file_url
-  }, [selectedMaterials, videoAssets])
-
   /* ── 克隆声音 ── */
   const { clones: clonedVoices, addClone, hasProcessing } = useCloneProgress()
 
@@ -113,8 +96,8 @@ const GeneratePage: React.FC = () => {
     return id ? [id] : []
   }, [voiceMode, selectedVoice, selectedClonedVoice])
 
-  /* ── Step5 预览生成（多预览 + voice_ids） ── */
-  const step5Preview = useStep5Preview({
+  /* ── Step4 预览生成（多预览 + voice_ids） ── */
+  const step4Preview = useStep4Preview({
     templates: userTemplates,
     selectedTemplate,
     materialMode,
@@ -125,7 +108,6 @@ const GeneratePage: React.FC = () => {
     voiceIds: previewVoiceIds,
     voiceLibraryId: selectedVoice || undefined,
     previewCount,
-    titleSettings,
   })
 
   /* ── 步骤导航 ── */
@@ -137,7 +119,7 @@ const GeneratePage: React.FC = () => {
     selectedMaterials,
     smartSelectedIds,
     titleSettings,
-    previewReady: step5Preview.canProceed,
+    previewReady: step4Preview.canProceed,
   })
 
   /* ── 视频生成核心逻辑 ── */
@@ -168,7 +150,7 @@ const GeneratePage: React.FC = () => {
     autoSubtitles,
     bgm,
     generateCount,
-    previewTaskId: step5Preview.selectedTaskId,
+    previewTaskId: step4Preview.selectedTaskId,
   })
 
   /* ================================================================
@@ -225,20 +207,20 @@ const GeneratePage: React.FC = () => {
             onDismissError={handleDismissError}
             presetVoices={presetVoices}
             videoRatio={videoRatio}
-            /* Step5 多预览 */
+            /* Step4 多预览 */
             previewCount={previewCount}
             onPreviewCountChange={setPreviewCount}
-            previewItems={step5Preview.items}
-            previewSelectedIndex={step5Preview.selectedIndex}
-            onSelectPreview={step5Preview.setSelectedIndex}
-            previewOverallStatus={step5Preview.previewStatus}
-            previewOverallError={step5Preview.previewError}
-            previewOverallProgress={step5Preview.progress}
-            previewAnyGenerating={step5Preview.anyGenerating}
-            previewTemplateName={step5Preview.templateName}
-            previewMaterialCount={step5Preview.materialCount}
-            onGeneratePreview={step5Preview.generatePreview}
-            onRegeneratePreview={step5Preview.regeneratePreview}
+            previewItems={step4Preview.items}
+            previewSelectedIndex={step4Preview.selectedIndex}
+            onSelectPreview={step4Preview.setSelectedIndex}
+            previewOverallStatus={step4Preview.previewStatus}
+            previewOverallError={step4Preview.previewError}
+            previewOverallProgress={step4Preview.progress}
+            previewAnyGenerating={step4Preview.anyGenerating}
+            previewTemplateName={step4Preview.templateName}
+            previewMaterialCount={step4Preview.materialCount}
+            onGeneratePreview={step4Preview.generatePreview}
+            onRegeneratePreview={step4Preview.regeneratePreview}
           />
 
           <GenerateStepActions
@@ -254,24 +236,22 @@ const GeneratePage: React.FC = () => {
 
         {/* ════ 右侧：预览 + 生成结果 ════ */}
         <div className="xx-generate-right-col">
-          {/* 预览视频面板（Step4+ 常驻，Step4 显示标题预览，Step5+ 显示预览视频） */}
+          {/* 预览视频面板（Step4+ 常驻，展示选中的预览） */}
           {currentStep >= 4 && (
             <PreviewVideoPanel
-              previewStatus={step5Preview.previewStatus}
-              previewResult={step5Preview.previewResult}
-              previewError={step5Preview.previewError}
-              progress={step5Preview.progress}
+              previewStatus={step4Preview.previewStatus}
+              previewResult={step4Preview.previewResult}
+              previewError={step4Preview.previewError}
+              progress={step4Preview.progress}
               videoRatio={videoRatio}
-              onRegenerate={step5Preview.regeneratePreview}
+              onRegenerate={step4Preview.regeneratePreview}
               titleText={titleSettings.title}
-              titleSettings={titleSettings}
-              showTitlePreview={currentStep === 4}
-              sourceVideoUrl={sourceVideoUrl}
+              titleSettings={currentStep >= 5 ? titleSettings : undefined}
             />
           )}
 
-          {/* 正式生成结果（Step6+ 才显示） */}
-          {currentStep >= 6 && (
+          {/* 正式生成结果（Step5+ 才显示） */}
+          {currentStep >= 5 && (
             <GenerateResultPanel
               generated={generated}
               generating={generating}

apps/web/src/pages/generate/components/GenerateStepContent.tsx

@@ -1,7 +1,7 @@
 /**
  * GeneratePage 步骤内容渲染
  * 根据当前步骤渲染对应的 Step 组件
- * 步骤顺序：模板(1) → 素材(2) → 配音(3) → 标题(4) → 预览(5) → 封面(6) → 确认(7)
+ * 步骤顺序：模板(1) → 素材(2) → 配音(3) → 预览(4) → 标题(5) → 封面(6) → 确认(7)
  */
 import React from "react"
 import type { EditingTemplate } from "@/api/editing-planner"
@@ -9,12 +9,12 @@ import type { PresetVoiceItem } from "@/api/voices"
 import type { VoiceClone } from "@/api/voice-clone"
 import type { CoverConfig } from "../types/cover"
 import type { TitleSettings } from "../types"
-import type { PreviewItem, PreviewStatus } from "../hooks/useStep5Preview"
+import type { PreviewItem, PreviewStatus } from "../hooks/useStep4Preview"
 import Step1TemplateSelect from "../components/Step1TemplateSelect"
 import Step2MaterialSelect from "../components/Step2MaterialSelect"
 import Step3VoiceSelect from "../components/Step5VoiceSelect"
-import Step5GeneratePreview from "../components/Step5GeneratePreview"
-import Step4TitleSettings from "../components/Step4TitleSettings"
+import Step4GeneratePreview from "../components/Step4GeneratePreview"
+import Step5TitleSettings from "../components/Step4TitleSettings"
 import Step6CoverSettings from "../components/Step6CoverSettings"
 import Step7ConfirmGenerate from "../components/Step7ConfirmGenerate"
 import type { GeneratedVideo } from "@/api/template-editor"
@@ -157,14 +157,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
       )
     case 4:
       return (
-        <Step4TitleSettings
-          titleSettings={titleSettings}
-          onTitleSettingsChange={onTitleSettingsChange}
-        />
-      )
-    case 5:
-      return (
-        <Step5GeneratePreview
+        <Step4GeneratePreview
           templateName={previewTemplateName}
           materialCount={previewMaterialCount}
           duration={duration}
@@ -182,6 +175,13 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           onRegeneratePreview={onRegeneratePreview}
         />
       )
+    case 5:
+      return (
+        <Step5TitleSettings
+          titleSettings={titleSettings}
+          onTitleSettingsChange={onTitleSettingsChange}
+        />
+      )
     case 6:
       return (
         <Step6CoverSettings

apps/web/src/pages/generate/components/PreviewVideoPanel.tsx

@@ -13,10 +13,8 @@
  */
 import React, { useRef, useEffect, useCallback } from "react"
 import { PlayCircleOutlined, LoadingOutlined } from "@ant-design/icons"
-import type { PreviewResult, PreviewStatus } from "../hooks/useStep5Preview"
+import type { PreviewResult, PreviewStatus } from "../hooks/useStep4Preview"
 import type { TitleSettings } from "../types"
-import { drawTitleOnCanvas } from "../utils/drawTitleOnCanvas"
-import TitlePreviewCanvas from "./title/TitlePreviewCanvas"
 
 interface PreviewVideoPanelProps {
   previewStatus: PreviewStatus
@@ -25,14 +23,134 @@ interface PreviewVideoPanelProps {
   progress: number
   videoRatio: string
   onRegenerate: () => void
-  /** 标题文字 */
+  /** 标题文字（Step5 起传入） */
   titleText?: string
-  /** 标题样式设置 */
+  /** 标题样式设置（Step5 起传入） */
   titleSettings?: TitleSettings
-  /** Step4 标题预览模式 */
-  showTitlePreview?: boolean
-  /** 素材视频 URL（用于 Step4 标题预览背景） */
-  sourceVideoUrl?: string
+}
+
+/* ── Canvas 绘制工具函数 ── */
+
+/**
+ * 将文本按 maxWidth 逐字换行，返回行数组。
+ * 与 ASS 字幕引擎的逐字换行行为一致。
+ */
+function wrapText(ctx: CanvasRenderingContext2D, text: string, maxWidth: number): string[] {
+  const lines: string[] = []
+  let currentLine = ""
+  for (const char of text) {
+    const testLine = currentLine + char
+    if (ctx.measureText(testLine).width > maxWidth && currentLine) {
+      lines.push(currentLine)
+      currentLine = char
+    } else {
+      currentLine = testLine
+    }
+  }
+  if (currentLine) lines.push(currentLine)
+  return lines
+}
+
+/**
+ * 在 canvas 上绘制标题文字（含描边/阴影/多行居中）
+ *
+ * Canvas 已通过 CSS 定位到视频实际渲染位置，
+ * 坐标系基于 Canvas 自身尺寸，居中直接使用 w/2。
+ *
+ * @param ctx canvas 上下文
+ * @param w canvas CSS 宽度（= 视频渲染宽度）
+ * @param h canvas CSS 高度（= 视频渲染高度）
+ * @param text 标题文字
+ * @param settings 标题样式
+ * @param paddingX 左右边距（px），与 ASS 的 MarginL/MarginR 对应
+ * @param position "top" | "center" | "bottom"
+ * @param topOffset 顶部/底部偏移量
+ */
+function drawTitleOnCanvas(
+  ctx: CanvasRenderingContext2D,
+  w: number,
+  h: number,
+  text: string,
+  settings: TitleSettings,
+  paddingX: number,
+  position: string,
+  topOffset: number,
+) {
+  const dpr = window.devicePixelRatio || 1
+
+  // 设置 canvas 物理像素尺寸（高清屏适配）
+  ctx.canvas.width = Math.round(w * dpr)
+  ctx.canvas.height = Math.round(h * dpr)
+  ctx.scale(dpr, dpr)
+
+  // 清除
+  ctx.clearRect(0, 0, w, h)
+
+  // 可用宽度 = 总宽 - 左右边距
+  const availableWidth = w - paddingX * 2
+  if (availableWidth <= 0) return
+
+  // 字体设置
+  const fontSize = Math.round(Math.min(settings.size, 36))
+  const fontWeight = settings.bold ? "bold" : "normal"
+  const fontStyle = settings.italic ? "italic" : "normal"
+  ctx.font = `${fontStyle} ${fontWeight} ${fontSize}px "${settings.font}"`
+
+  // 文字属性
+  ctx.textAlign = "center"
+  ctx.textBaseline = "middle"
+  ctx.fillStyle = settings.color
+
+  const lineHeight = fontSize * 1.4
+
+  // 描边 & 阴影
+  if (settings.stroke) {
+    ctx.strokeStyle = "rgba(0,0,0,0.6)"
+    ctx.lineWidth = 2
+    ctx.lineJoin = "round"
+  }
+  if (settings.shadow) {
+    ctx.shadowColor = "rgba(0,0,0,0.7)"
+    ctx.shadowBlur = 4
+    ctx.shadowOffsetX = 2
+    ctx.shadowOffsetY = 2
+  }
+
+  // 换行
+  const displayText = text && text.trim() ? text : "请选择或输入标题"
+  const lines = wrapText(ctx, displayText, availableWidth)
+
+  // 起始 Y：根据 position 计算
+  const totalTextHeight = lines.length * lineHeight
+  let startY: number
+  switch (position) {
+    case "top":
+      startY = topOffset
+      break
+    case "center":
+      startY = (h - totalTextHeight) / 2 + lineHeight / 2
+      break
+    case "bottom":
+    default:
+      startY = h - topOffset - totalTextHeight + lineHeight / 2
+      break
+  }
+
+  // 居中 x = w/2（Canvas 已定位到视频位置，无需额外偏移）
+  const x = w / 2
+  lines.forEach((line, i) => {
+    const y = startY + i * lineHeight
+    if (settings.stroke) ctx.strokeText(line, x, y)
+    ctx.fillText(line, x, y)
+  })
+
+  // 重置 shadow（避免影响后续绘制）
+  if (settings.shadow) {
+    ctx.shadowColor = "transparent"
+    ctx.shadowBlur = 0
+    ctx.shadowOffsetX = 0
+    ctx.shadowOffsetY = 0
+  }
 }
 
 /* ── 组件 ── */
@@ -46,12 +164,11 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
   onRegenerate,
   titleText,
   titleSettings,
-  showTitlePreview,
-  sourceVideoUrl,
 }) => {
   const hasPreview = previewStatus === "ready" && previewResult
   const isLoading = previewStatus === "pending" || previewStatus === "generating"
   const isError = previewStatus === "error"
+  const showTitlePreview = !!titleSettings
   const videoAspectStyle = { aspectRatio: (videoRatio || "16:9").replace(":", "/") }
 
   // video 模式 refs
@@ -176,35 +293,12 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
   return (
     <div className="xx-generate-preview">
       <div className="xx-preview-header">
-        <h3>{showTitlePreview ? "标题预览" : "预览视频"}</h3>
-        {hasPreview && !showTitlePreview && <span className="xx-preview-badge">480p 预览版</span>}
+        <h3>预览视频</h3>
+        {hasPreview && <span className="xx-preview-badge">480p 预览版</span>}
       </div>
 
-      {/* Step4 标题预览模式 */}
-      {showTitlePreview && titleSettings && titleText && (
-        <div style={{ padding: "0 16px 16px" }}>
-          <TitlePreviewCanvas
-            titleText={titleText}
-            titleSettings={titleSettings}
-            videoRatio="9:16"
-            sourceVideoUrl={sourceVideoUrl}
-          />
-        </div>
-      )}
-
-      {/* Step4 但无标题时的空状态 */}
-      {showTitlePreview && (!titleText || !titleSettings) && (
-        <div className="xx-preview-empty">
-          <PlayCircleOutlined
-            style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
-          />
-          <p className="xx-preview-empty-title">请输入标题</p>
-          <p className="xx-preview-empty-desc">在左侧设置标题后，这里会实时预览效果</p>
-        </div>
-      )}
-
-      {/* 空状态：还没生成预览（非 Step4 模式） */}
-      {!showTitlePreview && previewStatus === "idle" && (
+      {/* 空状态：还没生成预览 */}
+      {previewStatus === "idle" && (
         <div className="xx-preview-empty">
           <PlayCircleOutlined
             style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
@@ -214,8 +308,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 生成中（非 Step4 模式） */}
-      {!showTitlePreview && isLoading && (
+      {/* 生成中 */}
+      {isLoading && (
         <div className="xx-preview-loading-panel">
           <div className="xx-preview-video" style={videoAspectStyle}>
             <div className="xx-preview-loading-center">
@@ -231,8 +325,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 生成失败（非 Step4 模式） */}
-      {!showTitlePreview && isError && (
+      {/* 生成失败 */}
+      {isError && (
         <div className="xx-preview-error-panel">
           <div className="xx-preview-video xx-preview-video--error" style={videoAspectStyle}>
             <p style={{ color: "rgba(255,255,255,0.8)", fontSize: 14 }}>预览生成失败</p>
@@ -275,8 +369,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 预览信息（非 Step4 模式） */}
-      {!showTitlePreview && hasPreview && previewResult && (
+      {/* 预览信息 */}
+      {hasPreview && previewResult && (
         <div className="xx-preview-info">
           <div className="xx-preview-info-row">
             <span>时长</span>

apps/web/src/pages/generate/components/Step4GeneratePreview.tsx

@@ -1,5 +1,5 @@
 /**
- * Step 5 生成预览组件（支持多预览）
+ * Step 4 生成预览组件（支持多预览）
  * 调用后端预览生成接口，展示多个真实视频预览（网格布局）
  */
 import React from "react"
@@ -12,9 +12,9 @@ import {
   ClockCircleOutlined,
 } from "@ant-design/icons"
 import { InputNumber } from "antd"
-import type { PreviewItem, PreviewStatus } from "../hooks/useStep5Preview"
+import type { PreviewItem, PreviewStatus } from "../hooks/useStep4Preview"
 
-interface Step5GeneratePreviewProps {
+interface Step4GeneratePreviewProps {
   templateName: string
   materialCount: string
   duration: number
@@ -39,7 +39,7 @@ const PREVIEW_COUNT_OPTIONS = [
   { value: 3, label: "3个" },
 ]
 
-const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
+const Step4GeneratePreview: React.FC<Step4GeneratePreviewProps> = ({
   templateName: _templateName,
   materialCount: _materialCount,
   videoRatio,
@@ -161,57 +161,54 @@ const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
                   transition: "all 0.2s",
                 }}
               >
-                {/* 轻量卡片：深色背景 + 状态指示 */}
+                {/* 缩略图/状态区域 */}
                 <div
                   style={{
                     aspectRatio,
-                    background: "#1a1a2e",
+                    background: "#000",
                     display: "flex",
-                    flexDirection: "column",
                     alignItems: "center",
                     justifyContent: "center",
                     position: "relative",
-                    gap: 8,
                   }}
                 >
-                  {/* 中心：预览编号 */}
-                  <span
-                    style={{
-                      fontSize: 24,
-                      fontWeight: 700,
-                      color: "#fff",
-                      opacity: 0.9,
-                    }}
-                  >
-                    预览 #{item.index + 1}
-                  </span>
-
-                  {/* 状态指示 */}
-                  {item.status === "generating" && (
-                    <div style={{ display: "flex", alignItems: "center", gap: 6 }}>
-                      <LoadingOutlined style={{ fontSize: 14, color: "#fff" }} spin />
-                      <span style={{ color: "rgba(255,255,255,0.8)", fontSize: 12 }}>
-                        生成中 {item.progress}%
-                      </span>
-                    </div>
+                  {item.status === "ready" && item.result && (
+                    <video
+                      src={item.result.videoUrl}
+                      controls
+                      autoPlay
+                      muted
+                      loop
+                      preload="metadata"
+                      style={{
+                        width: "100%",
+                        height: "100%",
+                        objectFit: "cover",
+                      }}
+                    />
                   )}
-                  {item.status === "pending" && (
-                    <div style={{ display: "flex", alignItems: "center", gap: 6 }}>
-                      <ClockCircleOutlined
-                        style={{ fontSize: 14, color: "rgba(255,255,255,0.6)" }}
-                      />
-                      <span style={{ color: "rgba(255,255,255,0.6)", fontSize: 12 }}>
-                        排队中...
-                      </span>
+                  {(item.status === "pending" || item.status === "generating") && (
+                    <div style={{ textAlign: "center" }}>
+                      <LoadingOutlined style={{ fontSize: 24, color: "#fff" }} spin />
+                      <p style={{ color: "rgba(255,255,255,0.8)", fontSize: 12, marginTop: 8 }}>
+                        {item.status === "pending" ? "排队中..." : `生成中 ${item.progress}%`}
+                      </p>
                     </div>
                   )}
-                  {item.status === "ready" && (
-                    <CheckCircleFilled style={{ fontSize: 18, color: "#52c41a" }} />
-                  )}
                   {item.status === "error" && (
-                    <ExclamationCircleFilled style={{ fontSize: 18, color: "#ef4444" }} />
+                    <div style={{ textAlign: "center", padding: 8 }}>
+                      <ExclamationCircleFilled style={{ fontSize: 20, color: "#ef4444" }} />
+                      <p
+                        style={{
+                          color: "rgba(255,255,255,0.7)",
+                          fontSize: 11,
+                          marginTop: 4,
+                        }}
+                      >
+                        生成失败
+                      </p>
+                    </div>
                   )}
-
                   {/* 选中角标 */}
                   {isSelected && item.status === "ready" && (
                     <div
@@ -230,6 +227,22 @@ const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
                     </div>
                   )}
                 </div>
+                {/* 底部信息 */}
+                {item.status === "ready" && item.result && (
+                  <div
+                    style={{
+                      padding: "6px 8px",
+                      background: "#fafafa",
+                      fontSize: 11,
+                      color: "#666",
+                      display: "flex",
+                      justifyContent: "space-between",
+                    }}
+                  >
+                    <span>{item.result.duration.toFixed(1)}秒</span>
+                    <span>{item.result.clipCount}段</span>
+                  </div>
+                )}
               </div>
             )
           })}
@@ -278,4 +291,4 @@ const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
   )
 }
 
-export default Step5GeneratePreview
+export default Step4GeneratePreview

apps/web/src/pages/generate/components/title/TitlePreviewCanvas.tsx

@@ -1,188 +0,0 @@
-/**
- * 标题实时预览 Canvas 组件
- *
- * 在 Step4 标题设置面板中嵌入，让用户实时看到标题文字、字体、大小、颜色、
- * 位置、描边、阴影等样式的实际渲染效果（所见即所得）。
- *
- * 使用共享的 drawTitleOnCanvas 工具函数，与 PreviewVideoPanel 行为一致。
- */
-import React, { useRef, useEffect } from "react"
-import type { TitleSettings } from "../../types"
-import { drawTitleOnCanvas } from "../../utils/drawTitleOnCanvas"
-
-interface TitlePreviewCanvasProps {
-  /** 标题文字 */
-  titleText: string
-  /** 标题样式设置 */
-  titleSettings: TitleSettings
-  /** 视频比例，默认 "9:16"（竖屏） */
-  videoRatio?: string
-  /** 素材视频 URL（作为背景显示） */
-  sourceVideoUrl?: string
-}
-
-/**
- * 解析 videoRatio 字符串为 aspect-ratio CSS 值
- */
-function parseAspect(ratio: string): string {
-  return (ratio || "9:16").replace(":", "/")
-}
-
-const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
-  titleText,
-  titleSettings,
-  videoRatio = "9:16",
-  sourceVideoUrl,
-}) => {
-  const containerRef = useRef<HTMLDivElement>(null)
-  const canvasRef = useRef<HTMLCanvasElement>(null)
-
-  // 字体加载状态
-  const fontLoadedRef = useRef(false)
-
-  /** 在 Canvas 上绘制标题 */
-  const draw = () => {
-    const canvas = canvasRef.current
-    const container = containerRef.current
-    if (!canvas || !container) return
-    const ctx = canvas.getContext("2d")
-    if (!ctx) return
-
-    const rect = container.getBoundingClientRect()
-    if (rect.width <= 0 || rect.height <= 0) return
-
-    const w = rect.width
-    const h = rect.height
-
-    // 更新 Canvas CSS 尺寸匹配容器
-    canvas.style.width = `${w}px`
-    canvas.style.height = `${h}px`
-
-    drawTitleOnCanvas(ctx, w, h, titleText, titleSettings, 24, titleSettings.position, 40)
-  }
-
-  // 字体加载：确保 measureText 使用正确字体
-  useEffect(() => {
-    let cancelled = false
-    fontLoadedRef.current = false
-
-    const fontWeight = titleSettings.bold ? "bold" : ""
-    const fontStyle = titleSettings.italic ? "italic" : ""
-    const fontSpec =
-      `${fontStyle} ${fontWeight} ${titleSettings.size}px "${titleSettings.font}"`.trim()
-
-    const onFontReady = () => {
-      if (cancelled) return
-      fontLoadedRef.current = true
-      requestAnimationFrame(() => {
-        if (!cancelled) draw()
-      })
-    }
-
-    // 用 FontFace API 加载字体，失败则降级
-    try {
-      const fontFace = new FontFace(titleSettings.font, `local("${titleSettings.font}")`)
-      fontFace
-        .load()
-        .then(() => {
-          if (!cancelled) {
-            ;(document.fonts as any).add(fontFace)
-            onFontReady()
-          }
-        })
-        .catch(() => {
-          // 字体加载失败，用默认字体继续
-          onFontReady()
-        })
-    } catch {
-      // FontFace 不可用，直接绘制
-      onFontReady()
-    }
-
-    // 同时检查 document.fonts 是否已有该字体
-    if (document.fonts.check(fontSpec)) {
-      onFontReady()
-      return
-    }
-
-    return () => {
-      cancelled = true
-    }
-  }, [titleSettings.font, titleSettings.size, titleSettings.bold, titleSettings.italic])
-
-  // props 变化时重绘
-  useEffect(() => {
-    requestAnimationFrame(draw)
-  }, [titleText, titleSettings])
-
-  // ResizeObserver 监听容器尺寸变化
-  useEffect(() => {
-    const container = containerRef.current
-    if (!container) return
-
-    const observer = new ResizeObserver(() => {
-      requestAnimationFrame(draw)
-    })
-    observer.observe(container)
-
-    return () => observer.disconnect()
-  }, [])
-
-  return (
-    <div>
-      <div
-        style={{
-          fontSize: 12,
-          color: "var(--text-tertiary, #999)",
-          marginBottom: 6,
-          fontWeight: 500,
-        }}
-      >
-        预览效果
-      </div>
-      <div
-        ref={containerRef}
-        style={{
-          width: "100%",
-          aspectRatio: parseAspect(videoRatio),
-          background: sourceVideoUrl
-            ? "#000"
-            : "linear-gradient(135deg, #1a1a2e, #16213e, #0f3460)",
-          borderRadius: 8,
-          overflow: "hidden",
-          position: "relative",
-        }}
-      >
-        {sourceVideoUrl && (
-          <video
-            src={sourceVideoUrl}
-            muted
-            loop
-            autoPlay
-            playsInline
-            style={{
-              position: "absolute",
-              top: 0,
-              left: 0,
-              width: "100%",
-              height: "100%",
-              objectFit: "cover",
-            }}
-          />
-        )}
-        <canvas
-          ref={canvasRef}
-          style={{
-            position: "absolute",
-            top: 0,
-            left: 0,
-            width: "100%",
-            height: "100%",
-          }}
-        />
-      </div>
-    </div>
-  )
-}
-
-export default TitlePreviewCanvas

apps/web/src/pages/generate/constants.ts

@@ -32,8 +32,8 @@ export const STEPS = [
   { key: 1, label: "选择模板" },
   { key: 2, label: "选择素材" },
   { key: 3, label: "选择配音" },
-  { key: 4, label: "选择标题" },
-  { key: 5, label: "生成预览" },
+  { key: 4, label: "生成预览" },
+  { key: 5, label: "选择标题" },
   { key: 6, label: "选择封面" },
   { key: 7, label: "确认生成" },
 ]

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

@@ -56,41 +56,10 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
 
     try {
       // 使用确认生成 API（基于预览任务）
-      // 解析分辨率：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
-      }
+      // 解析分辨率
+      const [widthStr, heightStr] = (props.videoRatio || "1080x1920").split("x")
+      const outputWidth = parseInt(widthStr, 10) || 1080
+      const outputHeight = parseInt(heightStr, 10) || 1920
 
       await confirmGeneration(props.previewTaskId, {
         output_width: outputWidth,

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

@@ -1,5 +1,5 @@
 /**
- * Step 5 生成预览 Hook（支持多预览 + voice_ids）
+ * Step 4 生成预览 Hook（支持多预览 + voice_ids）
  * 调用 /generation/preview 接口创建多个预览任务，轮询状态直到全部完成
  */
 import { useState, useCallback, useMemo, useEffect, useRef } from "react"
@@ -8,7 +8,6 @@ import { updateEditPlan } from "@/api/template-editor/editPlans"
 import type { PreviewTaskResponse, PreviewStatus as ApiPreviewStatus } from "@/api/generation"
 import type { EditingTemplate } from "@/api/editing-planner"
 import { safeExtractError } from "./generate-video/errorUtils"
-import type { TitleSettings } from "../types"
 
 /** 安全地将值转为字符串，防止对象被直接渲染导致 React Error #31 */
 const safeString = (val: unknown, fallback: string): string => {
@@ -27,7 +26,7 @@ const safeNumber = (val: unknown, fallback = 0): number => {
   return fallback
 }
 
-interface UseStep5PreviewProps {
+interface UseStep4PreviewProps {
   templates: EditingTemplate[]
   selectedTemplate: string
   materialMode: "manual" | "auto"
@@ -41,8 +40,6 @@ interface UseStep5PreviewProps {
   voiceLibraryId?: string
   /** 要生成的预览数量 */
   previewCount?: number
-  /** 标题设置（传递给后端，让预览视频包含标题） */
-  titleSettings?: TitleSettings
 }
 
 export type PreviewStatus = "idle" | "pending" | "generating" | "ready" | "error"
@@ -81,7 +78,7 @@ const createInitialItem = (index: number): PreviewItem => ({
   progress: 0,
 })
 
-export function useStep5Preview({
+export function useStep4Preview({
   templates,
   selectedTemplate,
   materialMode,
@@ -92,8 +89,7 @@ export function useStep5Preview({
   voiceIds,
   voiceLibraryId,
   previewCount = 1,
-  titleSettings,
-}: UseStep5PreviewProps) {
+}: UseStep4PreviewProps) {
   const templateName = useMemo(
     () => templates.find((t) => t.id === selectedTemplate)?.name ?? "未选择",
     [templates, selectedTemplate],
@@ -157,7 +153,6 @@ export function useStep5Preview({
     duration,
     videoRatio,
     voiceIds: [...(voiceIds || [])].sort().join(","),
-    titleSettings: titleSettings?.title || "",
   })
 
   useEffect(() => {
@@ -169,7 +164,6 @@ export function useStep5Preview({
       duration,
       videoRatio,
       [...(voiceIds || [])].sort().join(","),
-      titleSettings?.title || "",
     ].join("|")
 
     const prevKey = [
@@ -180,7 +174,6 @@ export function useStep5Preview({
       prevDepsRef.current.duration,
       prevDepsRef.current.videoRatio,
       prevDepsRef.current.voiceIds,
-      prevDepsRef.current.titleSettings,
     ].join("|")
 
     if (prevKey !== currentKey && items.some((it) => it.status !== "idle")) {
@@ -198,7 +191,6 @@ export function useStep5Preview({
       duration,
       videoRatio,
       voiceIds: [...(voiceIds || [])].sort().join(","),
-      titleSettings: titleSettings?.title || "",
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [
@@ -210,7 +202,6 @@ export function useStep5Preview({
     videoRatio,
     voiceIds,
     previewCount,
-    titleSettings,
   ])
 
   // 组件卸载时清理所有轮询
@@ -358,19 +349,6 @@ export function useStep5Preview({
           video_ratio: videoRatio,
           voice_ids: voiceIds && voiceIds.length > 0 ? voiceIds : undefined,
           voice_library_id: voiceLibraryId || undefined,
-          // 标题烧录配置
-          title_config: titleSettings?.title
-            ? {
-                text: titleSettings.title,
-                font: titleSettings.font,
-                font_size: titleSettings.size,
-                font_color: titleSettings.color,
-                position: titleSettings.position,
-                bold: titleSettings.bold,
-                stroke: titleSettings.stroke,
-                shadow: titleSettings.shadow,
-              }
-            : undefined,
         })
 
         if (startTimeRef.current === 0) return
@@ -396,7 +374,6 @@ export function useStep5Preview({
     voiceIds,
     voiceLibraryId,
     previewCount,
-    titleSettings,
     clearPollTimer,
     pollPreviewStatus,
   ])
@@ -473,4 +450,4 @@ export function useStep5Preview({
   }
 }
 
-export default useStep5Preview
+export default useStep4Preview

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

@@ -1,7 +1,7 @@
 /**
  * GeneratePage 步骤导航
  * 管理步骤切换与各步骤的前置校验
- * 步骤顺序：模板(1) → 素材(2) → 配音(3) → 标题(4) → 预览(5) → 封面(6) → 确认(7)
+ * 步骤顺序：模板(1) → 素材(2) → 配音(3) → 预览(4) → 标题(5) → 封面(6) → 确认(7)
  */
 import { message } from "antd"
 import type { TitleSettings } from "../types"
@@ -49,12 +49,12 @@ export const useStepNavigation = (options: UseStepNavigationOptions): UseStepNav
       return
     }
     // Step3 配音：配音为可选项，不强制校验，用户可跳过
-    if (currentStep === 4 && !titleSettings.title.trim()) {
-      message.warning("请选择或输入标题")
+    if (currentStep === 4 && !previewReady) {
+      message.warning("请先生成剪辑预览")
       return
     }
-    if (currentStep === 5 && !previewReady) {
-      message.warning("请先生成剪辑预览")
+    if (currentStep === 5 && !titleSettings.title.trim()) {
+      message.warning("请选择或输入标题")
       return
     }
     if (currentStep < 7) {

apps/web/src/pages/generate/utils/drawTitleOnCanvas.ts

@@ -1,126 +0,0 @@
-/**
- * Canvas 标题绘制工具函数（共享模块）
- *
- * 供 PreviewVideoPanel（预览视频标题叠加）和 TitlePreviewCanvas（标题设置实时预览）共用。
- * 绘制行为与 ASS 字幕引擎一致：逐字换行、居中、描边/阴影。
- */
-import type { TitleSettings } from "../types"
-
-/**
- * 将文本按 maxWidth 逐字换行，返回行数组。
- * 与 ASS 字幕引擎的逐字换行行为一致。
- */
-export function wrapText(ctx: CanvasRenderingContext2D, text: string, maxWidth: number): string[] {
-  const lines: string[] = []
-  let currentLine = ""
-  for (const char of text) {
-    const testLine = currentLine + char
-    if (ctx.measureText(testLine).width > maxWidth && currentLine) {
-      lines.push(currentLine)
-      currentLine = char
-    } else {
-      currentLine = testLine
-    }
-  }
-  if (currentLine) lines.push(currentLine)
-  return lines
-}
-
-/**
- * 在 canvas 上绘制标题文字（含描边/阴影/多行居中）
- *
- * @param ctx canvas 上下文
- * @param w canvas CSS 宽度
- * @param h canvas CSS 高度
- * @param text 标题文字
- * @param settings 标题样式
- * @param paddingX 左右边距（px），与 ASS 的 MarginL/MarginR 对应
- * @param position "top" | "center" | "bottom"
- * @param topOffset 顶部/底部偏移量
- */
-export function drawTitleOnCanvas(
-  ctx: CanvasRenderingContext2D,
-  w: number,
-  h: number,
-  text: string,
-  settings: TitleSettings,
-  paddingX: number,
-  position: string,
-  topOffset: number,
-) {
-  const dpr = window.devicePixelRatio || 1
-
-  // 设置 canvas 物理像素尺寸（高清屏适配）
-  ctx.canvas.width = Math.round(w * dpr)
-  ctx.canvas.height = Math.round(h * dpr)
-  ctx.scale(dpr, dpr)
-
-  // 清除
-  ctx.clearRect(0, 0, w, h)
-
-  // 可用宽度 = 总宽 - 左右边距
-  const availableWidth = w - paddingX * 2
-  if (availableWidth <= 0) return
-
-  // 字体设置
-  const fontSize = Math.round(Math.min(settings.size, 36))
-  const fontWeight = settings.bold ? "bold" : "normal"
-  const fontStyle = settings.italic ? "italic" : "normal"
-  ctx.font = `${fontStyle} ${fontWeight} ${fontSize}px "${settings.font}"`
-
-  // 文字属性
-  ctx.textAlign = "center"
-  ctx.textBaseline = "middle"
-  ctx.fillStyle = settings.color
-
-  const lineHeight = fontSize * 1.4
-
-  // 描边 & 阴影
-  if (settings.stroke) {
-    ctx.strokeStyle = "rgba(0,0,0,0.6)"
-    ctx.lineWidth = 2
-    ctx.lineJoin = "round"
-  }
-  if (settings.shadow) {
-    ctx.shadowColor = "rgba(0,0,0,0.7)"
-    ctx.shadowBlur = 4
-    ctx.shadowOffsetX = 2
-    ctx.shadowOffsetY = 2
-  }
-
-  // 换行
-  const displayText = text && text.trim() ? text : "请选择或输入标题"
-  const lines = wrapText(ctx, displayText, availableWidth)
-
-  // 起始 Y：根据 position 计算
-  const totalTextHeight = lines.length * lineHeight
-  let startY: number
-  switch (position) {
-    case "top":
-      startY = topOffset
-      break
-    case "center":
-      startY = (h - totalTextHeight) / 2 + lineHeight / 2
-      break
-    case "bottom":
-    default:
-      startY = h - topOffset - totalTextHeight + lineHeight / 2
-      break
-  }
-
-  // 居中 x = w/2
-  const x = w / 2
-  lines.forEach((line, i) => {
-    const y = startY + i * lineHeight
-    if (settings.stroke) ctx.strokeText(line, x, y)
-    ctx.fillText(line, x, y)
-  })
-
-  // 重置 shadow（避免影响后续绘制）
-  if (settings.shadow) {
-    ctx.shadowColor = "transparent"
-    ctx.shadowBlur = 0
-    ctx.shadowOffsetX = 0
-    ctx.shadowOffsetY = 0
-  }
-}

apps/web/src/test/pages/generate/smoke.test.tsx

@@ -17,7 +17,7 @@ import "@/api/generation/types"
 // 直接引入所有 Step 组件，建立完整依赖链
 import "@/pages/generate/GeneratePage"
 import "@/pages/generate/components/Step2MaterialSelect"
-import "@/pages/generate/components/Step5GeneratePreview"
+import "@/pages/generate/components/Step4GeneratePreview"
 import "@/pages/generate/components/Step4TitleSettings"
 import "@/pages/generate/components/Step5VoiceSelect"
 import "@/pages/generate/components/PreviewVideoPanel"
@@ -47,7 +47,7 @@ describe("GeneratePage module smoke test", () => {
   })
 })
 import "@/pages/generate/hooks/useGenerateVideo"
-import "@/pages/generate/hooks/useStep5Preview"
+import "@/pages/generate/hooks/useStep4Preview"
 import "@/pages/generate/hooks/generate-video/useGenerationPolling"
 import "@/pages/generate/hooks/useGenerateFormState"
 import "@/pages/generate/hooks/useGenerateFormState/useTemplateSelection"

apps/web/src/test/pages/generate/step4-preview.smoke.test.tsx

@@ -0,0 +1,17 @@
+/**
+ * Step4GeneratePreview smoke test
+ * 确保 vitest related 模式能匹配到第4步预览生成相关文件的改动
+ */
+import { describe, it, expect } from "vitest"
+
+import "@/pages/generate/components/Step4GeneratePreview"
+import "@/pages/generate/hooks/useStep4Preview"
+import "@/pages/generate/hooks/useStepNavigation"
+import "@/pages/generate/components/GenerateStepContent"
+import "@/pages/generate/GeneratePage"
+
+describe("Step4GeneratePreview module smoke test", () => {
+  it("should load all step4 preview modules", () => {
+    expect(true).toBe(true)
+  })
+})

apps/web/src/test/pages/generate/step5-preview.smoke.test.tsx

@@ -1,17 +0,0 @@
-/**
- * Step5GeneratePreview smoke test
- * 确保 vitest related 模式能匹配到第5步预览生成相关文件的改动
- */
-import { describe, it, expect } from "vitest"
-
-import "@/pages/generate/components/Step5GeneratePreview"
-import "@/pages/generate/hooks/useStep5Preview"
-import "@/pages/generate/hooks/useStepNavigation"
-import "@/pages/generate/components/GenerateStepContent"
-import "@/pages/generate/GeneratePage"
-
-describe("Step5GeneratePreview module smoke test", () => {
-  it("should load all step5 preview modules", () => {
-    expect(true).toBe(true)
-  })
-})

apps/worker/video_processing/dedup_helpers.py

@@ -85,9 +85,19 @@ 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 set for video %s: %s", video_id, thumbnail_url[:80] if thumbnail_url else "")
+            logger.info("Thumbnail reused (pre-generated) for video %s", video_id)
         else:
-            logger.debug("No thumbnail_url provided for video %s, skipping", video_id)
+            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)
 
         # 计算视频指纹
         deduplicator = VideoDeduplicator()

apps/worker/video_processing/render_adapter.py

@@ -47,14 +47,7 @@ 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())
-        # 最小 100px 防护：避免前端传入宽高比（如 "9:16"）被 parseInt 截断为极小值
-        if width < 100 or height < 100:
-            logger.warning(
-                "分辨率异常小 (%dx%d)，使用默认值。原始值: %s",
-                width,
-                height,
-                resolution_str,
-            )
+        if width <= 0 or height <= 0:
             return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT
         return width, height
     except (ValueError, TypeError):
@@ -81,7 +74,9 @@ class RenderAdapterResult:
     failed_clip_ids: list[str] = None  # 失败的 clip id 列表
     error_message: str = ""
     error_detail: str = ""  # 详细错误信息（如 ffmpeg stderr），用于排查
-    cover_url: str = ""  # 封面图片 URL（从渲染后视频抽帧，天然带标题）
+    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:
@@ -560,33 +555,37 @@ 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. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题）
-        cover_url = ""
-        cover_frame_path = None
+        # 6. 生成封面缩略图
+        thumbnail_url = ""
         try:
-            from video_processing.thumbnail_generator import extract_first_frame
+            from video_processing.thumbnail_generator import generate_and_upload_thumbnail
 
-            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:
+            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:
                 logger.info(
-                    "[render-adapter] 封面帧提取成功: plan_id=%s url=%s",
+                    "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
                     plan_id,
-                    cover_url[:80],
+                    len(cover_candidates),
                 )
         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,
             )
@@ -614,7 +613,7 @@ class RenderAdapter:
             success=True,
             output_url=output_url or "",
             output_path=result.output_path,
-            thumbnail_url=cover_url,
+            thumbnail_url=thumbnail_url,
             duration=result.duration,
             file_size=result.file_size,
             width=result.width,
@@ -622,7 +621,7 @@ class RenderAdapter:
             clip_count=len(clips),
             rendered_clip_ids=final_rendered_ids,
             failed_clip_ids=final_failed_ids,
-            cover_url=cover_url,
+            cover_candidates=cover_candidates,
         )
 
     def render_from_memory(

apps/worker/video_processing/thumbnail_generator.py

@@ -1,8 +1,4 @@
-"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。
-
-统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
-渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
-"""
+"""视频缩略图生成工具 — 抽取首帧上传到 OSS。"""
 
 from __future__ import annotations
 
@@ -17,30 +13,28 @@ def extract_first_frame(
     video_path: str,
     output_path: str | None = None,
     *,
-    width: int = -1,
+    width: int = 640,
     height: int = -1,
     timeout: int = 30,
     seek_ratio: float = 0.15,
     min_seek_seconds: float = 1.0,
 ) -> str:
-    """抽取视频封面帧（默认取视频时长 15% 处的帧，避开片头纯色画面）。
-
-    因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。
+    """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。
 
     Args:
         video_path: 视频文件路径
         output_path: 输出图片路径，不传则用临时文件
-        width: 输出宽度（默认 -1，保持原始分辨率）
-        height: 输出高度（默认 -1，保持原始分辨率）
+        width: 输出宽度（默认 640，-1 表示按比例缩放）
+        height: 输出高度（默认 -1，按比例缩放）
         timeout: 超时时间（秒）
         seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
         min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0
 
     Returns:
-        生成的封面帧文件路径
+        生成的缩略图文件路径
 
     Raises:
-        RuntimeError: ffmpeg 执行失败或输出文件为空
+        subprocess.CalledProcessError: ffmpeg 执行失败
     """
     from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 
@@ -63,20 +57,10 @@ def extract_first_frame(
         # 格式化为 HH:MM:SS.xx
         seek_str = _format_seek_time(seek_time)
 
-        # 构建 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，更快）
+        # -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",
@@ -115,7 +99,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"Cover frame extraction failed: {output_path}")
+            raise RuntimeError(f"Thumbnail generation failed: {output_path}")
 
         return output_path
     except Exception:
@@ -134,3 +118,171 @@ 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/video_processing/unified_render_service.py

@@ -52,7 +52,6 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr
 from video_processing.tts_engine import TtsEngine
 from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
 
-from packages.domain.ass_subtitle_builder import build_ass_content
 from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX
 from packages.domain.render_layer_utils import clip_adjusted_duration as _clip_adjusted_duration_pure
 from packages.domain.render_layer_utils import clip_effective_duration as _clip_effective_duration_pure
@@ -132,83 +131,6 @@ _PIP_SCALE = 0.25  # PiP 占主画面的比例
 # ── 统一渲染引擎 ─────────────────────────────────────────────────────────────
 
 
-def _overlay_title_on_ass(
-    ass_path: Path,
-    *,
-    title_text: str,
-    title_config: dict,
-    video_width: int,
-    video_height: int,
-    video_duration: float,
-) -> None:
-    """在已有的 ASS 文件上叠加标题事件。
-
-    用于 ASR 字幕路径：ASR 生成的 ASS 只含字幕事件，此函数将标题
-    作为独立的 TitleStyle + Dialogue 事件追加进去，使标题显示在
-    ASR 字幕之上（封面抽帧时也能看到标题）。
-
-    Args:
-        ass_path: 已有的 ASS 文件路径（由 generate_ass_from_timeline 生成）
-        title_text: 标题文本
-        title_config: 标题样式配置
-        video_width: 视频宽度
-        video_height: 视频高度
-        video_duration: 视频时长
-    """
-    if not title_text or not title_text.strip():
-        return
-
-    # 生成仅包含标题的 ASS 内容
-    title_only_content = build_ass_content(
-        video_width=video_width,
-        video_height=video_height,
-        video_duration=video_duration,
-        title_text=title_text,
-        title_config=title_config,
-    )
-    if not title_only_content:
-        return
-
-    # 从 title_only_content 中提取 TitleStyle 行和标题 Dialogue 行
-    title_style_line = None
-    title_dialogue_line = None
-    for line in title_only_content.splitlines():
-        if line.startswith("Style: TitleStyle"):
-            title_style_line = line
-        elif "TitleStyle" in line and line.startswith("Dialogue:"):
-            title_dialogue_line = line
-
-    if not title_style_line or not title_dialogue_line:
-        logger.warning("标题 ASS 内容解析失败，跳过叠加")
-        return
-
-    # 读取现有 ASS 文件
-    existing_content = ass_path.read_text(encoding="utf-8")
-
-    # 插入 TitleStyle 到 [V4+ Styles] 段（最后一个 Style: 行之后）
-    # 插入标题 Dialogue 到 [Events] 段（Format 行之后）
-    lines = existing_content.splitlines()
-    last_style_idx = -1
-    events_format_idx = -1
-    for i, line in enumerate(lines):
-        if line.startswith("Style:"):
-            last_style_idx = i
-        if line.startswith("Format: Layer,"):
-            events_format_idx = i
-
-    if last_style_idx >= 0:
-        lines.insert(last_style_idx + 1, title_style_line)
-        # events_format_idx 需要 +1 因为插入了一行
-        events_format_idx += 1
-
-    # 2. 在 Events Format 行之后、第一个 Dialogue 之前插入标题 Dialogue
-    # 标题应该显示在整个视频时长，放在最前面（最先渲染，在底层）
-    if events_format_idx >= 0:
-        lines.insert(events_format_idx + 1, title_dialogue_line)
-
-    ass_path.write_text("\n".join(lines), encoding="utf-8")
-
-
 class UnifiedRenderService:
     """统一渲染引擎。
 
@@ -558,11 +480,7 @@ class UnifiedRenderService:
         """
         config = self.plan.config or {}
         title_cfg = config.get("title", {}) or {}
-        if not isinstance(title_cfg, dict):
-            title_cfg = {}
         subtitle_cfg = config.get("subtitle", {}) or {}
-        if not isinstance(subtitle_cfg, dict):
-            subtitle_cfg = {}
 
         title_enabled = title_cfg.get("enabled", True)
         subtitle_enabled = subtitle_cfg.get("enabled", True)
@@ -597,71 +515,14 @@ class UnifiedRenderService:
                         timeline.segment_count,
                         video_duration,
                     )
-                    # ASR 路径也需要叠加标题（标题作为独立 ASS Event 追加到 ASR 字幕之上）
-                    # 用独立 try-except 包裹，避免叠加失败时覆盖已生成的 ASR 数据
-                    if has_title:
-                        try:
-                            _overlay_title_on_ass(
-                                ass_path,
-                                title_text=title_text,
-                                title_config=title_cfg,
-                                video_width=self.output_width,
-                                video_height=self.output_height,
-                                video_duration=video_duration,
-                            )
-                            logger.info(
-                                "ASR字幕叠加标题: plan_id=%s title=%s",
-                                self.plan.id,
-                                title_text[:30],
-                            )
-                        except Exception:
-                            logger.warning(
-                                "ASR字幕叠加标题失败，保留纯ASR字幕: plan_id=%s",
-                                self.plan.id,
-                                exc_info=True,
-                            )
                     return ass_path
                 else:
-                    # ASR 无结果：如果有标题，仍然生成标题 ASS
-                    if has_title:
-                        generate_ass_subtitles(
-                            ass_path,
-                            video_width=self.output_width,
-                            video_height=self.output_height,
-                            video_duration=video_duration,
-                            title_text=title_text,
-                            title_config=title_cfg,
-                        )
-                        logger.info(
-                            "ASR无结果但生成标题: plan_id=%s title=%s",
-                            self.plan.id,
-                            title_text[:30],
-                        )
-                        return ass_path
+                    # ASR 无结果，不生成字幕
                     logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id)
                     return None
             except Exception:
-                # ASR 失败降级：如果有标题，仍然生成标题 ASS
-                if has_title:
-                    try:
-                        generate_ass_subtitles(
-                            ass_path,
-                            video_width=self.output_width,
-                            video_height=self.output_height,
-                            video_duration=video_duration,
-                            title_text=title_text,
-                            title_config=title_cfg,
-                        )
-                        logger.info(
-                            "ASR失败但生成标题: plan_id=%s title=%s",
-                            self.plan.id,
-                            title_text[:30],
-                        )
-                        return ass_path
-                    except Exception:
-                        logger.warning("ASR失败后标题生成也失败", exc_info=True)
-                else:
-                    logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
+                # ASR 失败降级：不生成字幕，不阻断主流程
+                logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
                 return None
 
         # 静态字幕模式（原有逻辑）
@@ -793,8 +654,6 @@ class UnifiedRenderService:
         config = self.plan.config or {}
         tts_cfg = config.get("tts", {}) or {}
         subtitle_cfg = config.get("subtitle", {}) or {}
-        if not isinstance(subtitle_cfg, dict):
-            subtitle_cfg = {}
         use_subtitle_align = False  # 是否使用字幕对齐模式
 
         # 兼容前端顶层字段：voice_id / custom_text / voice_clone_profile_id

apps/worker/worker_app/tasks/generation.py

@@ -13,7 +13,6 @@
 
 from __future__ import annotations
 
-import json
 import logging
 import os
 import tempfile
@@ -984,9 +983,9 @@ def _load_template_plan_config(template_id: str) -> dict:
 
             # 从独立字段组装成 plan.config 格式
             plan_config: dict[str, Any] = {}
-            title_cfg = template.title_config if isinstance(template.title_config, dict) else {}
-            subtitle_cfg = template.subtitle_config if isinstance(template.subtitle_config, dict) else {}
-            bgm_cfg = template.bgm_config if isinstance(template.bgm_config, dict) else {}
+            title_cfg = template.title_config or {}
+            subtitle_cfg = template.subtitle_config or {}
+            bgm_cfg = template.bgm_config or {}
 
             if title_cfg:
                 plan_config["title"] = title_cfg
@@ -1124,15 +1123,14 @@ def _render_video(
     resolution: str = "",
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
-    custom_title: str = "",
-) -> tuple[Path, float, str]:
+) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration, cover_url)
+        (output_path, render_duration)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1171,61 +1169,6 @@ def _render_video(
             merged_bgm.get("source", ""),
         )
 
-    # 用户自定义标题覆盖模板标题（用户指定优先级最高）
-    # 支持两种格式：
-    # 1. JSON 格式（新）：{"text": "xxx", "font_size": 32, ...} — 包含标题文本和样式
-    # 2. 纯文本格式（旧）：直接作为标题文本使用
-    if custom_title and custom_title.strip():
-        plan_cfg = dict(virtual_plan.config or {})
-        _raw_title = plan_cfg.get("title", {}) or {}
-        title_cfg = dict(_raw_title) if isinstance(_raw_title, dict) else {}
-        ct_stripped = custom_title.strip()
-        parsed_config = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed_config = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed_config = None
-        if parsed_config and isinstance(parsed_config, dict):
-            # JSON 格式：合并完整标题配置（文本 + 样式）
-            title_text = (parsed_config.get("text") or "").strip()
-            if title_text:
-                title_cfg["text"] = title_text
-                title_cfg["enabled"] = True
-                # 合并样式字段（用户指定 > 模板默认）
-                style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
-                for key in style_keys:
-                    if key in parsed_config and parsed_config[key] is not None:
-                        # 前端字段名映射到 ASS 字段名
-                        mapped_key = {
-                            "font_size": "size",
-                            "font_color": "color",
-                            "font_preset": "font",
-                        }.get(key, key)
-                        title_cfg[mapped_key] = parsed_config[key]
-                logger.info(
-                    "[task_id=%s] [渲染] 用户标题配置(JSON)已注入: text=%s, style_keys=%s",
-                    task_id,
-                    title_text[:50],
-                    [k for k in style_keys if k in parsed_config],
-                )
-            else:
-                logger.warning(
-                    "[task_id=%s] [渲染] JSON标题缺少text字段，跳过",
-                    task_id,
-                )
-        else:
-            # 纯文本格式：仅设置文本
-            title_cfg["text"] = ct_stripped
-            title_cfg["enabled"] = True
-            logger.info(
-                "[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
-                task_id,
-                ct_stripped[:50],
-            )
-        plan_cfg["title"] = title_cfg
-        virtual_plan.config = plan_cfg
-
     # 确保输出分辨率配置存在
     # 优先级：用户指定 > 模板配置 > 默认 1280x720
     # 预览模式：强制 854x480 + 低码率
@@ -1246,10 +1189,7 @@ def _render_video(
         plan_cfg = dict(virtual_plan.config or {})
         plan_cfg["voice_id"] = voice_ids[0]
         subtitle_cfg = plan_cfg.get("subtitle", {}) or {}
-        if not isinstance(subtitle_cfg, dict):
-            subtitle_cfg = {}
         subtitle_cfg["auto_generated"] = True
-        subtitle_cfg["enabled"] = True  # 确保 ASR 字幕路径被触发，标题叠加也依赖此路径
         plan_cfg["subtitle"] = subtitle_cfg
         virtual_plan.config = plan_cfg
         logger.info(
@@ -1304,9 +1244,8 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
-    cover_url = getattr(render_result, "cover_url", "") or ""
 
-    return output_path, render_duration, cover_url
+    return output_path, render_duration
 
 
 def _upload_and_record(
@@ -1317,7 +1256,6 @@ def _upload_and_record(
     editing_mode,
     user_id: str = "",
     video_name: str = "",
-    thumbnail_url: str = "",
 ) -> tuple[str, float, int, int]:
     """上传 OSS、创建视频记录并查重。
 
@@ -1563,22 +1501,12 @@ 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, cover_url = _render_video(
+            output_path, render_duration = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1591,42 +1519,12 @@ 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 + 查重记录 ───────────────────────────────────────
@@ -1639,7 +1537,6 @@ 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,21 +206,11 @@ def ingest_asset(job_id: str) -> dict:
             # 视频类型：生成缩略图（文件还在的时候生成）
             thumbnail_url = None
             if media_type == "video" and extract_success:
-                frame_path = None
                 try:
-                    from video_processing.oss_helpers import upload_to_oss
-                    from video_processing.thumbnail_generator import extract_first_frame
+                    from video_processing.thumbnail_generator import generate_and_upload_thumbnail
 
-                    frame_path = extract_first_frame(str(local_file), width=640)
                     thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg"
-                    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
+                    thumbnail_url = generate_and_upload_thumbnail(str(local_file), thumb_storage_key)
                     if thumbnail_url:
                         logger.info(
                             "素材缩略图生成成功: job_id=%s url=%s",

packages/application/generation_tasks.py

@@ -32,7 +32,6 @@ class CreateGenerationTaskCommand:
     output_height: int = 720
     cover_url: str = ""
     custom_title: str = ""
-    title_config: dict = field(default_factory=dict)
 
 
 class CreateGenerationTaskUseCase:

packages/shared/ai_service.py

@@ -13,6 +13,8 @@ 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
 
@@ -352,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],
@@ -361,9 +450,9 @@ def _call_ai_cover_service(
 ) -> Dict[str, Any]:
     """调用 AI 封面生成服务.
 
-    统一封面管道下，封面已由渲染后视频抽帧生成并持久化到 GenerationTask.cover_url。
-    此函数仅处理 manual/upload 等需要前端交互的类型，
-    ai_frame/ai_regenerate 类型应由调用方直接从持久化的封面 URL 读取。
+    优先级：
+    1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理
+    2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）
 
     失败时抛出 RuntimeError。
 
@@ -395,14 +484,88 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # 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})。请重新生成预览视频以触发封面自动提取。")
+    # 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 是否可访问。")
+
+
+# ── 公共入口 ────────────────────────────────────────────────────────────────
 
 
 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, _cover_url = _render_video(
+            output_path, render_duration = _render_video(
                 task_id="test_task_123",
                 downloaded_videos=[Path("/tmp/video1.mp4")],
                 voice_path=None,

tests/unit/test_bool_config_defense.py

@@ -1,130 +0,0 @@
-"""
-测试：模板 config 字段存储了非 dict 值（如 True / False / str）时，
-渲染链路不会崩溃（'bool' object has no attribute 'get'）。
-
-覆盖两个关键文件：
-1. generation.py — _load_template_plan_config 旧系统路径
-2. unified_render_service.py — _maybe_generate_ass
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# Add worker app to path
-sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-
-class TestLoadTemplatePlanConfigBoolDefense:
-    """_load_template_plan_config 旧系统路径对非 dict 值的防护。"""
-
-    def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg):
-        """通过 mock 新模板系统返回 None，强制走旧模板系统 fallback 路径。"""
-        from worker_app.tasks.generation import _load_template_plan_config
-
-        mock_old_template = MagicMock()
-        mock_old_template.title_config = title_cfg
-        mock_old_template.subtitle_config = subtitle_cfg
-        mock_old_template.bgm_config = bgm_cfg
-
-        mock_session = MagicMock()
-        # 旧系统 query 返回 mock template
-        mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template
-
-        # Mock 新模板系统 repo.get() 返回 None（强制走 fallback）
-        mock_repo_cls = MagicMock()
-        mock_repo_cls.return_value.get.return_value = None
-
-        with (
-            patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session),
-            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls),
-            patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()),
-        ):
-            return _load_template_plan_config("fake-id")
-
-    def test_bool_values_return_empty(self):
-        """title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉"""
-        result = self._call_old_path(True, False, "not_a_dict")
-        assert isinstance(result, dict)
-        assert "title" not in result
-        assert "subtitle" not in result
-        assert "bgm" not in result
-
-    def test_valid_dict_passes_through(self):
-        """正常 dict 正常传递"""
-        result = self._call_old_path(
-            {"text": "标题", "enabled": True},
-            {"text": "副标题"},
-            {"enabled": True, "source": "test.mp3"},
-        )
-        assert result["title"] == {"text": "标题", "enabled": True}
-        assert result["subtitle"] == {"text": "副标题"}
-        assert result["bgm"] == {"enabled": True, "source": "test.mp3"}
-
-    def test_none_returns_empty(self):
-        """None → 空 dict"""
-        result = self._call_old_path(None, None, None)
-        assert result == {}
-
-    def test_mixed_valid_and_invalid(self):
-        """部分有效、部分无效时只保留有效的"""
-        result = self._call_old_path({"text": "OK"}, True, None)
-        assert "title" in result
-        assert "subtitle" not in result
-        assert "bgm" not in result
-
-    def test_int_and_list_also_filtered(self):
-        """int / list 类型也被过滤"""
-        result = self._call_old_path(42, [1, 2, 3], 0)
-        assert result == {}
-
-
-class TestUnifiedRenderBoolConfigDefense:
-    """_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。"""
-
-    def _make_service(self, config):
-        from video_processing.unified_render_service import UnifiedRenderService
-
-        service = UnifiedRenderService.__new__(UnifiedRenderService)
-        mock_plan = MagicMock()
-        mock_plan.config = config
-        service.plan = mock_plan
-        service.task_id = "test-task"
-        return service
-
-    def test_bool_title_does_not_crash(self):
-        """config['title']=True → 不崩溃，返回 None"""
-        service = self._make_service({"title": True, "subtitle": {}})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_bool_subtitle_does_not_crash(self):
-        """config['subtitle']=False → 不崩溃，返回 None"""
-        service = self._make_service({"title": {}, "subtitle": False})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_str_title_does_not_crash(self):
-        """config['title']='plain string' → 不崩溃"""
-        service = self._make_service({"title": "plain string", "subtitle": {}})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_none_config_does_not_crash(self):
-        """config=None → 不崩溃"""
-        service = self._make_service(None)
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None
-
-    def test_int_title_does_not_crash(self):
-        """config['title']=42 → 不崩溃"""
-        service = self._make_service({"title": 42, "subtitle": 0})
-        result = service._maybe_generate_ass(video_duration=10.0)
-        assert result is None

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,149 +1,486 @@
-"""Tests for unified cover frame extraction pipeline.
+"""Tests for cover frame pre-extraction during rendering.
 
-统一封面管道测试：
-- extract_first_frame: 从已渲染视频抽取封面帧
-- 封面天然带标题（ASS 字幕已烧录到视频中）
+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
 
-import tempfile
-import unittest
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import MagicMock, Mock, call, patch
+
+import pytest
 
 
-class TestExtractFirstFrame(unittest.TestCase):
-    """extract_first_frame 单元测试."""
+class TestExtractCoverCandidates:
+    """extract_cover_candidates 测试."""
 
     @patch("video_processing.ffmpeg_utils.run_ffmpeg")
-    @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
+    @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
 
-        # Mock run_ffmpeg 创建输出文件（ffmpeg 真实行为）
-        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
+        from video_processing.thumbnail_generator import extract_cover_candidates
 
-        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
-            Path(video.name).write_bytes(b"fake video")
-            result = extract_first_frame(video.name)
+        # 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 ("", "")
 
-            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)
+        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=10.0)
-    def test_custom_seek_ratio(self, mock_probe, mock_run):
-        """自定义抽帧位置."""
-        from video_processing.thumbnail_generator import extract_first_frame
+    @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
 
-        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
+        from video_processing.thumbnail_generator import extract_cover_candidates
 
-        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)
+        call_count = 0
 
-            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)
+        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 ("", "")
 
-    @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
+        mock_run.side_effect = fake_run
 
-        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
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
+            tmp.write(b"fake video")
+            video_path = tmp.name
 
-            # Create the file so ffmpeg "succeeds"
-            mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image")
+        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)
 
-            result = extract_first_frame(video.name, output_path=out.name)
-            self.assertEqual(result, out.name)
-            Path(out.name).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
 
-    @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
+        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")
+        # 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 tempfile.NamedTemporaryFile(suffix=".mp4") as video:
-            Path(video.name).write_bytes(b"fake video")
-            result = extract_first_frame(video.name)
+        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
 
-            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)
+            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 TestRenderAdapterCoverUrl(unittest.TestCase):
-    """RenderAdapterResult.cover_url 字段测试."""
+class TestExtractAndUploadCoverFrames:
+    """extract_and_upload_cover_frames 测试."""
 
-    def test_result_has_cover_url_field(self):
-        """RenderAdapterResult 包含 cover_url 字段."""
-        from video_processing.render_adapter import RenderAdapterResult
+    @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
 
-        result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg")
-        self.assertEqual(result.cover_url, "https://example.com/cover.jpg")
+        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
-    def test_result_cover_url_defaults_empty(self):
-        """cover_url 默认为空字符串."""
+        # 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)
-        self.assertEqual(result.cover_url, "")
+        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
 
 
-if __name__ == "__main__":
-    unittest.main()
+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_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 (统一封面管道: 仅保留 extract_first_frame)
+        # mock video_processing.thumbnail_generator
         mock_thumb = MagicMock()
-        mock_thumb.extract_first_frame = MagicMock()
+        mock_thumb.generate_and_upload_thumbnail = 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 时直接复用，统一封面管道不再自动生成缩略图。"""
+        """传入 thumbnail_url 时直接复用，不调用 generate_and_upload_thumbnail。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -339,23 +339,26 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                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,
-                )
+                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,
+                    )
 
             assert result == 1
+            # 预生成缩略图时不应调用 generate_and_upload_thumbnail
+            mock_gen.assert_not_called()
 
             from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
 
@@ -365,8 +368,8 @@ class TestThumbnailInDedupHelpers:
         finally:
             session.close()
 
-    def test_no_thumbnail_when_not_provided(self):
-        """未传 thumbnail_url 时不生成缩略图（统一封面管道已移除自动缩略图生成）。"""
+    def test_thumbnail_generated_when_not_provided(self):
+        """未传 thumbnail_url 时调用 generate_and_upload_thumbnail 生成。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -374,6 +377,8 @@ 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
@@ -381,34 +386,43 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                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,
-                )
+                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,
+                    )
 
             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
-            # 统一封面管道下，不传 thumbnail_url 时不自动生成
-            assert not video.thumbnail_url
+            assert video.thumbnail_url == generated_thumb_url
         finally:
             session.close()
 
-    def test_no_thumbnail_does_not_block(self):
-        """统一封面管道下，缩略图不再在 dedup 阶段生成。"""
+    def test_thumbnail_generation_failure_does_not_block(self):
+        """缩略图生成失败不影响主流程。"""
         from video_processing.dedup_helpers import create_video_record_and_dedup
 
         engine = create_engine("sqlite:///:memory:")
@@ -423,20 +437,24 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
 
-                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,
-                )
+                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,
+                    )
 
             assert result == 1  # 不阻断
 

tests/unit/test_generation_cover.py

@@ -92,463 +92,3 @@ 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_all_fallbacks_fail_returns_400(self):
-        """当所有步骤都找不到 cover_url 时，返回 400 而非 500。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-        from fastapi import HTTPException
-
-        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.storage.get_shared_storage_service") as mock_storage_getter,
-        ):
-            mock_repo = MagicMock()
-            mock_repo.get.return_value = mock_task
-            # No tasks found by source_edit_plan_id or user+template
-            mock_repo.list_by_source_edit_plan.return_value = []
-            mock_repo.list_latest_completed_preview.return_value = []
-            mock_repo_cls.return_value = mock_repo
-
-            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
-
-            from app.api.routes.generation_cover import generate_cover
-
-            with pytest.raises(HTTPException) as exc_info:
-                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(),
-                )
-
-            assert exc_info.value.status_code == 400
-            assert "封面尚未生成" in exc_info.value.detail
-
-    def test_cover_url_found_via_source_edit_plan(self):
-        """步骤B：通过 source_edit_plan_id 找到预览任务的 cover_url。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-
-        mock_plan = MagicMock()
-        # No generation_task_id, so step A is skipped
-        mock_plan.config = {
-            "rendered_storage_key": "rendered/plan-x/video.mp4",
-        }
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_template_svc = MagicMock()
-
-        # Preview task found by source_edit_plan_id with cover_url
-        mock_preview_task = MagicMock()
-        mock_preview_task.id = "preview-task-abc"
-        mock_preview_task.status = "completed"
-        mock_preview_task.cover_url = "https://oss.example.com/rendered/preview/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.list_by_source_edit_plan.return_value = [mock_preview_task]
-            mock_repo_cls.return_value = mock_repo
-
-            mock_normalize.return_value = {
-                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/preview/cover.jpg"}
-            }
-
-            from app.api.routes.generation_cover import generate_cover
-
-            result = generate_cover(
-                body=body,
-                template_id="template-x",
-                plan_id="plan-x",
-                services=(mock_template_svc, mock_plan_svc),
-                db=mock_db,
-                current_user=MagicMock(),
-            )
-
-            assert result.cover["image_url"] == "https://oss.example.com/rendered/preview/cover.jpg"
-            mock_repo.list_by_source_edit_plan.assert_called_once_with("plan-x")
-
-    def test_cover_url_found_via_user_template(self):
-        """步骤C：通过 user+template 找到预览任务的 cover_url。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-
-        mock_plan = MagicMock()
-        mock_plan.config = {
-            "rendered_storage_key": "rendered/plan-y/video.mp4",
-        }
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_template_svc = MagicMock()
-
-        # Step B finds nothing, step C finds a task
-        mock_preview_task = MagicMock()
-        mock_preview_task.id = "preview-task-def"
-        mock_preview_task.status = "completed"
-        mock_preview_task.cover_url = "https://oss.example.com/rendered/user-template-cover.jpg"
-
-        mock_current_user = MagicMock()
-        mock_current_user.user.id = "user-123"
-        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.list_by_source_edit_plan.return_value = []
-            mock_repo.list_latest_completed_preview.return_value = [mock_preview_task]
-            mock_repo_cls.return_value = mock_repo
-
-            mock_normalize.return_value = {
-                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/user-template-cover.jpg"}
-            }
-
-            from app.api.routes.generation_cover import generate_cover
-
-            result = generate_cover(
-                body=body,
-                template_id="template-y",
-                plan_id="plan-y",
-                services=(mock_template_svc, mock_plan_svc),
-                db=mock_db,
-                current_user=mock_current_user,
-            )
-
-            assert result.cover["image_url"] == "https://oss.example.com/rendered/user-template-cover.jpg"
-            mock_repo.list_latest_completed_preview.assert_called_once_with(
-                user_id="user-123",
-                template_id="template-y",
-            )
-
-
-class TestSourceEditPlanFallback:
-    """测试步骤 2.5：通过 source_edit_plan_id 查找预览视频兜底逻辑。"""
-
-    def test_step25_finds_video_by_source_edit_plan_id(self):
-        """当步骤1和步骤2都找不到时，步骤2.5通过source_edit_plan_id找到预览视频和封面。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-
-        # plan.config 没有 rendered_storage_key 和 generation_task_id
-        mock_plan = MagicMock()
-        mock_plan.config = {}
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_template_svc = MagicMock()
-
-        # Mock preview task found by source_edit_plan_id — with cover_url
-        mock_preview_task = MagicMock()
-        mock_preview_task.id = "preview-task-789"
-        mock_preview_task.status = "completed"
-        mock_preview_task.is_preview = True
-        mock_preview_task.cover_url = "https://oss.example.com/rendered/cover.jpg"
-
-        # Mock generated video
-        mock_video = MagicMock()
-        mock_video.file_url = "rendered/plan-x/video.mp4"
-
-        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.get_generated_video_repository") as mock_video_repo,
-            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-        ):
-            mock_repo = MagicMock()
-            mock_repo.get.return_value = None
-            # Both video lookup (step 2.5) and cover_url lookup (step B) use this
-            mock_repo.list_by_source_edit_plan.return_value = [mock_preview_task]
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = [mock_video]
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_normalize.return_value = {
-                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/cover.jpg"}
-            }
-
-            from app.api.routes.generation_cover import generate_cover
-
-            result = generate_cover(
-                body=body,
-                template_id="template-1",
-                plan_id="plan-x",
-                services=(mock_template_svc, mock_plan_svc),
-                db=mock_db,
-                current_user=MagicMock(),
-            )
-
-            # Verify step 2.5 was called for video
-            mock_repo.list_by_source_edit_plan.assert_called_with("plan-x")
-            # Cover was found via unified pipeline step B
-            assert result.cover["image_url"] == "https://oss.example.com/rendered/cover.jpg"
-
-    def test_step25_skips_non_completed_or_non_preview_tasks(self):
-        """步骤2.5跳过非completed或非is_preview的任务，继续到步骤3。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-
-        mock_plan = MagicMock()
-        mock_plan.config = {}
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_template_svc = MagicMock()
-
-        # Task that is not completed
-        mock_task_failed = MagicMock()
-        mock_task_failed.id = "task-failed"
-        mock_task_failed.status = "failed"
-        mock_task_failed.is_preview = True
-        mock_task_failed.cover_url = ""
-
-        # Task that is not preview
-        mock_task_full = MagicMock()
-        mock_task_full.id = "task-full"
-        mock_task_full.status = "completed"
-        mock_task_full.is_preview = False
-        mock_task_full.cover_url = ""
-
-        mock_db = MagicMock()
-
-        # Step 3 fallback finds a valid preview task WITH cover_url
-        mock_step3_task = MagicMock()
-        mock_step3_task.id = "step3-task"
-        mock_step3_task.status = "completed"
-        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"
-
-        mock_video = MagicMock()
-        mock_video.file_url = "rendered/step3/video.mp4"
-
-        body = GenerateCoverRequest(cover_type="ai_frame")
-
-        with (
-            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
-            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
-            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
-        ):
-            mock_repo = MagicMock()
-            mock_repo.get.return_value = None
-            mock_repo.list_by_source_edit_plan.return_value = [mock_task_failed, mock_task_full]
-            # Video step 3 and cover step C both use list_latest_completed_preview
-            mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = [mock_video]
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_normalize.return_value = {
-                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
-            }
-
-            mock_storage_svc = MagicMock()
-            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
-            mock_storage_getter.return_value = mock_storage_svc
-
-            from app.api.routes.generation_cover import generate_cover
-
-            result = generate_cover(
-                body=body,
-                template_id="template-1",
-                plan_id="plan-y",
-                services=(mock_template_svc, mock_plan_svc),
-                db=mock_db,
-                current_user=MagicMock(),
-            )
-
-            # Step 2.5 found tasks but none matched -> step 3 should be called
-            mock_repo.list_by_source_edit_plan.assert_called()
-            mock_repo.list_latest_completed_preview.assert_called()
-            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"
-
-    def test_step25_exception_does_not_block_step3(self):
-        """步骤2.5异常时不影响步骤3兜底（视频和封面都通过步骤3找到）。"""
-        from unittest.mock import MagicMock, patch
-
-        from app.api.routes.generation_cover import GenerateCoverRequest
-
-        mock_plan = MagicMock()
-        mock_plan.config = {}
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-        mock_template_svc = MagicMock()
-
-        mock_step3_task = MagicMock()
-        mock_step3_task.id = "step3-task"
-        mock_step3_task.status = "completed"
-        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"
-
-        mock_video = MagicMock()
-        mock_video.file_url = "rendered/step3/video.mp4"
-
-        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.get_generated_video_repository") as mock_video_repo,
-            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
-            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
-        ):
-            mock_repo = MagicMock()
-            mock_repo.get.return_value = None
-            # Both video step 2.5 and cover step B raise
-            mock_repo.list_by_source_edit_plan.side_effect = RuntimeError("db error")
-            # Step 3 / step C succeeds
-            mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
-            mock_repo_cls.return_value = mock_repo
-
-            mock_usecase = MagicMock()
-            mock_usecase.execute.return_value = [mock_video]
-            mock_usecase_cls.return_value = mock_usecase
-
-            mock_normalize.return_value = {
-                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
-            }
-
-            mock_storage_svc = MagicMock()
-            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
-            mock_storage_getter.return_value = mock_storage_svc
-
-            from app.api.routes.generation_cover import generate_cover
-
-            result = generate_cover(
-                body=body,
-                template_id="template-1",
-                plan_id="plan-z",
-                services=(mock_template_svc, mock_plan_svc),
-                db=mock_db,
-                current_user=MagicMock(),
-            )
-
-            # Step 3 was called after step 2.5 failed
-            mock_repo.list_latest_completed_preview.assert_called()
-            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"
-
-
-class TestStrayLoggerRemoved:
-    """验证多余的 logger.info(plan_id, generation_task_id) 已被删除。"""
-
-    def test_no_stray_logger_call_in_source(self):
-        """源码中不应存在 logger.info(plan_id, generation_task_id) 这样的调用。"""
-        import inspect
-
-        from app.api.routes import generation_cover
-
-        source = inspect.getsource(generation_cover)
-        # The stray call was logger.info(\n plan_id,\n generation_task_id,\n)
-        # with no format string — should not exist
-        assert (
-            "logger.info(\n                            plan_id," not in source
-        ), "Stray logger.info(plan_id, generation_task_id) should be removed"

tests/unit/test_mediakit_cover.py

@@ -141,13 +141,93 @@ class TestMediaKitClient:
 
 
 class TestAICoverService:
-    """AI 封面服务测试（统一封面管道后）。"""
+    """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
 
-    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="封面数据不可用"):
+        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="无法从视频抽帧"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -155,22 +235,11 @@ 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):
-        """ai_frame without video URL still raises RuntimeError."""
+        """没有视频 URL 时抛出 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"],
@@ -207,6 +276,23 @@ 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_overlay_title_on_ass.py

@@ -1,242 +0,0 @@
-"""测试 ASR 字幕路径的标题叠加功能。
-
-验证：
-1. _overlay_title_on_ass 函数正确地将标题事件追加到 ASR 生成的 ASS 文件中
-2. _maybe_generate_ass 在 ASR 路径中正确叠加标题
-3. ASR 无结果但有标题时，仍然生成标题 ASS
-4. ASR 失败但有标题时，降级生成标题 ASS
-"""
-
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from apps.worker.video_processing.unified_render_service import _overlay_title_on_ass
-
-
-class TestOverlayTitleOnAss:
-    """_overlay_title_on_ass 函数测试"""
-
-    def test_overlay_title_adds_style_and_dialogue(self, tmp_path):
-        """标题 Style 和 Dialogue 正确插入 ASS 文件"""
-        # 准备一个模拟 ASR 生成的 ASS 文件
-        ass_content = """[Script Info]
-ScriptType: v4.00+
-PlayResX: 1280
-PlayResY: 720
-ScaledBorderAndShadow: yes
-WrapStyle: 2
-Encoding: UTF-8
-
-[V4+ Styles]
-Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
-Style: Default,思源黑体,24,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1.5,0,2,40,40,60,1
-
-[Events]
-Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
-Dialogue: 0,0:00:00.00,0:00:10.00,Default,,0,0,0,,这是ASR字幕
-"""
-        ass_path = tmp_path / "test.ass"
-        ass_path.write_text(ass_content, encoding="utf-8")
-
-        # 叠加标题
-        _overlay_title_on_ass(
-            ass_path,
-            title_text="测试标题",
-            title_config={"position": "top", "font": "思源黑体", "size": 48, "color": "#ffffff"},
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-        )
-
-        # 验证结果
-        result = ass_path.read_text(encoding="utf-8")
-        assert "Style: TitleStyle" in result, "TitleStyle 应被插入"
-        assert "测试标题" in result, "标题文本应出现在 Dialogue 中"
-        # 原有的 ASR 字幕应该保留
-        assert "这是ASR字幕" in result, "原有 ASR 字幕应保留"
-        # TitleStyle 应该在 Default Style 之后
-        lines = result.splitlines()
-        style_lines = [i for i, ln in enumerate(lines) if ln.startswith("Style:")]
-        assert len(style_lines) >= 2, "应有至少两个 Style 行"
-
-    def test_overlay_title_empty_text_noop(self, tmp_path):
-        """空标题文本时不修改 ASS 文件"""
-        ass_content = "[Script Info]\n\n[V4+ Styles]\nStyle: Default,test\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\nDialogue: 0,0:00:00.00,0:00:10.00,Default,,0,0,0,,test\n"
-        ass_path = tmp_path / "test.ass"
-        ass_path.write_text(ass_content, encoding="utf-8")
-
-        _overlay_title_on_ass(
-            ass_path,
-            title_text="",
-            title_config={},
-            video_width=1280,
-            video_height=720,
-            video_duration=10.0,
-        )
-
-        result = ass_path.read_text(encoding="utf-8")
-        assert "TitleStyle" not in result, "空标题不应添加 TitleStyle"
-
-    def test_overlay_title_preserves_asr_events(self, tmp_path):
-        """叠加标题后 ASR 字幕事件保持不变"""
-        ass_content = """[Script Info]
-ScriptType: v4.00+
-PlayResX: 1920
-PlayResY: 1080
-
-[V4+ Styles]
-Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
-Style: Default,思源黑体,24,&H00FFFFFF,&H000000FF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1.5,0,2,40,40,60,1
-
-[Events]
-Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
-Dialogue: 0,0:00:00.50,0:00:03.00,Default,,0,0,0,,第一段字幕
-Dialogue: 0,0:00:03.50,0:00:06.00,Default,,0,0,0,,第二段字幕
-Dialogue: 0,0:00:06.50,0:00:10.00,Default,,0,0,0,,第三段字幕
-"""
-        ass_path = tmp_path / "test.ass"
-        ass_path.write_text(ass_content, encoding="utf-8")
-
-        _overlay_title_on_ass(
-            ass_path,
-            title_text="我的标题",
-            title_config={"position": "top", "size": 48},
-            video_width=1920,
-            video_height=1080,
-            video_duration=10.0,
-        )
-
-        result = ass_path.read_text(encoding="utf-8")
-        # 所有 ASR 字幕段都应保留
-        assert "第一段字幕" in result
-        assert "第二段字幕" in result
-        assert "第三段字幕" in result
-        # 标题也应存在
-        assert "我的标题" in result
-
-
-class TestMaybeGenerateAssWithTitle:
-    """_maybe_generate_ass 方法在 ASR 路径中标题叠加的集成测试"""
-
-    def _make_service(self, tmp_path, plan_config, asr_service=None):
-        """创建简化的 UnifiedRenderService 实例用于测试"""
-        from apps.worker.video_processing.unified_render_service import UnifiedRenderService
-
-        service = object.__new__(UnifiedRenderService)
-        service.plan = MagicMock()
-        service.plan.id = "test_plan_001"
-        service.plan.config = plan_config
-        service.work_dir = tmp_path
-        service.output_width = 1280
-        service.output_height = 720
-        service.asr_service = asr_service
-        service._asr_timeline_cached = False
-        service._asr_timeline_cache = None
-        return service
-
-    def test_asr_path_with_title_overlays_title(self, tmp_path):
-        """ASR 路径 + 有标题 → 标题叠加到 ASS 文件"""
-        plan_config = {
-            "title": {
-                "text": "测试标题",
-                "enabled": True,
-                "position": "top",
-                "size": 48,
-                "font": "思源黑体",
-                "color": "#ffffff",
-            },
-            "subtitle": {"enabled": True, "auto_generated": True},
-        }
-
-        # Mock ASR service
-        mock_asr = MagicMock()
-        service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
-
-        # Mock _generate_asr_subtitles to return a timeline with segments
-        mock_timeline = MagicMock()
-        mock_segment = MagicMock()
-        mock_segment.start = 0.0
-        mock_segment.end = 3.0
-        mock_segment.text = "ASR识别的文字"
-        mock_timeline.segments = [mock_segment]
-        mock_timeline.segment_count = 1
-
-        with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
-            result = service._maybe_generate_ass(video_duration=10.0)
-
-        assert result is not None, "应生成 ASS 文件"
-        content = result.read_text(encoding="utf-8")
-        assert "测试标题" in content, "标题应出现在 ASS 文件中"
-        assert "ASR识别的文字" in content, "ASR 字幕也应保留"
-
-    def test_asr_no_result_with_title_generates_title_ass(self, tmp_path):
-        """ASR 无结果 + 有标题 → 仍然生成标题 ASS"""
-        plan_config = {
-            "title": {"text": "仅标题", "enabled": True, "position": "top", "size": 48},
-            "subtitle": {"enabled": True, "auto_generated": True},
-        }
-
-        mock_asr = MagicMock()
-        service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
-
-        # Mock ASR returns empty timeline
-        mock_timeline = MagicMock()
-        mock_timeline.segments = []
-
-        with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
-            result = service._maybe_generate_ass(video_duration=10.0)
-
-        assert result is not None, "有标题时应生成 ASS 文件"
-        content = result.read_text(encoding="utf-8")
-        assert "仅标题" in content, "标题应出现在 ASS 文件中"
-
-    def test_asr_failure_with_title_generates_title_ass(self, tmp_path):
-        """ASR 失败 + 有标题 → 降级生成标题 ASS"""
-        plan_config = {
-            "title": {"text": "降级标题", "enabled": True, "position": "top", "size": 48},
-            "subtitle": {"enabled": True, "auto_generated": True},
-        }
-
-        mock_asr = MagicMock()
-        service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
-
-        # Mock ASR raises exception
-        with patch.object(service, "_generate_asr_subtitles", side_effect=RuntimeError("ASR error")):
-            result = service._maybe_generate_ass(video_duration=10.0)
-
-        assert result is not None, "ASR 失败但有标题时应生成 ASS 文件"
-        content = result.read_text(encoding="utf-8")
-        assert "降级标题" in content, "标题应出现在降级 ASS 文件中"
-
-    def test_overlay_failure_preserves_asr_data(self, tmp_path):
-        """_overlay_title_on_ass 抛异常时，ASR 生成的 ASS 文件应保留并返回"""
-        plan_config = {
-            "title": {"text": "测试标题", "enabled": True, "position": "top", "size": 48},
-            "subtitle": {"enabled": True, "auto_generated": True},
-        }
-
-        mock_asr = MagicMock()
-        service = self._make_service(tmp_path, plan_config, asr_service=mock_asr)
-
-        mock_timeline = MagicMock()
-        mock_segment = MagicMock()
-        mock_segment.start = 0.0
-        mock_segment.end = 3.0
-        mock_segment.text = "ASR识别的文字"
-        mock_timeline.segments = [mock_segment]
-        mock_timeline.segment_count = 1
-
-        with patch.object(service, "_generate_asr_subtitles", return_value=mock_timeline):
-            with patch(
-                "apps.worker.video_processing.unified_render_service._overlay_title_on_ass",
-                side_effect=RuntimeError("模拟叠加标题失败"),
-            ):
-                result = service._maybe_generate_ass(video_duration=10.0)
-
-        # 即使 _overlay_title_on_ass 失败，仍返回 ASS 文件
-        assert result is not None, "应返回 ASS 文件路径"
-        assert result.exists(), "ASS 文件应存在"
-        content = result.read_text(encoding="utf-8")
-        assert "ASR识别的文字" in content, "ASR 字幕数据应保留"

tests/unit/test_preview_title_config.py

@@ -1,190 +0,0 @@
-"""Tests for preview title_config feature.
-
-验证预览 API 的 title_config 字段和 Worker 的标题配置解析逻辑。
-"""
-
-from __future__ import annotations
-
-import json
-
-import pytest
-
-
-class TestPreviewTitleConfigSchema:
-    """测试 CreatePreviewGenerationTaskRequest 的 title_config 字段."""
-
-    def test_title_config_default_empty(self):
-        """title_config 默认为空 dict."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-        )
-        assert req.title_config == {}
-
-    def test_title_config_with_text(self):
-        """传入标题文本."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-            title_config={"text": "测试标题"},
-        )
-        assert req.title_config["text"] == "测试标题"
-
-    def test_title_config_with_full_style(self):
-        """传入完整标题样式配置."""
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-
-        config = {
-            "text": "我的视频标题",
-            "font": "思源黑体",
-            "font_size": 48,
-            "font_color": "#ffffff",
-            "position": "top",
-            "bold": True,
-            "stroke": 2,
-            "shadow": True,
-        }
-        req = CreatePreviewGenerationTaskRequest(
-            template_id="test_template",
-            asset_ids=["asset1"],
-            title_config=config,
-        )
-        assert req.title_config["text"] == "我的视频标题"
-        assert req.title_config["font_size"] == 48
-        assert req.title_config["position"] == "top"
-
-
-class TestCommandTitleConfig:
-    """测试 CreateGenerationTaskCommand 的 title_config 字段."""
-
-    def test_command_has_title_config(self):
-        """Command 包含 title_config 字段."""
-        from packages.application.generation_tasks import CreateGenerationTaskCommand
-
-        cmd = CreateGenerationTaskCommand(
-            title_config={"text": "hello", "font_size": 32},
-        )
-        assert cmd.title_config["text"] == "hello"
-        assert cmd.title_config["font_size"] == 32
-
-    def test_command_title_config_default_empty(self):
-        """Command 的 title_config 默认为空 dict."""
-        from packages.application.generation_tasks import CreateGenerationTaskCommand
-
-        cmd = CreateGenerationTaskCommand()
-        assert cmd.title_config == {}
-
-
-class TestWorkerTitleConfigParsing:
-    """测试 Worker 渲染时的标题配置解析逻辑."""
-
-    def test_json_format_parsing(self):
-        """JSON 格式的 custom_title 能正确解析."""
-        config = {"text": "测试标题", "font_size": 48, "font_color": "#ff0000"}
-        custom_title = json.dumps(config, ensure_ascii=False)
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is not None
-        assert parsed["text"] == "测试标题"
-        assert parsed["font_size"] == 48
-
-    def test_plain_text_fallback(self):
-        """纯文本的 custom_title 不触发 JSON 解析."""
-        custom_title = "简单的标题文字"
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is None
-
-    def test_invalid_json_fallback(self):
-        """无效 JSON 的 custom_title 降级为纯文本."""
-        custom_title = "{invalid json"
-
-        ct_stripped = custom_title.strip()
-        parsed = None
-        if ct_stripped.startswith("{"):
-            try:
-                parsed = json.loads(ct_stripped)
-            except (json.JSONDecodeError, ValueError):
-                parsed = None
-
-        assert parsed is None
-
-    def test_json_without_text_skipped(self):
-        """JSON 格式但缺少 text 字段时，跳过标题注入."""
-        config = {"font_size": 48}
-        custom_title = json.dumps(config, ensure_ascii=False)
-
-        ct_stripped = custom_title.strip()
-        parsed = json.loads(ct_stripped)
-        title_text = (parsed.get("text") or "").strip()
-
-        assert title_text == ""
-
-    def test_style_key_mapping(self):
-        """前端字段名正确映射到 ASS 字段名."""
-        config = {
-            "text": "标题",
-            "font_size": 48,
-            "font_color": "#ffffff",
-            "font_preset": "思源黑体",
-        }
-
-        style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
-        title_cfg = {}
-        for key in style_keys:
-            if key in config and config[key] is not None:
-                mapped_key = {
-                    "font_size": "size",
-                    "font_color": "color",
-                    "font_preset": "font",
-                }.get(key, key)
-                title_cfg[mapped_key] = config[key]
-
-        assert title_cfg["size"] == 48
-        assert title_cfg["color"] == "#ffffff"
-        assert title_cfg["font"] == "思源黑体"
-
-
-class TestPreviewRouteTitleConfigPassing:
-    """测试预览路由正确序列化 title_config 到 custom_title."""
-
-    def test_title_config_serialization(self):
-        """title_config 序列化为 JSON 字符串."""
-        title_config = {
-            "text": "我的标题",
-            "font_size": 32,
-            "font_color": "#d4a843",
-        }
-        serialized = json.dumps(title_config, ensure_ascii=False)
-
-        parsed = json.loads(serialized)
-        assert parsed["text"] == "我的标题"
-        assert parsed["font_size"] == 32
-
-    def test_empty_title_config_produces_empty_string(self):
-        """空 title_config 时 custom_title 为空字符串."""
-        title_config = {}
-        title_text = (title_config.get("text") or "").strip()
-        custom_title_value = ""
-        if title_text:
-            custom_title_value = json.dumps(title_config, ensure_ascii=False)
-
-        assert custom_title_value == ""

tests/unit/test_render_adapter.py

@@ -348,32 +348,24 @@ 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.extract_first_frame",
-            return_value=_fake_frame.name,
+            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+            return_value=fake_thumb,
         ):
             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
-        # cover_url from upload_to_oss (mocked globally)
-        assert result.thumbnail_url == "https://oss.example.com/out.mp4"
+        assert result.thumbnail_url == fake_thumb
 
     @patch("video_processing.render_adapter.upload_to_oss")
     @patch("video_processing.render_adapter.UnifiedRenderService")
@@ -405,8 +397,8 @@ class TestRenderPlan:
         adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         with patch(
-            "video_processing.thumbnail_generator.extract_first_frame",
-            side_effect=RuntimeError("ffmpeg not available"),
+            "video_processing.thumbnail_generator.generate_and_upload_thumbnail",
+            side_effect=RuntimeError("cv2 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):