fix: 修复PreviewVideoPanel JSX结构-将视频+Canvas叠加层移出!showTitlePreview块

- 预览成功+Canvas标题叠加层提升到顶层，仅在 hasPreview 条件下渲染
- 空状态/生成中/生成失败/预览信息添加 !showTitlePreview 守卫
- 修复 Canvas overlay 在 showTitlePreview=true 时永远不渲染的死代码问题

alembic/versions/055_cover_templates.py

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

alembic/versions/056_fix_cover_templates_config.py

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

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

@@ -31,6 +31,8 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=["Generation"])
 
 
+
+
 # ── Schemas ──────────────────────────────────────────────────────────────
 
 
@@ -105,10 +107,6 @@ 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",
@@ -116,6 +114,33 @@ 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:
@@ -185,28 +210,103 @@ def generate_cover(
             detail=f"获取预览视频URL失败: {e}",
         ) from e
 
-    # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回）
-    cover_candidates = (plan.config or {}).get("cover_candidates", [])
-    if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"):
-        logger.info(
-            "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d",
-            plan_id,
-            len(cover_candidates),
-        )
-        first_frame = cover_candidates[0]
-        cover_data = {
-            "type": "ai_frame",
-            "image_url": first_frame.get("image_url", ""),
-            "frame_time": first_frame.get("frame_time", 0.0),
-            "confidence": 0.9,
-        }
-        if cover_data["image_url"]:
+    # 统一封面管道：优先从 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,
+            }
             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,6 +5,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -267,6 +268,20 @@ 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:
@@ -290,6 +305,7 @@ 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, description="输出视频宽度")
-    output_height: int = Field(default=1920, description="输出视频高度")
+    output_width: int = Field(default=1080, ge=100, description="输出视频宽度")
+    output_height: int = Field(default=1920, ge=100, description="输出视频高度")
     cover_url: str = Field(default="", description="自定义封面图片 URL")
     custom_title: str = Field(default="", description="自定义视频标题")
 
@@ -181,6 +181,10 @@ 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,15 +200,7 @@ test.describe("Core generation flow", () => {
     await expect(page.getByRole("heading", { name: /选择配音/ })).toBeVisible({ timeout: 15000 })
     await page.getByRole("button", { name: "下一步" }).click()
 
-    // 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
+    // Step 4: title（新顺序：标题在预览之前）
     await expect(page.getByRole("heading", { name: /选择标题/ })).toBeVisible({ timeout: 15000 })
     // 等待组件完全渲染
     await page.waitForTimeout(2000)
@@ -222,6 +214,14 @@ 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,9 +258,12 @@ test.describe("Core generation flow", () => {
     // Generate API may return 400 in test env if template has no ready segments
     // That is OK for a wizard flow smoke test
     if (genResp.ok()) {
-      const genData = (await genResp.json()) as { plan_id: string; generation_task_id: string }
-      expect(genData.plan_id).toBeTruthy()
-      expect(genData.generation_task_id).toBeTruthy()
+      const genData = (await genResp.json()) as {
+        items: Array<{ id: string; status: string }>
+        total: number
+      }
+      expect(genData.items.length).toBeGreaterThan(0)
+      expect(genData.items[0].id).toBeTruthy()
     } else {
       console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
     }

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

@@ -13,6 +13,17 @@ 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,7 +12,10 @@
 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"
@@ -24,7 +27,7 @@ import GenerateStepActions from "./components/GenerateStepActions"
 import { useGenerateFormState } from "./hooks/useGenerateFormState"
 import { useStepNavigation } from "./hooks/useStepNavigation"
 import { useGenerateVideo } from "./hooks/useGenerateVideo"
-import { useStep4Preview } from "./hooks/useStep4Preview"
+import { useStep5Preview } from "./hooks/useStep5Preview"
 import "./generate.css"
 
 const GeneratePage: React.FC = () => {
@@ -71,6 +74,20 @@ 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()
 
@@ -96,8 +113,8 @@ const GeneratePage: React.FC = () => {
     return id ? [id] : []
   }, [voiceMode, selectedVoice, selectedClonedVoice])
 
-  /* ── Step4 预览生成（多预览 + voice_ids） ── */
-  const step4Preview = useStep4Preview({
+  /* ── Step5 预览生成（多预览 + voice_ids） ── */
+  const step5Preview = useStep5Preview({
     templates: userTemplates,
     selectedTemplate,
     materialMode,
@@ -108,6 +125,7 @@ const GeneratePage: React.FC = () => {
     voiceIds: previewVoiceIds,
     voiceLibraryId: selectedVoice || undefined,
     previewCount,
+    titleSettings,
   })
 
   /* ── 步骤导航 ── */
@@ -119,7 +137,7 @@ const GeneratePage: React.FC = () => {
     selectedMaterials,
     smartSelectedIds,
     titleSettings,
-    previewReady: step4Preview.canProceed,
+    previewReady: step5Preview.canProceed,
   })
 
   /* ── 视频生成核心逻辑 ── */
@@ -150,7 +168,7 @@ const GeneratePage: React.FC = () => {
     autoSubtitles,
     bgm,
     generateCount,
-    previewTaskId: step4Preview.selectedTaskId,
+    previewTaskId: step5Preview.selectedTaskId,
   })
 
   /* ================================================================
@@ -207,20 +225,20 @@ const GeneratePage: React.FC = () => {
             onDismissError={handleDismissError}
             presetVoices={presetVoices}
             videoRatio={videoRatio}
-            /* Step4 多预览 */
+            /* Step5 多预览 */
             previewCount={previewCount}
             onPreviewCountChange={setPreviewCount}
-            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}
+            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}
           />
 
           <GenerateStepActions
@@ -236,22 +254,24 @@ const GeneratePage: React.FC = () => {
 
         {/* ════ 右侧：预览 + 生成结果 ════ */}
         <div className="xx-generate-right-col">
-          {/* 预览视频面板（Step4+ 常驻，展示选中的预览） */}
+          {/* 预览视频面板（Step4+ 常驻，Step4 显示标题预览，Step5+ 显示预览视频） */}
           {currentStep >= 4 && (
             <PreviewVideoPanel
-              previewStatus={step4Preview.previewStatus}
-              previewResult={step4Preview.previewResult}
-              previewError={step4Preview.previewError}
-              progress={step4Preview.progress}
+              previewStatus={step5Preview.previewStatus}
+              previewResult={step5Preview.previewResult}
+              previewError={step5Preview.previewError}
+              progress={step5Preview.progress}
               videoRatio={videoRatio}
-              onRegenerate={step4Preview.regeneratePreview}
+              onRegenerate={step5Preview.regeneratePreview}
               titleText={titleSettings.title}
-              titleSettings={currentStep >= 5 ? titleSettings : undefined}
+              titleSettings={titleSettings}
+              showTitlePreview={currentStep === 4}
+              sourceVideoUrl={sourceVideoUrl}
             />
           )}
 
-          {/* 正式生成结果（Step5+ 才显示） */}
-          {currentStep >= 5 && (
+          {/* 正式生成结果（Step6+ 才显示） */}
+          {currentStep >= 6 && (
             <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/useStep4Preview"
+import type { PreviewItem, PreviewStatus } from "../hooks/useStep5Preview"
 import Step1TemplateSelect from "../components/Step1TemplateSelect"
 import Step2MaterialSelect from "../components/Step2MaterialSelect"
 import Step3VoiceSelect from "../components/Step5VoiceSelect"
-import Step4GeneratePreview from "../components/Step4GeneratePreview"
-import Step5TitleSettings from "../components/Step4TitleSettings"
+import Step5GeneratePreview from "../components/Step5GeneratePreview"
+import Step4TitleSettings from "../components/Step4TitleSettings"
 import Step6CoverSettings from "../components/Step6CoverSettings"
 import Step7ConfirmGenerate from "../components/Step7ConfirmGenerate"
 import type { GeneratedVideo } from "@/api/template-editor"
@@ -157,7 +157,14 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
       )
     case 4:
       return (
-        <Step4GeneratePreview
+        <Step4TitleSettings
+          titleSettings={titleSettings}
+          onTitleSettingsChange={onTitleSettingsChange}
+        />
+      )
+    case 5:
+      return (
+        <Step5GeneratePreview
           templateName={previewTemplateName}
           materialCount={previewMaterialCount}
           duration={duration}
@@ -175,13 +182,6 @@ 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,8 +13,10 @@
  */
 import React, { useRef, useEffect, useCallback } from "react"
 import { PlayCircleOutlined, LoadingOutlined } from "@ant-design/icons"
-import type { PreviewResult, PreviewStatus } from "../hooks/useStep4Preview"
+import type { PreviewResult, PreviewStatus } from "../hooks/useStep5Preview"
 import type { TitleSettings } from "../types"
+import { drawTitleOnCanvas } from "../utils/drawTitleOnCanvas"
+import TitlePreviewCanvas from "./title/TitlePreviewCanvas"
 
 interface PreviewVideoPanelProps {
   previewStatus: PreviewStatus
@@ -23,134 +25,14 @@ interface PreviewVideoPanelProps {
   progress: number
   videoRatio: string
   onRegenerate: () => void
-  /** 标题文字（Step5 起传入） */
+  /** 标题文字 */
   titleText?: string
-  /** 标题样式设置（Step5 起传入） */
+  /** 标题样式设置 */
   titleSettings?: TitleSettings
-}
-
-/* ── 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
-  }
+  /** Step4 标题预览模式 */
+  showTitlePreview?: boolean
+  /** 素材视频 URL（用于 Step4 标题预览背景） */
+  sourceVideoUrl?: string
 }
 
 /* ── 组件 ── */
@@ -164,11 +46,12 @@ 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
@@ -293,12 +176,35 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
   return (
     <div className="xx-generate-preview">
       <div className="xx-preview-header">
-        <h3>预览视频</h3>
-        {hasPreview && <span className="xx-preview-badge">480p 预览版</span>}
+        <h3>{showTitlePreview ? "标题预览" : "预览视频"}</h3>
+        {hasPreview && !showTitlePreview && <span className="xx-preview-badge">480p 预览版</span>}
       </div>
 
-      {/* 空状态：还没生成预览 */}
-      {previewStatus === "idle" && (
+      {/* 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" && (
         <div className="xx-preview-empty">
           <PlayCircleOutlined
             style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
@@ -308,8 +214,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 生成中 */}
-      {isLoading && (
+      {/* 生成中（非 Step4 模式） */}
+      {!showTitlePreview && isLoading && (
         <div className="xx-preview-loading-panel">
           <div className="xx-preview-video" style={videoAspectStyle}>
             <div className="xx-preview-loading-center">
@@ -325,8 +231,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 生成失败 */}
-      {isError && (
+      {/* 生成失败（非 Step4 模式） */}
+      {!showTitlePreview && 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>
@@ -369,8 +275,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
         </div>
       )}
 
-      {/* 预览信息 */}
-      {hasPreview && previewResult && (
+      {/* 预览信息（非 Step4 模式） */}
+      {!showTitlePreview && hasPreview && previewResult && (
         <div className="xx-preview-info">
           <div className="xx-preview-info-row">
             <span>时长</span>

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

@@ -1,5 +1,5 @@
 /**
- * Step 4 生成预览组件（支持多预览）
+ * Step 5 生成预览组件（支持多预览）
  * 调用后端预览生成接口，展示多个真实视频预览（网格布局）
  */
 import React from "react"
@@ -12,9 +12,9 @@ import {
   ClockCircleOutlined,
 } from "@ant-design/icons"
 import { InputNumber } from "antd"
-import type { PreviewItem, PreviewStatus } from "../hooks/useStep4Preview"
+import type { PreviewItem, PreviewStatus } from "../hooks/useStep5Preview"
 
-interface Step4GeneratePreviewProps {
+interface Step5GeneratePreviewProps {
   templateName: string
   materialCount: string
   duration: number
@@ -39,7 +39,7 @@ const PREVIEW_COUNT_OPTIONS = [
   { value: 3, label: "3个" },
 ]
 
-const Step4GeneratePreview: React.FC<Step4GeneratePreviewProps> = ({
+const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
   templateName: _templateName,
   materialCount: _materialCount,
   videoRatio,
@@ -161,54 +161,57 @@ const Step4GeneratePreview: React.FC<Step4GeneratePreviewProps> = ({
                   transition: "all 0.2s",
                 }}
               >
-                {/* 缩略图/状态区域 */}
+                {/* 轻量卡片：深色背景 + 状态指示 */}
                 <div
                   style={{
                     aspectRatio,
-                    background: "#000",
+                    background: "#1a1a2e",
                     display: "flex",
+                    flexDirection: "column",
                     alignItems: "center",
                     justifyContent: "center",
                     position: "relative",
+                    gap: 8,
                   }}
                 >
-                  {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" || 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>
+                  {/* 中心：预览编号 */}
+                  <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 === "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>
+                    </div>
+                  )}
+                  {item.status === "ready" && (
+                    <CheckCircleFilled style={{ fontSize: 18, color: "#52c41a" }} />
+                  )}
                   {item.status === "error" && (
-                    <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>
+                    <ExclamationCircleFilled style={{ fontSize: 18, color: "#ef4444" }} />
                   )}
+
                   {/* 选中角标 */}
                   {isSelected && item.status === "ready" && (
                     <div
@@ -227,22 +230,6 @@ const Step4GeneratePreview: React.FC<Step4GeneratePreviewProps> = ({
                     </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>
             )
           })}
@@ -291,4 +278,4 @@ const Step4GeneratePreview: React.FC<Step4GeneratePreviewProps> = ({
   )
 }
 
-export default Step4GeneratePreview
+export default Step5GeneratePreview

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

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

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

@@ -1,5 +1,5 @@
 /**
- * Step 4 生成预览 Hook（支持多预览 + voice_ids）
+ * Step 5 生成预览 Hook（支持多预览 + voice_ids）
  * 调用 /generation/preview 接口创建多个预览任务，轮询状态直到全部完成
  */
 import { useState, useCallback, useMemo, useEffect, useRef } from "react"
@@ -8,6 +8,7 @@ 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 => {
@@ -26,7 +27,7 @@ const safeNumber = (val: unknown, fallback = 0): number => {
   return fallback
 }
 
-interface UseStep4PreviewProps {
+interface UseStep5PreviewProps {
   templates: EditingTemplate[]
   selectedTemplate: string
   materialMode: "manual" | "auto"
@@ -40,6 +41,8 @@ interface UseStep4PreviewProps {
   voiceLibraryId?: string
   /** 要生成的预览数量 */
   previewCount?: number
+  /** 标题设置（传递给后端，让预览视频包含标题） */
+  titleSettings?: TitleSettings
 }
 
 export type PreviewStatus = "idle" | "pending" | "generating" | "ready" | "error"
@@ -78,7 +81,7 @@ const createInitialItem = (index: number): PreviewItem => ({
   progress: 0,
 })
 
-export function useStep4Preview({
+export function useStep5Preview({
   templates,
   selectedTemplate,
   materialMode,
@@ -89,7 +92,8 @@ export function useStep4Preview({
   voiceIds,
   voiceLibraryId,
   previewCount = 1,
-}: UseStep4PreviewProps) {
+  titleSettings,
+}: UseStep5PreviewProps) {
   const templateName = useMemo(
     () => templates.find((t) => t.id === selectedTemplate)?.name ?? "未选择",
     [templates, selectedTemplate],
@@ -153,6 +157,7 @@ export function useStep4Preview({
     duration,
     videoRatio,
     voiceIds: [...(voiceIds || [])].sort().join(","),
+    titleSettings: titleSettings?.title || "",
   })
 
   useEffect(() => {
@@ -164,6 +169,7 @@ export function useStep4Preview({
       duration,
       videoRatio,
       [...(voiceIds || [])].sort().join(","),
+      titleSettings?.title || "",
     ].join("|")
 
     const prevKey = [
@@ -174,6 +180,7 @@ export function useStep4Preview({
       prevDepsRef.current.duration,
       prevDepsRef.current.videoRatio,
       prevDepsRef.current.voiceIds,
+      prevDepsRef.current.titleSettings,
     ].join("|")
 
     if (prevKey !== currentKey && items.some((it) => it.status !== "idle")) {
@@ -191,6 +198,7 @@ export function useStep4Preview({
       duration,
       videoRatio,
       voiceIds: [...(voiceIds || [])].sort().join(","),
+      titleSettings: titleSettings?.title || "",
     }
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, [
@@ -202,6 +210,7 @@ export function useStep4Preview({
     videoRatio,
     voiceIds,
     previewCount,
+    titleSettings,
   ])
 
   // 组件卸载时清理所有轮询
@@ -349,6 +358,19 @@ export function useStep4Preview({
           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
@@ -374,6 +396,7 @@ export function useStep4Preview({
     voiceIds,
     voiceLibraryId,
     previewCount,
+    titleSettings,
     clearPollTimer,
     pollPreviewStatus,
   ])
@@ -450,4 +473,4 @@ export function useStep4Preview({
   }
 }
 
-export default useStep4Preview
+export default useStep5Preview

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 && !previewReady) {
-      message.warning("请先生成剪辑预览")
+    if (currentStep === 4 && !titleSettings.title.trim()) {
+      message.warning("请选择或输入标题")
       return
     }
-    if (currentStep === 5 && !titleSettings.title.trim()) {
-      message.warning("请选择或输入标题")
+    if (currentStep === 5 && !previewReady) {
+      message.warning("请先生成剪辑预览")
       return
     }
     if (currentStep < 7) {

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

@@ -0,0 +1,126 @@
+/**
+ * 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/Step4GeneratePreview"
+import "@/pages/generate/components/Step5GeneratePreview"
 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/useStep4Preview"
+import "@/pages/generate/hooks/useStep5Preview"
 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

@@ -1,17 +0,0 @@
-/**
- * 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

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

apps/worker/video_processing/render_adapter.py

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

apps/worker/video_processing/thumbnail_generator.py

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

apps/worker/video_processing/unified_render_service.py

@@ -52,6 +52,7 @@ 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
@@ -131,6 +132,83 @@ _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:
     """统一渲染引擎。
 
@@ -480,7 +558,11 @@ 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)
@@ -515,14 +597,71 @@ 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 无结果，不生成字幕
+                    # 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
                     logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id)
                     return None
             except Exception:
-                # ASR 失败降级：不生成字幕，不阻断主流程
-                logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
+                # 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)
                 return None
 
         # 静态字幕模式（原有逻辑）
@@ -654,6 +793,8 @@ 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,6 +13,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import os
 import tempfile
@@ -983,9 +984,9 @@ def _load_template_plan_config(template_id: str) -> dict:
 
             # 从独立字段组装成 plan.config 格式
             plan_config: dict[str, Any] = {}
-            title_cfg = template.title_config or {}
-            subtitle_cfg = template.subtitle_config or {}
-            bgm_cfg = template.bgm_config or {}
+            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 {}
 
             if title_cfg:
                 plan_config["title"] = title_cfg
@@ -1123,14 +1124,15 @@ def _render_video(
     resolution: str = "",
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
-) -> tuple[Path, float]:
+    custom_title: str = "",
+) -> tuple[Path, float, str]:
     """渲染视频（含配音混音）。
 
-    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
+    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_url)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1169,6 +1171,61 @@ 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 + 低码率
@@ -1189,7 +1246,10 @@ 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(
@@ -1244,8 +1304,9 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    cover_url = getattr(render_result, "cover_url", "") or ""
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_url
 
 
 def _upload_and_record(
@@ -1256,6 +1317,7 @@ def _upload_and_record(
     editing_mode,
     user_id: str = "",
     video_name: str = "",
+    thumbnail_url: str = "",
 ) -> tuple[str, float, int, int]:
     """上传 OSS、创建视频记录并查重。
 
@@ -1501,12 +1563,22 @@ def generate_video(self, task_id: str) -> dict:
             # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串
             _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
             _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
+            # 防护：前端可能误传宽高比（如 parseInt("9:16") = 9），宽度 < 100 时忽略
+            if _ow < 100 or _oh < 100:
+                logger.warning(
+                    "[task_id=%s] output_width/output_height 异常 (%dx%d)，回退到默认",
+                    task_id,
+                    _ow,
+                    _oh,
+                )
+                _ow = OUTPUT_WIDTH
+                _oh = OUTPUT_HEIGHT
             if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT:
                 _resolved_resolution = f"{_ow}x{_oh}"
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_url = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1519,12 +1591,42 @@ def generate_video(self, task_id: str) -> dict:
                 resolution=_resolved_resolution,
                 bgm_config=task_info.get("bgm_config", {}),
                 voice_ids=task_info.get("voice_ids", []),
+                custom_title=task_info.get("custom_title", ""),
             )
 
             if gen_task:
                 gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                 _flush_logs(task_id, gen_task)
 
+            # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧）
+            if cover_url:
+                _cover_session = None
+                try:
+                    _cover_session = SessionLocal()
+                    try:
+                        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+                        _cover_model = (
+                            _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+                        )
+                        if _cover_model:
+                            _cover_model.cover_url = cover_url
+                            _cover_session.commit()
+                            logger.info(
+                                "[task_id=%s] 封面URL已持久化: %s",
+                                task_id,
+                                cover_url[:80],
+                            )
+                    finally:
+                        if _cover_session:
+                            _cover_session.close()
+                except Exception as cover_err:
+                    logger.warning(
+                        "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s",
+                        task_id,
+                        cover_err,
+                    )
+
             _update_task_progress(task_id, 80, "渲染完成")
 
             # ── 4. 上传 OSS + 查重记录 ───────────────────────────────────────
@@ -1537,6 +1639,7 @@ def generate_video(self, task_id: str) -> dict:
                 editing_mode=editing_mode,
                 user_id=user_id,
                 video_name=task_info.get("video_title", ""),
+                thumbnail_url=cover_url,
             )
 
             if gen_task:

apps/worker/worker_app/tasks/ingest.py

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

infra/docker/api.Dockerfile

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

packages/application/generation_tasks.py

@@ -32,6 +32,7 @@ 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,8 +13,6 @@ import random
 import time
 from typing import Any, Dict, List, Optional
 
-import requests as http_requests
-
 from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG
 from packages.shared.ai_client import get_doubao_client
 
@@ -354,93 +352,6 @@ def _transfer_cover_frame_to_storage(frame_url: str, plan_id: str) -> str:
         return frame_url
 
 
-def _extract_frames_with_ffmpeg(
-    video_url: str,
-    num_frames: int = 3,
-    timeout: int = 30,
-) -> list[dict]:
-    """用 FFmpeg 从远程视频 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）。
-
-    Args:
-        video_url: 视频 URL
-        num_frames: 抽帧数量
-        timeout: 单帧超时（秒）
-
-    Returns:
-        [{"local_path": "...", "frame_time": 5.0}, ...]
-    """
-    import re as _re
-    import tempfile
-    from pathlib import Path as _Path
-
-    from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
-
-    video_url = _re.sub(r"(?<!:)//", "/", video_url)
-
-    # 先用 ffprobe 获取视频时长
-    import subprocess as _subprocess
-
-    from packages.shared.ffmpeg_utils import FFPROBE_BIN
-
-    duration = 30.0  # 默认假设 30 秒
-    try:
-        probe_result = _subprocess.run(
-            [
-                FFPROBE_BIN,
-                "-v",
-                "error",
-                "-show_entries",
-                "format=duration",
-                "-of",
-                "default=noprint_wrappers=1:nokey=1",
-                video_url,
-            ],
-            capture_output=True,
-            text=True,
-            timeout=15,
-        )
-        if probe_result.returncode == 0 and probe_result.stdout.strip():
-            duration = float(probe_result.stdout.strip())
-    except Exception as e:
-        logger.warning("FFprobe 远程视频时长失败，使用默认值: %s", e)
-
-    ratios = [i / (num_frames + 1) for i in range(1, num_frames + 1)]
-    results = []
-
-    for _idx, ratio in enumerate(ratios):
-        frame_time = max(0.5, duration * ratio)
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.close()
-        output_path = tmp.name
-
-        try:
-            seek_str = f"{int(frame_time // 3600):02d}:{int((frame_time % 3600) // 60):02d}:{frame_time % 60:05.2f}"
-            cmd = [
-                FFMPEG_BIN,
-                "-y",
-                "-ss",
-                seek_str,
-                "-i",
-                video_url,
-                "-vframes",
-                "1",
-                "-q:v",
-                "2",
-                output_path,
-            ]
-            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
-
-            if _Path(output_path).exists() and _Path(output_path).stat().st_size > 0:
-                results.append({"local_path": output_path, "frame_time": round(frame_time, 2)})
-            else:
-                _Path(output_path).unlink(missing_ok=True)
-        except Exception as e:
-            logger.warning("FFmpeg 远程抽帧失败 ratio=%.2f: %s", ratio, e)
-            _Path(output_path).unlink(missing_ok=True)
-
-    return results
-
-
 def _call_ai_cover_service(
     plan_id: str,
     asset_ids: List[str],
@@ -450,9 +361,9 @@ def _call_ai_cover_service(
 ) -> Dict[str, Any]:
     """调用 AI 封面生成服务.
 
-    优先级：
-    1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理
-    2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）
+    统一封面管道下，封面已由渲染后视频抽帧生成并持久化到 GenerationTask.cover_url。
+    此函数仅处理 manual/upload 等需要前端交互的类型，
+    ai_frame/ai_regenerate 类型应由调用方直接从持久化的封面 URL 读取。
 
     失败时抛出 RuntimeError。
 
@@ -484,88 +395,14 @@ def _call_ai_cover_service(
             "frame_time": frame_time,
         }
 
-    # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧
-    if primary_video_url:
-        import re as _re
-
-        primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
-
-        # 先检查视频 URL 是否可访问
-        try:
-            head_resp = http_requests.head(primary_video_url, timeout=10, allow_redirects=True)
-            if head_resp.status_code != 200:
-                logger.error(
-                    "封面视频URL不可访问: plan_id=%s url=%s status=%d",
-                    plan_id,
-                    primary_video_url,
-                    head_resp.status_code,
-                )
-                raise RuntimeError(
-                    f"封面生成失败: 预览视频URL不可访问 (HTTP {head_resp.status_code})。" f"请重新生成预览视频后再试。"
-                )
-        except http_requests.RequestException as e:
-            logger.error("封面视频URL连通性检查失败: plan_id=%s url=%s error=%s", plan_id, primary_video_url, e)
-            raise RuntimeError(
-                f"封面生成失败: 无法访问预览视频 ({e.__class__.__name__})。请重新生成预览视频后再试。"
-            ) from e
-
-        # 使用 FFmpeg 从 URL 流式 seek 抽帧
-        try:
-            logger.info("FFmpeg 远程抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
-            frames = _extract_frames_with_ffmpeg(primary_video_url, num_frames=3)
-
-            if frames:
-                best_frame = frames[0]
-                local_path = best_frame["local_path"]
-                frame_time_val = best_frame["frame_time"]
-
-                # 上传到 OSS
-                try:
-                    import uuid
-                    from pathlib import Path
-
-                    from packages.shared.storage import get_shared_storage_service
-
-                    storage = get_shared_storage_service()
-                    cover_key = f"covers/{plan_id}/ffmpeg_frame_{uuid.uuid4().hex[:8]}.jpg"
-                    storage.upload_file(
-                        file_or_path=local_path,
-                        storage_key=cover_key,
-                        content_type="image/jpeg",
-                    )
-                    public_url = storage.get_url(cover_key)
-
-                    logger.info(
-                        "FFmpeg 抽帧成功: plan_id=%s frame_time=%.2f url=%s",
-                        plan_id,
-                        frame_time_val,
-                        public_url[:80],
-                    )
-
-                    return {
-                        "type": "ai_frame",
-                        "image_url": public_url,
-                        "frame_time": round(frame_time_val, 1),
-                        "confidence": 0.85,
-                    }
-                finally:
-                    # 清理所有临时文件
-                    for frame in frames:
-                        try:
-                            Path(frame["local_path"]).unlink(missing_ok=True)
-                        except Exception:
-                            pass
-
-        except RuntimeError:
-            raise
-        except Exception as e:
-            logger.exception("FFmpeg 远程抽帧失败: %s", str(e))
-
-    # 封面生成失败
-    raise RuntimeError(f"封面生成失败: plan_id={plan_id}，无法从视频抽帧。请检查 primary_video_url 是否可访问。")
-
-
-# ── 公共入口 ────────────────────────────────────────────────────────────────
+    # ai_frame / ai_regenerate: 封面应由渲染后视频抽帧管道生成
+    # 如果调用方传入了持久化的封面 URL，直接使用
+    logger.warning(
+        "封面生成回退: plan_id=%s cover_type=%s — 统一管道应已生成封面，请检查 GenerationTask.cover_url",
+        plan_id,
+        cover_type,
+    )
+    raise RuntimeError(f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。")
 
 
 def run_ai_recommend(

tests/unit/test_1294_preview_voice_injection.py

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

tests/unit/test_bool_config_defense.py

@@ -0,0 +1,130 @@
+"""
+测试：模板 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,486 +1,149 @@
-"""Tests for cover frame pre-extraction during rendering.
+"""Tests for unified cover frame extraction pipeline.
 
-Tests:
-- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75%
-- extract_and_upload_cover_frames: extraction + OSS upload
-- RenderAdapterResult.cover_candidates field
-- generation_cover route uses pre-stored candidates
-- ai_service FFmpeg fallback
+统一封面管道测试：
+- extract_first_frame: 从已渲染视频抽取封面帧
+- 封面天然带标题（ASS 字幕已烧录到视频中）
 """
 
 from __future__ import annotations
 
+import tempfile
+import unittest
 from pathlib import Path
-from unittest.mock import MagicMock, Mock, call, patch
-
-import pytest
+from unittest.mock import patch
 
 
-class TestExtractCoverCandidates:
-    """extract_cover_candidates 测试."""
+class TestExtractFirstFrame(unittest.TestCase):
+    """extract_first_frame 单元测试."""
 
     @patch("video_processing.ffmpeg_utils.run_ffmpeg")
-    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
-    def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run):
-        """在 25%/50%/75% 处抽取 3 帧."""
-        import tempfile
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run):
+        """默认在视频 15% 处抽帧."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        # Mock run_ffmpeg 创建输出文件（ffmpeg 真实行为）
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        # Create temp files that look like they were created
-        def fake_run(cmd, **kwargs):
-            # Find the output path (last arg)
-            output_path = cmd[-1]
-            Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
-            return ("", "")
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name)
 
-        mock_run.side_effect = fake_run
-
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(b"fake video")
-            video_path = tmp.name
-
-        try:
-            results = extract_cover_candidates(video_path, num_frames=3)
-            assert len(results) == 3
-
-            # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0
-            assert results[0]["frame_time"] == 5.0
-            assert results[1]["frame_time"] == 10.0
-            assert results[2]["frame_time"] == 15.0
-
-            # Check local paths exist
-            for r in results:
-                assert Path(r["local_path"]).exists()
-
-            # Clean up
-            for r in results:
-                Path(r["local_path"]).unlink(missing_ok=True)
-        finally:
-            Path(video_path).unlink(missing_ok=True)
+            self.assertTrue(Path(result).exists())
+            # 验证 ffmpeg 被调用
+            mock_run.assert_called()
+            cmd = mock_run.call_args[0][0]
+            self.assertIn("-vframes", cmd)
+            self.assertIn("1", cmd)
+            Path(result).unlink(missing_ok=True)
 
     @patch("video_processing.ffmpeg_utils.run_ffmpeg")
-    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
-    def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run):
-        """FFmpeg 失败时跳过该帧，继续抽取其他帧."""
-        import tempfile
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_seek_ratio(self, mock_probe, mock_run):
+        """自定义抽帧位置."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        call_count = 0
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, seek_ratio=0.5)
 
-        def fake_run(cmd, **kwargs):
-            nonlocal call_count
-            call_count += 1
-            output_path = cmd[-1]
-            if call_count == 2:
-                # Second frame fails - don't create file
-                raise RuntimeError("ffmpeg error")
-            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-            return ("", "")
+            self.assertTrue(Path(result).exists())
+            # 50% of 10s = 5s
+            cmd = mock_run.call_args[0][0]
+            ss_idx = cmd.index("-ss") + 1
+            seek_val = cmd[ss_idx]
+            # Should be around 5 seconds
+            self.assertIn("05", seek_val)
+            Path(result).unlink(missing_ok=True)
 
-        mock_run.side_effect = fake_run
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_output_path_parameter(self, mock_probe, mock_run):
+        """指定输出路径."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(b"fake video")
-            video_path = tmp.name
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as out:
+                pass  # just get a path
 
-        try:
-            results = extract_cover_candidates(video_path, num_frames=3)
-            # Should get 2 frames (1st and 3rd), 2nd failed
-            assert len(results) == 2
-        finally:
-            Path(video_path).unlink(missing_ok=True)
-            for r in results:
-                Path(r["local_path"]).unlink(missing_ok=True)
+            # Create the file so ffmpeg "succeeds"
+            mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image")
 
-    @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed"))
-    def test_fallback_duration_when_probe_fails(self, mock_probe):
-        """probe 失败时使用默认时长."""
-        import tempfile
+            result = extract_first_frame(video.name, output_path=out.name)
+            self.assertEqual(result, out.name)
+            Path(out.name).unlink(missing_ok=True)
 
-        from video_processing.thumbnail_generator import extract_cover_candidates
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_keeps_original_resolution_by_default(self, mock_probe, mock_run):
+        """默认保持原始分辨率（width=-1, height=-1）."""
+        from video_processing.thumbnail_generator import extract_first_frame
 
-        # Mock run_ffmpeg to create output files
-        def fake_run(cmd, **kwargs):
-            output_path = cmd[-1]
-            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-            return ("", "")
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
 
-        with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run):
-            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-                tmp.write(b"fake")
-                video_path = tmp.name
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name)
 
-            try:
-                results = extract_cover_candidates(video_path, num_frames=3)
-                assert len(results) == 3
-                # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75
-                assert results[0]["frame_time"] == 1.25
-                assert results[1]["frame_time"] == 2.5
-                assert results[2]["frame_time"] == 3.75
-            finally:
-                Path(video_path).unlink(missing_ok=True)
-                for r in results:
-                    Path(r["local_path"]).unlink(missing_ok=True)
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            # Should NOT have scale filter (only format)
+            self.assertNotIn("scale", vf_filter)
+            self.assertIn("format", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_custom_width_triggers_scale(self, mock_probe, mock_run):
+        """指定宽度时添加 scale 滤镜."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake")
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+            result = extract_first_frame(video.name, width=640)
+
+            cmd = mock_run.call_args[0][0]
+            vf_idx = cmd.index("-vf") + 1
+            vf_filter = cmd[vf_idx]
+            self.assertIn("scale=640", vf_filter)
+            Path(result).unlink(missing_ok=True)
+
+    @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=RuntimeError("fail"))
+    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0)
+    def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run):
+        """失败时清理临时文件."""
+        from video_processing.thumbnail_generator import extract_first_frame
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4") as video:
+            Path(video.name).write_bytes(b"fake video")
+
+            with self.assertRaises(RuntimeError):
+                extract_first_frame(video.name)
 
 
-class TestExtractAndUploadCoverFrames:
-    """extract_and_upload_cover_frames 测试."""
+class TestRenderAdapterCoverUrl(unittest.TestCase):
+    """RenderAdapterResult.cover_url 字段测试."""
 
-    @patch("video_processing.oss_helpers.upload_to_oss")
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
-    def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload):
-        """上传帧到 OSS 并返回正确格式."""
-        import tempfile
+    def test_result_has_cover_url_field(self):
+        """RenderAdapterResult 包含 cover_url 字段."""
+        from video_processing.render_adapter import RenderAdapterResult
 
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
+        result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg")
+        self.assertEqual(result.cover_url, "https://example.com/cover.jpg")
 
-        # Create actual temp files
-        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp1.close()
-        tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp2.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp2.close()
-
-        mock_extract.return_value = [
-            {"local_path": tmp1.name, "frame_time": 5.0},
-            {"local_path": tmp2.name, "frame_time": 10.0},
-        ]
-        mock_upload.side_effect = [
-            "https://oss.example.com/covers/plan1/frame_0.jpg",
-            "https://oss.example.com/covers/plan1/frame_1.jpg",
-        ]
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-
-        assert len(results) == 2
-        assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg"
-        assert results[0]["frame_time"] == 5.0
-        assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg"
-
-        assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg"
-        assert results[1]["frame_time"] == 10.0
-
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[])
-    def test_returns_empty_when_no_candidates(self, mock_extract):
-        """没有候选帧时返回空列表."""
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-        assert results == []
-
-    @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error"))
-    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
-    def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload):
-        """上传失败时跳过该帧."""
-        import tempfile
-
-        from video_processing.thumbnail_generator import extract_and_upload_cover_frames
-
-        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp1.close()
-
-        mock_extract.return_value = [
-            {"local_path": tmp1.name, "frame_time": 5.0},
-        ]
-
-        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
-        assert results == []
-
-
-class TestRenderAdapterResultCoverCandidates:
-    """RenderAdapterResult 的 cover_candidates 字段."""
-
-    def test_default_none(self):
-        """默认为 None."""
+    def test_result_cover_url_defaults_empty(self):
+        """cover_url 默认为空字符串."""
         from video_processing.render_adapter import RenderAdapterResult
 
         result = RenderAdapterResult(success=True)
-        assert result.cover_candidates is None
-
-    def test_can_set_candidates(self):
-        """可以设置候选帧列表."""
-        from video_processing.render_adapter import RenderAdapterResult
-
-        candidates = [
-            {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"},
-        ]
-        result = RenderAdapterResult(success=True, cover_candidates=candidates)
-        assert len(result.cover_candidates) == 1
-        assert result.cover_candidates[0]["frame_time"] == 5.0
+        self.assertEqual(result.cover_url, "")
 
 
-class TestAICoverServiceFFmpegFallback:
-    """AI 封面服务 FFmpeg 兜底测试."""
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head):
-        """FFmpeg 兜底抽帧成功."""
-        import tempfile
-
-        mock_head.return_value.status_code = 200
-
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp.close()
-
-        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}]
-
-        # Mock storage
-        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
-            mock_storage = Mock()
-            mock_storage.upload_file = Mock()
-            mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
-            mock_storage_fn.return_value = mock_storage
-
-            from packages.shared.ai_service import _call_ai_cover_service
-
-            result = _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-            assert result["type"] == "ai_frame"
-            assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
-            assert result["frame_time"] == 5.0
-            assert result["confidence"] == 0.85
-
-        Path(tmp.name).unlink(missing_ok=True)
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    def test_ffmpeg_no_video_url_raises(self, mock_head):
-        """没有视频 URL 时抛出 RuntimeError."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url=None,
-            )
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 抽帧为空时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-    def test_upload_type_returns_immediately(self):
-        """upload 类型直接返回."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        result = _call_ai_cover_service(
-            plan_id="plan1",
-            asset_ids=["a1"],
-            cover_type="upload",
-            primary_video_url="https://example.com/video.mp4",
-        )
-        assert result["type"] == "upload"
-
-    def test_manual_type_returns_immediately(self):
-        """manual 类型直接返回."""
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        result = _call_ai_cover_service(
-            plan_id="plan1",
-            asset_ids=["a1"],
-            cover_type="manual",
-            frame_time=5.0,
-            primary_video_url="https://example.com/video.mp4",
-        )
-        assert result["type"] == "manual"
-        assert result["frame_time"] == 5.0
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head):
-        """视频 URL 不可访问时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 404
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
-            _call_ai_cover_service(
-                plan_id="plan1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-
-class TestCoverTemplatesFix:
-    """CoverTemplateResponse config=None 修复测试."""
-
-    def test_config_none_becomes_empty_dict(self):
-        """config=None 时 CoverTemplateResponse 不报 ValidationError."""
-        from datetime import datetime
-
-        from app.schemas.cover_template import CoverTemplateResponse
-
-        # This should not raise
-        resp = CoverTemplateResponse(
-            id="1",
-            name="test",
-            thumbnail_url="",
-            is_system=True,
-            created_at=datetime.now(),
-            config={},
-        )
-        assert resp.config == {}
-
-
-class TestExtractFramesWithFFmpeg:
-    """_extract_frames_with_ffmpeg 单元测试."""
-
-    def test_extracts_frames_with_correct_seek_times(self):
-        """抽帧时间点正确计算."""
-        import subprocess
-        import tempfile
-
-        # Mock ffprobe to return duration
-        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="")
-        with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc:
-            # First call is ffprobe, rest are ffmpeg
-            call_count = 0
-
-            def side_effect(cmd, **kwargs):
-                nonlocal call_count
-                call_count += 1
-                if call_count == 1:
-                    # ffprobe call
-                    return mock_probe_result
-                else:
-                    # ffmpeg call - create output file
-                    output_path = cmd[-1]
-                    from pathlib import Path
-
-                    Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-                    return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
-
-            mock_subproc.side_effect = side_effect
-
-            from packages.shared.ai_service import _extract_frames_with_ffmpeg
-
-            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
-
-            assert len(results) == 3
-            # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0
-            assert results[0]["frame_time"] == 5.0
-            assert results[1]["frame_time"] == 10.0
-            assert results[2]["frame_time"] == 15.0
-
-            # Clean up
-            for r in results:
-                from pathlib import Path
-
-                Path(r["local_path"]).unlink(missing_ok=True)
-
-    def test_handles_ffmpeg_failure(self):
-        """FFmpeg 失败时跳过该帧."""
-        import subprocess
-        import tempfile
-        from pathlib import Path
-
-        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="")
-
-        call_count = 0
-
-        def side_effect(cmd, **kwargs):
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                return mock_probe_result
-            output_path = cmd[-1]
-            if call_count == 2:
-                # First frame succeeds
-                Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
-                return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
-            else:
-                # Other frames fail
-                raise subprocess.CalledProcessError(1, cmd)
-
-        with patch("subprocess.run", side_effect=side_effect):
-            from packages.shared.ai_service import _extract_frames_with_ffmpeg
-
-            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
-            assert len(results) == 1
-            Path(results[0]["local_path"]).unlink(missing_ok=True)
-
-
-class TestGenerationCoverPreStored:
-    """generation_cover.py 预存帧逻辑测试."""
-
-    def test_pre_stored_candidates_used_when_available(self):
-        """有预存帧时直接使用，不调用 AI 服务."""
-        from unittest.mock import patch
-
-        # Mock the dependencies
-        mock_plan = MagicMock()
-        mock_plan.config = {
-            "cover_candidates": [
-                {
-                    "image_url": "https://oss.example.com/covers/p1/frame_0.jpg",
-                    "frame_time": 5.0,
-                    "storage_key": "covers/p1/frame_0.jpg",
-                },
-                {
-                    "image_url": "https://oss.example.com/covers/p1/frame_1.jpg",
-                    "frame_time": 10.0,
-                    "storage_key": "covers/p1/frame_1.jpg",
-                },
-            ],
-            "rendered_storage_key": "rendered/p1/video.mp4",
-        }
-
-        mock_plan_svc = MagicMock()
-        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
-
-        mock_body = MagicMock()
-        mock_body.asset_ids = ["a1"]
-        mock_body.cover_type = "ai_frame"
-        mock_body.frame_time = None
-
-        with (
-            patch("app.api.routes.generation_cover.get_editor_services") as mock_services,
-            patch("app.api.routes.generation_cover.get_db_session"),
-            patch("app.api.routes.generation_cover.get_current_user"),
-            patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"),
-            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
-        ):
-
-            mock_services.return_value = (MagicMock(), mock_plan_svc)
-            mock_normalize.side_effect = lambda c: c
-
-            from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover
-
-            result = generate_cover(
-                body=mock_body,
-                template_id="t1",
-                plan_id="p1",
-                services=(MagicMock(), mock_plan_svc),
-                db=MagicMock(),
-                current_user=MagicMock(),
-            )
-
-            assert result.plan_id == "p1"
-            assert result.cover["type"] == "ai_frame"
-            assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg"
-            assert result.cover["frame_time"] == 5.0
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_generated_video_creation_logic.py

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

tests/unit/test_generation_cover.py

@@ -92,3 +92,463 @@ 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,93 +141,13 @@ class TestMediaKitClient:
 
 
 class TestAICoverService:
-    """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。"""
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_with_ffmpeg_success(self, mock_ffmpeg, mock_head):
-        """FFmpeg 本地抽帧成功."""
-        import tempfile
-
-        mock_head.return_value.status_code = 200
-
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
-        tmp.write(b"\xff\xd8" + b"\x00" * 50)
-        tmp.close()
-
-        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 3.5}]
-
-        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
-            mock_storage = Mock()
-            mock_storage.upload_file = Mock()
-            mock_storage.get_url.return_value = "https://example.com/frame.jpg"
-            mock_storage_fn.return_value = mock_storage
-
-            from packages.shared.ai_service import _call_ai_cover_service
-
-            result = _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
-            assert result["type"] == "ai_frame"
-            assert result["image_url"] == "https://example.com/frame.jpg"
-            assert result["frame_time"] == 3.5
-            assert result["confidence"] == 0.85
-
-        Path(tmp.name).unlink(missing_ok=True)
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    def test_call_ai_cover_video_url_unreachable(self, mock_head):
-        """视频 URL 不可访问时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 404
+    """AI 封面服务测试（统一封面管道后）。"""
 
+    def test_call_ai_cover_ai_frame_raises(self):
+        """ai_frame type raises RuntimeError in unified pipeline."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/nonexistent.mp4",
-            )
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_url_double_slash_normalized(self, mock_ffmpeg, mock_head):
-        """URL 路径中的双斜杠应被规范化."""
-        dirty_url = "https://oss.example.com/generated/projects//tasks/abc123/rendered.mp4"
-        clean_url = "https://oss.example.com/generated/projects/tasks/abc123/rendered.mp4"
-
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError):
-            _call_ai_cover_service(
-                plan_id="plan-1",
-                asset_ids=["a1"],
-                cover_type="ai_frame",
-                primary_video_url=dirty_url,
-            )
-
-        # HEAD 请求使用规范化后的 URL
-        mock_head.assert_called_once()
-        assert mock_head.call_args[0][0] == clean_url
-
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_ffmpeg_failure_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 抽帧失败时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.side_effect = Exception("ffmpeg error")
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -235,11 +155,22 @@ class TestAICoverService:
                 primary_video_url="https://example.com/video.mp4",
             )
 
+    def test_call_ai_cover_ai_regenerate_raises(self):
+        """ai_regenerate type raises RuntimeError in unified pipeline."""
+        from packages.shared.ai_service import _call_ai_cover_service
+
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
+            _call_ai_cover_service(
+                plan_id="plan-123",
+                asset_ids=["asset-1"],
+                cover_type="ai_regenerate",
+            )
+
     def test_call_ai_cover_without_video_url_raises(self):
-        """没有视频 URL 时抛出 RuntimeError."""
+        """ai_frame without video URL still raises RuntimeError."""
         from packages.shared.ai_service import _call_ai_cover_service
 
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service(
                 plan_id="plan-123",
                 asset_ids=["asset-1"],
@@ -276,23 +207,6 @@ class TestAICoverService:
         assert result["type"] == "manual"
         assert result["frame_time"] == 5.0
 
-    @patch("packages.shared.ai_service.http_requests.head")
-    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
-    def test_call_ai_cover_empty_frames_raises(self, mock_ffmpeg, mock_head):
-        """FFmpeg 返回空帧列表时抛出 RuntimeError."""
-        mock_head.return_value.status_code = 200
-        mock_ffmpeg.return_value = []
-
-        from packages.shared.ai_service import _call_ai_cover_service
-
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
-            _call_ai_cover_service(
-                plan_id="plan-123",
-                asset_ids=["asset-1"],
-                cover_type="ai_frame",
-                primary_video_url="https://example.com/video.mp4",
-            )
-
 
 class TestGenerateCover:
     """run_generate_cover 测试."""

tests/unit/test_overlay_title_on_ass.py

@@ -0,0 +1,242 @@
+"""测试 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

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

tests/unit/test_shared_ai_service.py

@@ -436,12 +436,12 @@ class TestAiCoverService:
 
     def test_cover_type_ai_frame_raises_without_mediakit(self):
         """ai_frame mode raises RuntimeError when MediaKit is unavailable."""
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service("plan1", ["a1"], "ai_frame")
 
     def test_cover_type_ai_regenerate_raises_without_mediakit(self):
         """ai_regenerate mode raises RuntimeError when MediaKit is unavailable."""
-        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
+        with pytest.raises(RuntimeError, match="封面数据不可用"):
             _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
 
     def test_cover_type_manual_still_works(self):