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

@@ -47,6 +47,10 @@ class GenerateCoverRequest(BaseModel):
         ge=0.0,
         description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
     )
+    title_text: Optional[str] = Field(
+        default=None,
+        description="用户选择的标题文字，叠加到封面图上",
+    )
 
 
 class GenerateCoverResponse(BaseModel):
@@ -59,6 +63,72 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
+
+# ── 封面标题叠加工具函数 ──────────────────────────────────────────────
+
+
+def _apply_title_to_cover_image(
+    image_url: str,
+    title_text: str,
+    plan_id: str,
+) -> str:
+    """下载封面图、叠加标题文字、重新上传，返回新 URL。
+
+    Args:
+        image_url: 原始封面图片 URL
+        title_text: 要叠加的标题文字
+        plan_id: 用于日志和 OSS 路径
+
+    Returns:
+        叠加标题后的新封面图片 URL，失败时返回原 URL
+    """
+    import os as _os
+    import tempfile as _tempfile
+    from pathlib import Path as _Path
+
+    if not title_text or not title_text.strip():
+        return image_url
+
+    # 下载原图
+    import urllib.request as _urllib_req
+    tmp_dir = _tempfile.mkdtemp(prefix=f"cover_title_{plan_id[:8]}_")
+    src_path = _os.path.join(tmp_dir, "cover_src.jpg")
+    try:
+        _urllib_req.urlretrieve(image_url, src_path)
+    except Exception:
+        logger.warning("[封面标题] 下载封面图失败: url=%s", image_url[:80])
+        return image_url
+
+    # 调用 thumbnail_generator 的 _overlay_title_on_image
+    try:
+        from video_processing.thumbnail_generator import _overlay_title_on_image
+        _overlay_title_on_image(src_path, title_text)
+    except Exception:
+        logger.warning("[封面标题] FFmpeg drawtext 失败: plan_id=%s", plan_id, exc_info=True)
+        return image_url
+
+    # 重新上传到 OSS
+    try:
+        from packages.shared.storage import get_shared_storage_service
+        storage_svc = get_shared_storage_service()
+        storage_key = f"covers/{plan_id}/titled_cover.jpg"
+        new_url = storage_svc.upload_file_smart(src_path, storage_key)
+        if new_url:
+            return new_url
+        logger.warning("[封面标题] OSS 上传返回空: plan_id=%s", plan_id)
+        return image_url
+    except Exception:
+        logger.warning("[封面标题] OSS 上传失败: plan_id=%s", plan_id, exc_info=True)
+        return image_url
+    finally:
+        # 清理临时文件
+        try:
+            _Path(src_path).unlink(missing_ok=True)
+            _os.rmdir(tmp_dir)
+        except Exception:
+            pass
+
+
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
 def generate_cover(
     body: GenerateCoverRequest,
@@ -201,6 +271,24 @@ def generate_cover(
             "confidence": 0.9,
         }
         if cover_data["image_url"]:
+            # 如果用户指定了标题且候选帧未叠加标题，则叠加标题到封面图上
+            title_text = (body.title_text or "").strip()
+            if title_text:
+                try:
+                    cover_data["image_url"] = _apply_title_to_cover_image(
+                        cover_data["image_url"], title_text, plan_id
+                    )
+                    logger.info(
+                        "[封面生成] 标题已叠加到封面候选帧: plan_id=%s title=%s",
+                        plan_id,
+                        title_text[:30],
+                    )
+                except Exception:
+                    logger.warning(
+                        "[封面生成] 标题叠加失败（使用原图）: plan_id=%s",
+                        plan_id,
+                        exc_info=True,
+                    )
             current_config = dict(plan.config) if plan.config else {}
             current_config["cover"] = cover_data
             normalized = normalize_plan_config(current_config)
@@ -221,6 +309,25 @@ def generate_cover(
     except RuntimeError as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
+    # 如果用户指定了标题，叠加到 AI 生成的封面图上
+    title_text = (body.title_text or "").strip()
+    if title_text and cover_data.get("image_url"):
+        try:
+            cover_data["image_url"] = _apply_title_to_cover_image(
+                cover_data["image_url"], title_text, plan_id
+            )
+            logger.info(
+                "[封面生成] 标题已叠加到 AI 封面: plan_id=%s title=%s",
+                plan_id,
+                title_text[:30],
+            )
+        except Exception:
+            logger.warning(
+                "[封面生成] AI 封面标题叠加失败（使用原图）: plan_id=%s",
+                plan_id,
+                exc_info=True,
+            )
+
     current_config = dict(plan.config) if plan.config else {}
     current_config["cover"] = cover_data
     normalized = normalize_plan_config(current_config)

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

@@ -4,6 +4,8 @@ export interface GenerateCoverRequest {
   asset_ids: string[]
   cover_type?: "ai_frame" | "manual" | "upload" | "ai_regenerate"
   frame_time?: number
+  /** 用户选择的标题文字，叠加到封面图上 */
+  title_text?: string
 }
 
 export interface GenerateCoverResponse {

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

@@ -190,6 +190,7 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
           duration={duration}
           assetIds={materialMode === "auto" ? smartSelectedIds : selectedMaterials}
           selectedTemplate={selectedTemplate}
+          titleText={titleSettings.title}
         />
       )
     case 7:

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

@@ -14,6 +14,8 @@ interface Step6CoverSettingsProps {
   assetIds?: string[]
   /** 当前选中的模板 ID */
   selectedTemplate?: string
+  /** Step 5 用户选择的标题文字，叠加到封面图上 */
+  titleText?: string
 }
 
 const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
@@ -40,6 +42,7 @@ const Step6CoverSettings: React.FC<Step6CoverSettingsProps> = (props) => {
     duration: props.duration,
     assetIds: props.assetIds,
     selectedTemplate: props.selectedTemplate,
+    titleText: props.titleText,
   })
 
   const handleAutoGenerate = () => {

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

@@ -21,6 +21,8 @@ interface UseStep6CoverProps {
   assetIds?: string[]
   /** 当前选中的模板 ID */
   selectedTemplate?: string
+  /** Step 5 用户选择的标题文字，叠加到封面图上 */
+  titleText?: string
 }
 
 export function useStep6Cover({
@@ -29,6 +31,7 @@ export function useStep6Cover({
   duration,
   assetIds = [],
   selectedTemplate = "",
+  titleText = "",
 }: UseStep6CoverProps) {
   const [generating, setGenerating] = useState(false)
 
@@ -91,6 +94,7 @@ export function useStep6Cover({
       const response = await generateCover(selectedTemplate, {
         asset_ids: assetIds,
         cover_type: "ai_frame",
+        title_text: titleText || undefined,
       })
       clearTimeout(timeoutId)
       const thumbnailUrl = response.cover?.image_url || ""
@@ -133,7 +137,7 @@ export function useStep6Cover({
       clearTimeout(timeoutId)
       setGenerating(false)
     }
-  }, [selectedTemplate, assetIds, coverSettings, onCoverSettingsChange, generating])
+  }, [selectedTemplate, assetIds, coverSettings, onCoverSettingsChange, generating, titleText])
 
   // ── 模板操作方法 ──
   const handleSelectTemplate = useCallback((id: string) => {

apps/worker/worker_app/tasks/generation.py

@@ -1123,6 +1123,7 @@ def _render_video(
     resolution: str = "",
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
+    custom_title: str = "",
 ) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
 
@@ -1155,6 +1156,21 @@ def _render_video(
                 list(template_config.keys()),
             )
 
+    # 注入用户自定义标题（custom_title 来自 Step 5 用户选择的视频标题）
+    # 覆盖模板的 title.text，确保渲染引擎和视频封面都使用用户选择的标题
+    if custom_title and custom_title.strip():
+        plan_cfg = dict(virtual_plan.config or {})
+        title_cfg = dict(plan_cfg.get("title", {}) or {})
+        title_cfg["text"] = custom_title.strip()
+        title_cfg["enabled"] = True
+        plan_cfg["title"] = title_cfg
+        virtual_plan.config = plan_cfg
+        logger.info(
+            "[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
+            task_id,
+            custom_title.strip()[:50],
+        )
+
     # 用户自定义 BGM 覆盖模板 BGM（用户指定优先级最高）
     if bgm_config:
         plan_cfg = virtual_plan.config or {}
@@ -1529,6 +1545,7 @@ 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: