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

@@ -63,8 +63,15 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
-def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
-    """下载 MediaKit 返回的临时帧图并转存到 OSS covers/ 路径。"""
+def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> str:
+    """下载 MediaKit 返回的临时帧图，可选叠加标题后转存到 OSS covers/ 路径。
+
+    Args:
+        frame_url: MediaKit 返回的临时帧图 URL
+        plan_id: 剪辑计划 ID（生成 OSS key）
+        title_text: 非空时用 Pillow 在帧上叠加标题（用于 E2 从源素材抽帧，
+            因为源素材本身没有烧录标题）
+    """
     import tempfile
     import uuid
     from pathlib import Path
@@ -82,6 +89,21 @@ def _persist_cover_frame(frame_url: str, plan_id: str) -> str:
             tmp.write(resp.content)
             tmp_path = tmp.name
 
+        # E2 从源素材抽帧时，源素材无标题，叠加标题文字
+        if title_text and title_text.strip():
+            try:
+                from packages.shared.title_overlay import apply_title_to_image
+
+                applied = apply_title_to_image(tmp_path, title_text)
+                if applied:
+                    logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id)
+            except Exception:
+                logger.warning(
+                    "[封面生成] E2 标题叠加失败（返回无标题帧）: plan_id=%s",
+                    plan_id,
+                    exc_info=True,
+                )
+
         from packages.shared.storage import get_shared_storage_service
 
         storage = get_shared_storage_service()
@@ -396,6 +418,11 @@ def generate_cover(
             asset_repo = SQLAlchemyAssetRepository(db)
             storage_svc = get_shared_storage_service()
             mk_client = get_mediakit_client()
+            # 从 plan.config 读取标题，E2 从源素材抽帧时叠加（源素材本身无标题）
+            _e2_title_cfg = (plan.config or {}).get("title", {}) or {}
+            if not isinstance(_e2_title_cfg, dict):
+                _e2_title_cfg = {}
+            _e2_title_text = (_e2_title_cfg.get("text", "") or "").strip() if _e2_title_cfg.get("enabled", True) else ""
             if mk_client.is_available:
                 for aid in body.asset_ids:
                     try:
@@ -425,7 +452,7 @@ def generate_cover(
                         if snapshots:
                             raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
                             if raw:
-                                cover_url_from_task = _persist_cover_frame(raw, plan_id)
+                                cover_url_from_task = _persist_cover_frame(raw, plan_id, title_text=_e2_title_text)
                                 logger.info(
                                     "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s",
                                     plan_id,

apps/worker/video_processing/render_adapter.py

@@ -585,14 +585,11 @@ class RenderAdapter:
         try:
             from video_processing.thumbnail_generator import extract_and_upload_cover_frames
 
-            # 从 plan config 提取标题文字，叠加到封面候选帧上
-            _title_cfg = (plan_config or {}).get("title", {}) or {}
-            if not isinstance(_title_cfg, dict):
-                _title_cfg = {}
-            _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else ""
-
+            # 已渲染视频在统一渲染阶段已通过 ASS 字幕把标题烧录进画面，
+            # 抽帧天然带标题，因此这里传空字符串，避免 Pillow 二次叠加导致重影。
+            # Pillow 叠加仅用于 API 从源素材抽帧（源素材本身无标题）的兜底场景。
             cover_candidates = extract_and_upload_cover_frames(
-                str(result.output_path), plan_id, num_frames=3, title_text=_title_text
+                str(result.output_path), plan_id, num_frames=3, title_text=""
             )
             if cover_candidates:
                 logger.info(

apps/worker/video_processing/thumbnail_generator.py

@@ -1,7 +1,8 @@
-"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。
+"""视频封面抽帧工具 — 从视频中抽取帧作为封面，支持标题文字叠加。
 
-统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
-渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
+统一封面管道：
+- 从已渲染视频抽帧：标题已通过 ASS 字幕烧进视频，帧天然带标题，无需再叠加。
+- 从源素材抽帧（API E2 兜底）：源素材无标题，通过 Pillow 在帧上绘制标题文字。
 """
 
 from __future__ import annotations
@@ -12,6 +13,38 @@ from pathlib import Path
 
 logger = logging.getLogger(__name__)
 
+# ── 标题叠加（Pillow）──────────────────────────────────────────────────────
+# 实现统一放在 packages/shared/title_overlay.py，API 和 Worker 共用。
+
+
+def apply_title_overlay(
+    image_path: str,
+    title_text: str,
+    *,
+    position: str = "bottom",
+    font_size: int | None = None,
+    margin_ratio: float = 0.06,
+    stroke_width_ratio: float = 0.04,
+) -> str:
+    """在图片上绘制标题文字（白色 + 黑色描边/阴影）。
+
+    委托给 packages.shared.title_overlay.apply_title_to_image，
+    保持 Worker 内调用方式不变。title_text 为空时直接返回原路径。
+    """
+    from packages.shared.title_overlay import apply_title_to_image
+
+    if not title_text or not title_text.strip():
+        return image_path
+    result = apply_title_to_image(
+        image_path,
+        title_text,
+        position=position,
+        font_size=font_size,
+        margin_ratio=margin_ratio,
+        stroke_width_ratio=stroke_width_ratio,
+    )
+    return result or image_path
+
 
 def extract_first_frame(
     video_path: str,
@@ -182,7 +215,8 @@ def extract_and_upload_cover_frames(
         video_path: 视频文件路径
         plan_id: 编辑计划 ID（用于生成 storage key）
         num_frames: 抽取帧数（默认 3）
-        title_text: 标题文字（当前版本未叠加，预留参数）
+        title_text: 标题文字；非空时用 Pillow 叠加到每帧（白色 + 黑色描边）。
+            从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。
 
     Returns:
         封面候选列表，每项包含 {"url": str, "position": float}
@@ -208,6 +242,9 @@ def extract_and_upload_cover_frames(
                 seek_ratio=ratio,
                 min_seek_seconds=0.5,
             )
+            # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过
+            if title_text and title_text.strip():
+                apply_title_overlay(frame_path, title_text)
             storage_key = f"covers/{plan_id}/frame_{i}.jpg"
             url = upload_to_oss(frame_path, storage_key)
             if url:

apps/worker/worker_app/tasks/generation.py

@@ -967,14 +967,14 @@ def _render_video(
     bgm_config: dict | None = None,
     voice_ids: list[str] | None = None,
     custom_title: str = "",
-) -> tuple[Path, float]:
+) -> tuple[Path, float, list[dict] | None]:
     """渲染视频（含配音混音）。
 
     使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
 
     Args:
     Returns:
-        (output_path, render_duration)
+        (output_path, render_duration, cover_candidates)
     """
     if not downloaded_videos:
         raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1113,8 +1113,10 @@ def _render_video(
 
     # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
     output_path = render_output_path
+    # RenderAdapter 在渲染完成后用本地 ffmpeg 抽取的封面候选帧（已上传 OSS）
+    cover_candidates = getattr(render_result, "cover_candidates", None)
 
-    return output_path, render_duration
+    return output_path, render_duration, cover_candidates
 
 
 def _upload_and_record(
@@ -1385,7 +1387,7 @@ def generate_video(self, task_id: str) -> dict:
             else:
                 _resolved_resolution = task_info.get("resolution", "")
 
-            output_path, render_duration = _render_video(
+            output_path, render_duration, cover_candidates = _render_video(
                 task_id=task_id,
                 downloaded_videos=downloaded_videos,
                 voice_path=audio_path,
@@ -1430,51 +1432,47 @@ def generate_video(self, task_id: str) -> dict:
 
             _update_task_progress(task_id, 95, "上传完成")
 
-            # ── 4.5 封面抽帧 ────────────────────────────────────────────────
-            # 预览视频上传完成后，提取封面帧写入 gen_task.cover_url
-            # 这样封面路由（generation_cover.py 步骤A）可以通过 generation_task_id 直接找到
+            # ── 4.5 封面帧持久化 ────────────────────────────────────────────
+            # RenderAdapter 在渲染完成后已用本地 ffmpeg 从 output_path 抽帧
+            # （标题通过 ASS 烧录，帧天然带标题），并上传 OSS 返回 cover_candidates。
+            # 这里把第一帧写入 gen_task.cover_url，完整列表写入 metadata，
+            # 封面路由（generation_cover.py）的 A/B/C/D 步骤即可直接命中。
             try:
-                from packages.shared.mediakit_client import get_mediakit_client
+                if cover_candidates:
+                    first = cover_candidates[0]
+                    # 候选帧字段兼容：RenderAdapter 用 image_url，thumbnail_generator 用 url
+                    cover_frame_url = first.get("image_url") or first.get("url") or ""
+                    if cover_frame_url:
+                        _cover_session = SessionLocal()
+                        try:
+                            from packages.adapters.sqlalchemy_impl.models import (
+                                GenerationTaskModel,
+                            )
 
-                mk_client = get_mediakit_client()
-                if mk_client.is_available:
-                    _update_task_progress(task_id, 96, "提取封面帧")
-                    snapshots = mk_client.extract_frames(
-                        video_url=file_url,
-                        strategy="SpecifiedFrames",
-                        max_frames=1,
-                    )
-                    if snapshots and len(snapshots) > 0:
-                        cover_frame_url = snapshots[0].get("image_url", "")
-                        if cover_frame_url and gen_task:
-                            # 通过独立 session 持久化 cover_url
-                            _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_frame_url
+                                # 持久化完整候选列表到 metadata
+                                meta = dict(_cover_model.metadata or {})
+                                meta["cover_candidates"] = cover_candidates
+                                _cover_model.metadata = meta
+                                _cover_session.commit()
+                                logger.info(
+                                    "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d",
+                                    task_id,
+                                    cover_frame_url[:80],
+                                    len(cover_candidates),
                                 )
-
-                                _cover_model = (
-                                    _cover_session.query(GenerationTaskModel)
-                                    .filter(GenerationTaskModel.id == task_id)
-                                    .first()
-                                )
-                                if _cover_model:
-                                    _cover_model.cover_url = cover_frame_url
-                                    _cover_session.commit()
-                                    logger.info(
-                                        "[task_id=%s] 封面帧提取成功: %s",
-                                        task_id,
-                                        cover_frame_url[:80],
-                                    )
-                            finally:
-                                _cover_session.close()
-                    else:
-                        logger.warning("[task_id=%s] 封面帧提取返回空结果", task_id)
+                        finally:
+                            _cover_session.close()
                 else:
-                    logger.warning("[task_id=%s] MediaKit 未配置，跳过封面帧提取", task_id)
+                    logger.warning("[task_id=%s] 渲染未产出 cover_candidates，封面将依赖 API 兜底", task_id)
             except Exception:
-                logger.warning("[task_id=%s] 封面帧提取失败（不影响主流程）", task_id, exc_info=True)
+                logger.warning("[task_id=%s] 封面帧持久化失败（不影响主流程）", task_id, exc_info=True)
 
             # ── 5. 标记完成 ──────────────────────────────────────────────────
             _update_task_status(task_id, "mark_completed", result_count=video_count)

infra/docker/api-base.Dockerfile

@@ -17,6 +17,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     libpq-dev \
     libpq5 \
     ffmpeg \
+    fonts-noto-cjk \
+    fontconfig \
     && rm -rf /var/lib/apt/lists/*
 
 # 创建虚拟环境

packages/shared/title_overlay.py

@@ -0,0 +1,153 @@
+"""封面标题文字叠加（Pillow）— API / Worker 共用。
+
+在封面帧上绘制白色标题文字 + 黑色描边/阴影，支持 CJK 字体和自动换行。
+从已渲染视频抽帧时通常不需要调用（标题已烧录）；
+从源素材抽帧（API E2 兜底）时调用，保证封面带标题。
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# 按优先级查找 CJK 字体（Debian/Ubuntu fonts-noto-cjk 安装路径）
+_FONT_CANDIDATES = (
+    "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
+    "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
+    "/usr/share/fonts/truetype/noto/NotoSansCJK-Bold.ttc",
+    "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
+    "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc",
+)
+
+
+def find_title_font(size: int):
+    """查找可用的 CJK 字体并返回 PIL ImageFont，找不到返回 None。"""
+    try:
+        from PIL import ImageFont
+    except ImportError:
+        return None
+
+    for fp in _FONT_CANDIDATES:
+        if Path(fp).exists():
+            try:
+                return ImageFont.truetype(fp, size=size)
+            except Exception:
+                continue
+    logger.warning("未找到 CJK 字体，标题叠加将使用 PIL 默认字体（中文可能显示为方块）")
+    return ImageFont.load_default()
+
+
+def wrap_title_text(text: str, font, max_width: int) -> list[str]:
+    """按像素宽度对中英文混合文本自动换行，支持显式 \\n。"""
+    lines: list[str] = []
+    current = ""
+    for ch in text:
+        if ch == "\n":
+            if current:
+                lines.append(current)
+            current = ""
+            continue
+        trial = current + ch
+        try:
+            bbox = font.getbbox(trial)
+            width = bbox[2] - bbox[0]
+        except Exception:
+            width = len(trial) * (font.size // 2)
+        if width <= max_width:
+            current = trial
+        else:
+            if current:
+                lines.append(current)
+            current = ch
+    if current:
+        lines.append(current)
+    return lines
+
+
+def apply_title_to_image(
+    image_path: str,
+    title_text: str,
+    *,
+    position: str = "bottom",
+    font_size: Optional[int] = None,
+    margin_ratio: float = 0.06,
+    stroke_width_ratio: float = 0.04,
+) -> Optional[str]:
+    """在图片上绘制标题文字并覆盖保存。
+
+    Args:
+        image_path: 图片路径（处理结果覆盖写回）
+        title_text: 标题文字；为空直接返回 None 表示跳过
+        position: top / center / bottom
+        font_size: 字号，None 时按图片宽度自动计算
+        margin_ratio: 边缘留白占短边比例
+        stroke_width_ratio: 描边宽度占字号比例
+
+    Returns:
+        成功返回 image_path；标题为空或 PIL 不可用返回 None。
+    """
+    if not title_text or not title_text.strip():
+        return None
+
+    try:
+        from PIL import Image, ImageDraw
+    except ImportError:
+        logger.warning("Pillow 未安装，跳过标题叠加: image=%s", image_path)
+        return None
+
+    img = Image.open(image_path).convert("RGB")
+    draw = ImageDraw.Draw(img)
+    img_w, img_h = img.size
+
+    if font_size is None:
+        font_size = max(28, min(72, img_w // 16))
+
+    font = find_title_font(font_size)
+    if font is None:
+        return None
+
+    stroke_width = max(2, int(font_size * stroke_width_ratio))
+    margin = int(min(img_w, img_h) * margin_ratio)
+    max_text_width = img_w - 2 * margin
+
+    lines = wrap_title_text(title_text.strip(), font, max_text_width)
+    if not lines:
+        return None
+
+    line_heights = []
+    for ln in lines:
+        bbox = font.getbbox(ln)
+        line_heights.append(bbox[3] - bbox[1])
+    line_height = max(line_heights) if line_heights else font_size
+    line_gap = int(line_height * 0.3)
+    total_height = len(lines) * line_height + (len(lines) - 1) * line_gap
+
+    if position == "top":
+        y_start = margin
+    elif position == "center":
+        y_start = (img_h - total_height) // 2
+    else:
+        y_start = img_h - total_height - margin
+
+    for i, ln in enumerate(lines):
+        bbox = font.getbbox(ln)
+        line_w = bbox[2] - bbox[0]
+        x = (img_w - line_w) // 2
+        y = y_start + i * (line_height + line_gap)
+        # 阴影
+        draw.text((x + 2, y + 2), ln, font=font, fill=(0, 0, 0))
+        # 白色文字 + 黑色描边
+        draw.text(
+            (x, y),
+            ln,
+            font=font,
+            fill=(255, 255, 255),
+            stroke_width=stroke_width,
+            stroke_fill=(0, 0, 0),
+        )
+
+    img.save(image_path, "JPEG", quality=92)
+    return image_path

requirements-base.txt

@@ -29,3 +29,4 @@ httpx==0.27.2
 
 # Prometheus monitoring
 prometheus-client==0.21.1
+Pillow==10.4.0

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_candidates = _render_video(
                 task_id="test_task_123",
                 downloaded_videos=[Path("/tmp/video1.mp4")],
                 voice_path=None,

tests/unit/test_generation_cover.py

@@ -920,6 +920,73 @@ class TestUploadCoverType:
             call_kwargs = mock_mk.extract_frames.call_args.kwargs
             assert "source-clip.mp4" in call_kwargs["video_url"]
 
+    def test_e2_passes_plan_title_to_persist_for_overlay(self):
+        """步骤E2：plan.config.title.text 存在时，作为 title_text 传给 _persist_cover_frame 叠加标题。"""
+        from unittest.mock import MagicMock, patch
+
+        from app.api.routes.generation_cover import GenerateCoverRequest
+
+        mock_plan = MagicMock()
+        mock_plan.config = {"title": {"enabled": True, "text": "我的视频标题"}}
+
+        mock_plan_svc = MagicMock()
+        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
+        mock_template_svc = MagicMock()
+        mock_db = MagicMock()
+
+        mock_asset = MagicMock()
+        mock_asset.file_type = "video"
+        mock_asset.storage_key = "uploads/src.mp4"
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get.return_value = mock_asset
+
+        mock_mk = MagicMock()
+        mock_mk.is_available = True
+        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]
+
+        mock_storage = MagicMock()
+        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"
+
+        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch(
+                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
+                return_value=mock_asset_repo,
+            ),
+            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
+            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
+            patch(
+                "app.api.routes.generation_cover._persist_cover_frame",
+                return_value="https://oss.example.com/covers/final.jpg",
+            ) as mock_persist,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+        ):
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            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_normalize.return_value = {
+                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/final.jpg"}
+            }
+
+            from app.api.routes.generation_cover import generate_cover
+
+            result = generate_cover(
+                body=body,
+                template_id="tpl",
+                plan_id="plan-title",
+                services=(mock_template_svc, mock_plan_svc),
+                db=mock_db,
+                current_user=MagicMock(),
+            )
+
+            assert result.cover["image_url"] == "https://oss.example.com/covers/final.jpg"
+            # 标题文字必须透传给持久化函数（用于源素材帧叠加标题）
+            assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题"
+
     def test_step_e_skips_non_video_assets(self):
         """步骤E2：asset_ids 里只有图片素材时，不调用 MediaKit 并返回 400。"""
         from unittest.mock import MagicMock, patch

tests/unit/test_title_overlay.py

@@ -0,0 +1,58 @@
+"""packages.shared.title_overlay 单元测试。"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from packages.shared.title_overlay import apply_title_to_image, wrap_title_text
+
+
+@pytest.fixture
+def sample_image():
+    """生成一张 640x360 的纯黑测试图片。"""
+    from PIL import Image
+
+    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
+    tmp.close()
+    img = Image.new("RGB", (640, 360), color=(0, 0, 0))
+    img.save(tmp.name, "JPEG")
+    yield tmp.name
+    Path(tmp.name).unlink(missing_ok=True)
+
+
+def test_apply_title_to_image_empty_text_returns_none(sample_image):
+    assert apply_title_to_image(sample_image, "") is None
+    assert apply_title_to_image(sample_image, "   ") is None
+
+
+def test_apply_title_to_image_draws_title(sample_image):
+    result = apply_title_to_image(sample_image, "测试标题")
+    assert result == sample_image
+    assert Path(sample_image).exists()
+    assert Path(sample_image).stat().st_size > 0
+
+
+def test_apply_title_to_image_respects_position(sample_image):
+    for pos in ("top", "center", "bottom"):
+        result = apply_title_to_image(sample_image, "位置测试", position=pos)
+        assert result == sample_image
+
+
+def test_wrap_title_text_supports_long_text():
+    from PIL import ImageFont
+
+    font = ImageFont.load_default()
+    lines = wrap_title_text("这是一个比较长的标题需要自动换行处理ABC", font, max_width=40)
+    assert isinstance(lines, list)
+    assert len(lines) >= 1
+
+
+def test_wrap_title_text_respects_explicit_newline():
+    from PIL import ImageFont
+
+    font = ImageFont.load_default()
+    lines = wrap_title_text("第一行\n第二行", font, max_width=10000)
+    assert lines == ["第一行", "第二行"]