fix(test): mock storage.get_download_url，修复_sign_media_url内部调用mock泄漏问题

fix(test): 统一test_lipsync_tts持久化测试upload_url参数与key一致，消除flaky
style: prettier format sentences.ts & titleCanvas.ts [skip ci-format-check]
2026-09-13 02:21:08 +08:00 · 2026-09-13 02:05:30 +08:00 · 2026-09-13 01:34:47 +08:00 · 2026-09-13 01:14:15 +08:00 · 2026-09-12 17:05:14 +00:00 · 2026-09-13 00:58:02 +08:00
13 changed files with 538 additions and 277 deletions
@@ -18,7 +18,6 @@ from app.dependencies import get_db_session
 from app.schemas.ai_avatar_render import (
    AiAvatarRenderJobResponse,
    CreateAiAvatarRenderRequest,
-    SmartCoverRequest,
    SmartCoverResponse,
 )
 from app.services.ai_avatar_cover_service import generate_smart_cover
@@ -191,51 +190,6 @@ def retry_render_job(
    return AiAvatarRenderJobResponse.model_validate(job)


-# ── POST /smart-cover — 智能获取封面（MediaKit 抽帧 + 评分选帧）────────
-
-
-@router.post("/smart-cover", response_model=SmartCoverResponse)
-def generate_avatar_smart_cover(
-    body: SmartCoverRequest,
-    current_user: AuthenticatedUser = Depends(get_current_user),
-) -> SmartCoverResponse:
-    """智能获取数字人视频封面.
-
-    复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧逻辑（非 FFmpeg 简单截帧），
-    并将选中帧转存到自家 OSS，返回非临时的封面公网 URL。
-
-    前端「智能获取封面」按钮可直接调用本接口；不依赖渲染任务完成。
-    """
-    video_url = (body.video_url or "").strip()
-    if not video_url.startswith(("http://", "https://")):
-        raise HTTPException(status_code=400, detail="video_url 必须是合法的 HTTP/HTTPS URL")
-
-    try:
-        cover_url = generate_smart_cover(
-            video_url,
-            max_frames=body.max_frames,
-            title_config=getattr(body, "title_config", None),
-        )
-    except Exception as exc:
-        logger.error(
-            "智能封面生成异常: user=%s video_url=%s err=%s",
-            current_user.user.id,
-            video_url[:80],
-            exc,
-            exc_info=True,
-        )
-        cover_url = ""
-
-    if not cover_url:
-        return SmartCoverResponse(
-            cover_url="",
-            status="fallback_failed",
-            message="智能抽帧失败（MediaKit 不可用或抽帧异常），请稍后重试",
-        )
-    logger.info("智能封面生成成功: user=%s cover_url=%s", current_user.user.id, cover_url[:120])
-    return SmartCoverResponse(cover_url=cover_url, status="completed")
-
-
 # ── POST /{job_id}/smart-cover — 从最终成片智能抽封面（步骤②）────────


@@ -263,7 +217,7 @@ def generate_render_smart_cover(
        raise HTTPException(status_code=400, detail="渲染成片视频 URL 为空")

    try:
-        # 成片已叠加标题，不传 title_config 避免双重叠加
+        # 从最终成片抽帧，帧本身已含标题/B-roll，直接转存 OSS
        cover_url = generate_smart_cover(video_url, job_id=job_id, max_frames=5)
    except Exception as exc:
        logger.error(
@@ -52,7 +52,9 @@ class CreateAiAvatarRenderRequest(BaseModel):
    lipsync_job_id: str = Field(..., description="对口型任务 ID")
    script_id: str = Field("", description="文案 ID（选自文案库时传；手动输入文案直生场景可留空）")
    b_roll_segments: list[BRollSegment] = Field(default_factory=list, description="B-roll 片段列表")
-    title_config: dict[str, Any] = Field(default_factory=dict, description="标题配置")
+    title_config: dict[str, Any] = Field(
+        default_factory=dict, description="标题配置（可含 title_image_dataurl：前端 Canvas 渲染的标题 PNG dataURL）"
+    )
    cover_config: dict[str, Any] = Field(default_factory=dict, description="封面配置")
    project_id: str = Field("", description="项目 ID")

@@ -67,7 +69,6 @@ class CreateAiAvatarRenderRequest(BaseModel):
    @field_validator("script_id")
    @classmethod
    def validate_script_id(cls, v: str) -> str:
-        # script_id 可选：手动输入文案（TTS 直生）场景不关联文案库条目
        return (v or "").strip()


@@ -109,18 +110,8 @@ class AiAvatarRenderProgressResponse(BaseModel):
    error_message: str


-class SmartCoverRequest(BaseModel):
-    """智能封面请求 — MediaKit 抽帧 + 质量评分选最佳帧 + 可选标题 drawtext 叠加."""
-
-    video_url: str = Field(..., description="数字人视频 URL（对口型/渲染成片）")
-    max_frames: int = Field(5, ge=1, le=10, description="抽帧数量（默认 5）")
-    title_config: Optional[dict[str, Any]] = Field(
-        None, description="标题配置；传入时在封面上用 drawtext 叠加标题（竖屏 720x1280）"
-    )
-
-
 class SmartCoverResponse(BaseModel):
-    """智能封面响应."""
+    """智能封面响应（封面从最终成片抽帧，不再叠加标题）."""

    cover_url: str = Field("", description="封面图公网 URL（OSS，非临时）；失败为空")
    status: str = Field("completed", description="completed / fallback_failed")
@@ -1,18 +1,19 @@
-"""AI 数字人封面服务 — 复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧.
+"""AI 数字人封面服务 — MediaKit 抽帧 + 质量评分选最佳帧 + 转存 OSS.

 与 generation_cover.py 的智能选帧能力对齐（不再用 FFmpeg 简单截帧）：
 1. MediaKit extract_frames 抽取多帧（默认 5 帧，SpecifiedFrames 策略）
 2. cover_frame_scorer.score_frames 按清晰度/亮度/色彩评分选最佳
 3. 下载最佳帧并转存 OSS，返回公网封面 URL

+设计原则：封面一律从最终成片（已叠加标题/B-roll）抽帧，帧本身已含标题，
+本服务**不再叠加标题**。对口型阶段的裸视频封面入口已删除（废弃）。
+
 降级：MediaKit 不可用或抽帧失败时返回空字符串，由调用方决定回退策略。
 """

 from __future__ import annotations

 import logging
-import os
-import subprocess
 import tempfile
 import uuid
 from pathlib import Path
@@ -51,7 +52,6 @@ def _sign_video_url_for_mediakit(video_url: str) -> str:
        own_host = urlparse(public_base).netloc.lower()
        url_host = urlparse(video_url).netloc.lower()
        if own_host and url_host == own_host:
-            # 是自家 OSS URL，重签 7 天有效期供 MediaKit 拉取
            signed = storage.get_download_url(video_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS)
            if signed:
                logger.info("[数字人封面] video_url 已重签（自家 OSS 私有桶）")
@@ -62,19 +62,10 @@ def _sign_video_url_for_mediakit(video_url: str) -> str:


 def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
-    """从视频抽取多帧并评分选最佳帧，返回最佳帧的临时 URL.
-
-    Args:
-        video_url: 可公网访问的视频 URL
-        max_frames: 抽帧数量
-
-    Returns:
-        最佳帧图片 URL；失败返回空字符串
-    """
+    """从视频抽取多帧并评分选最佳帧，返回最佳帧的临时 URL."""
    if not video_url:
        return ""

-    # 确保 MediaKit 能访问 video_url（自家 OSS 私有桶需重签）
    video_url = _sign_video_url_for_mediakit(video_url)

    try:
@@ -87,11 +78,9 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
            return ""

        logger.info(
-            "[数字人封面] 开始抽帧: video_url=%s max_frames=%d poll_interval=%.1f max_poll=%d",
+            "[数字人封面] 开始抽帧: video_url=%s max_frames=%d",
            video_url[:80],
            max_frames,
-            COVER_POLL_INTERVAL,
-            COVER_MAX_POLL_ATTEMPTS,
        )

        snapshots = mk.extract_frames(
@@ -109,7 +98,6 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
        if len(snapshots) == 1:
            return snapshots[0].get("image_url") or snapshots[0].get("url") or ""

-        # 使用连接池下载各帧（复用 TCP 连接，减少延迟）
        import httpx

        candidates = []
@@ -137,7 +125,6 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
        best = scored[0] if scored else None
        best_url = best.get("url", "") if best else ""

-        # 清理临时文件
        for c in candidates:
            p = c.get("image_path")
            if p:
@@ -158,87 +145,19 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
        return ""


-def apply_title_to_cover(local_frame: str, *, title_config: dict | None) -> str:
-    """用 ffmpeg drawtext 在封面图上叠加标题，返回叠加后图片的本地路径.
-
-    ffmpeg 失败时回退返回原始 local_frame。竖屏封面按 720x1280 计算位置。
-    """
-    if not title_config or not isinstance(title_config, dict):
-        return local_frame
-    text = (title_config.get("text") or title_config.get("content") or "").strip()
-    if not text:
-        return local_frame
-    enabled = title_config.get("enabled", True)
-    if not enabled:
-        return local_frame
-
-    try:
-        from packages.domain.video_filter_builder import build_title_drawtext_filter
-
-        drawtext_filter = build_title_drawtext_filter(
-            title_config,
-            output_width=720,
-            output_height=1280,
-        )
-        if not drawtext_filter:
-            return local_frame
-
-        base, ext = os.path.splitext(local_frame)
-        titled_path = f"{base}_titled{ext or '.jpg'}"
-        cmd = [
-            "ffmpeg",
-            "-i",
-            local_frame,
-            "-vf",
-            drawtext_filter,
-            "-y",
-            titled_path,
-        ]
-        logger.info("[数字人封面] 叠加标题: text=%s", text[:30])
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=30,
-        )
-        if result.returncode != 0:
-            logger.warning(
-                "[数字人封面] drawtext 失败，回退无标题: exit=%s stderr=%s",
-                result.returncode,
-                (result.stderr or "")[-300:],
-            )
-            return local_frame
-        if not os.path.exists(titled_path) or os.path.getsize(titled_path) == 0:
-            logger.warning("[数字人封面] drawtext 输出为空，回退无标题")
-            return local_frame
-        return titled_path
-    except Exception as exc:
-        logger.warning("[数字人封面] 标题叠加异常，回退无标题: %s", exc, exc_info=True)
-        return local_frame
-
-
 def persist_cover_to_oss(
    frame_url: str,
    *,
    job_id: str = "",
    prefix: str = "ai-avatar/covers",
-    title_config: dict | None = None,
 ) -> str:
-    """下载帧图并转存到 OSS，返回公网封面 URL.
+    """下载最佳帧图并转存到 OSS，返回公网封面 URL（预签名）.

-    Args:
-        frame_url: MediaKit 返回的临时帧图 URL
-        job_id: 关联任务 ID（用于 OSS key 命名）
-        prefix: OSS key 前缀
-        title_config: 可选标题配置；传入时用 drawtext 叠加标题（竖屏 720x1280）
-
-    Returns:
-        OSS 公网 URL；失败回退原始 frame_url
+    封面来自最终成片抽帧，帧本身已含标题，本函数不再做任何文字/图片叠加。
    """
    if not frame_url:
        return ""
    tmp_path: Optional[str] = None
-    titled_path: Optional[str] = None
    try:
        import httpx

@@ -259,21 +178,12 @@ def persist_cover_to_oss(
        token = job_id or uuid.uuid4().hex[:12]
        cover_key = f"{prefix}/{token}/cover_{uuid.uuid4().hex[:8]}.jpg"

-        upload_path = apply_title_to_cover(tmp_path, title_config=title_config)
-        if upload_path != tmp_path:
-            titled_path = upload_path
-
        public_url = storage.upload_file(
-            file_or_path=upload_path,
+            file_or_path=tmp_path,
            storage_key=cover_key,
            content_type="image/jpeg",
        )
-        logger.info(
-            "[数字人封面] 封面已转存 OSS: key=%s titled=%s",
-            cover_key,
-            bool(titled_path),
-        )
-        # 私有桶：返回预签名 URL（前端才能加载）
+        logger.info("[数字人封面] 封面已转存 OSS: key=%s", cover_key)
        if public_url:
            signed = storage.get_download_url(cover_key, expires_seconds=86400)
            return signed
@@ -282,12 +192,11 @@ def persist_cover_to_oss(
        logger.warning("[数字人封面] 封面转存 OSS 失败，返回原始 URL", exc_info=True)
        return frame_url
    finally:
-        for p in (tmp_path, titled_path):
-            if p:
-                try:
-                    Path(p).unlink(missing_ok=True)
-                except Exception:
-                    pass
+        if tmp_path:
+            try:
+                Path(tmp_path).unlink(missing_ok=True)
+            except Exception:
+                pass


 def generate_smart_cover(
@@ -295,19 +204,12 @@ def generate_smart_cover(
    *,
    job_id: str = "",
    max_frames: int = 5,
-    title_config: dict | None = None,
 ) -> str:
-    """一站式：MediaKit 智能抽帧选最佳 → （可选）drawtext 叠加标题 → 转存 OSS.
+    """一站式：MediaKit 智能抽帧选最佳 → 转存 OSS。失败返回空字符串。

-    供独立封面接口与渲染管线复用。失败返回空字符串。
-
-    Args:
-        video_url: 可公网访问的视频 URL
-        job_id: 关联任务 ID
-        max_frames: 抽帧数量
-        title_config: 可选标题配置；传入时在封面上叠加 drawtext 标题（竖屏 720x1280）
+    封面从最终成片抽帧，不再叠加任何标题（帧本身已含）。
    """
    best_frame = select_best_cover_frame(video_url, max_frames=max_frames)
    if not best_frame:
        return ""
-    return persist_cover_to_oss(best_frame, job_id=job_id, title_config=title_config)
+    return persist_cover_to_oss(best_frame, job_id=job_id)
@@ -9,6 +9,8 @@

 from __future__ import annotations

+import base64
+import binascii
 import logging
 import os
 import subprocess
@@ -27,6 +29,7 @@ from packages.adapters.sqlalchemy_impl.models import (
 from packages.domain.video_filter_builder import (
    build_broll_overlay_filter,
    build_title_drawtext_filter,
+    build_title_overlay_filter,
 )
 from packages.shared.storage import get_shared_storage_service

@@ -248,29 +251,12 @@ class AiAvatarRenderService:
                output_height=output_height,
            )

-            # 标题叠加（传入实际输出尺寸，保证位置计算正确）
-            title_filter = build_title_drawtext_filter(
-                job.title_config,
-                output_width=output_width,
-                output_height=output_height,
-            )
-
-            filter_complex = ""
-            final_label = None
-            if broll_filter and title_filter:
-                # B-roll → 标题叠在 B-roll 输出上
-                filter_complex = broll_filter + f";[{broll_label}]{title_filter}[vout_titled]"
-                final_label = "vout_titled"
-            elif broll_filter:
-                filter_complex = broll_filter
-                final_label = broll_label
-            elif title_filter:
-                filter_complex = f"[0:v]{title_filter}[vout_titled]"
-                final_label = "vout_titled"
-            else:
-                # 无滤镜：直接拷贝视频流
-                filter_complex = ""
-                final_label = None
+            # 标题叠加路径：优先前端 Canvas 渲染的 PNG 图层（所见即所得），
+            # 无 title_image_dataurl 时降级到 drawtext 重画文字。
+            title_cfg = job.title_config if isinstance(job.title_config, dict) else {}
+            title_dataurl = (title_cfg or {}).get("title_image_dataurl") if title_cfg else None
+            use_title_png = isinstance(title_dataurl, str) and title_dataurl.startswith("data:image/")
+            title_input_index = 1 + len(job.b_roll_segments or []) if use_title_png else None

            job.progress = 40
            self.db.commit()
@@ -279,9 +265,81 @@ class AiAvatarRenderService:
            with tempfile.TemporaryDirectory() as tmpdir:
                output_video_path = os.path.join(tmpdir, "output.mp4")

+                # 在临时目录里解码保存标题 PNG（with 退出自动清理）
+                title_png_path: Optional[str] = None
+                extra_inputs: list[str] = []
+                title_filter = None
+                if use_title_png:
+                    try:
+                        title_png_path = os.path.join(tmpdir, f"title_{job.id}.png")
+                        self._save_title_dataurl_to_file(title_dataurl, dst_path=title_png_path)
+                        extra_inputs.append(title_png_path)
+                        logger.info(
+                            "[数字人渲染] 标题 PNG 已保存: %s (input index %d)", title_png_path, title_input_index
+                        )
+                    except Exception as exc:
+                        logger.warning("[数字人渲染] 标题 PNG 解码/保存失败，降级 drawtext: %s", exc)
+                        title_png_path = None
+                        extra_inputs = []
+
+                # 构建标题滤镜
+                final_label = None
+                if title_png_path and title_input_index is not None:
+                    title_input_label = f"[{title_input_index}:v]"
+                    base_label = f"[{broll_label}]" if broll_label else "[0:v]"
+                    title_filter = build_title_overlay_filter(
+                        title_cfg,
+                        output_width=output_width,
+                        output_height=output_height,
+                        title_png_path=title_png_path,
+                        title_input_label=title_input_label,
+                        base_label=base_label,
+                        output_label="vout_titled",
+                    )
+                    if not title_filter:
+                        # build 返回 None → 文件不存在（极端并发情况），降级 drawtext
+                        title_png_path = None
+                        extra_inputs = []
+
+                if title_png_path:
+                    # overlay 路径
+                    if broll_filter and title_filter:
+                        filter_complex = broll_filter + f";{title_filter}"
+                    elif broll_filter:
+                        filter_complex = broll_filter
+                        final_label = broll_label
+                    elif title_filter:
+                        filter_complex = title_filter
+                    else:
+                        filter_complex = ""
+                    if title_filter:
+                        final_label = "vout_titled"
+                    elif not final_label:
+                        final_label = None
+                else:
+                    # 降级：drawtext 重画文字
+                    title_filter = build_title_drawtext_filter(
+                        title_cfg,
+                        output_width=output_width,
+                        output_height=output_height,
+                    )
+                    if broll_filter and title_filter:
+                        filter_complex = broll_filter + f";[{broll_label}]{title_filter}[vout_titled]"
+                        final_label = "vout_titled"
+                    elif broll_filter:
+                        filter_complex = broll_filter
+                        final_label = broll_label
+                    elif title_filter:
+                        filter_complex = f"[0:v]{title_filter}[vout_titled]"
+                        final_label = "vout_titled"
+                    else:
+                        filter_complex = ""
+                        final_label = None
+
                cmd_list = self._build_ffmpeg_command(
                    input_video=input_video_path,
                    b_roll_segments=job.b_roll_segments,
+                    extra_inputs=extra_inputs,
                    filter_complex=filter_complex,
                    final_label=final_label,
                    output_path=output_video_path,
@@ -408,6 +466,44 @@ class AiAvatarRenderService:
                os.unlink(tmp.name)
            raise

+    @staticmethod
+    def _save_title_dataurl_to_file(dataurl: str, *, dst_path: str | None = None, job_id: str = "") -> str:
+        """解码前端传来的 data:image/png;base64,... 并保存为本地 PNG 文件。
+
+        Args:
+            dataurl: 完整 dataURL 字符串
+            dst_path: 指定输出路径；为 None 时创建临时文件并返回路径
+            job_id: 仅在 dst_path 为空时用于临时文件命名
+
+        Returns:
+            保存后的本地文件路径
+        """
+        if not isinstance(dataurl, str) or not dataurl.startswith("data:image/"):
+            raise ValueError("title_image_dataurl 不是合法的 data:image URL")
+        # 拆分 data:image/png;base64,<payload>
+        try:
+            header, b64 = dataurl.split(",", 1)
+        except ValueError as exc:
+            raise ValueError("title_image_dataurl 缺少 base64 payload") from exc
+        if "base64" not in header:
+            raise ValueError("title_image_dataurl 不是 base64 编码")
+        try:
+            png_bytes = base64.b64decode(b64, validate=True)
+        except (binascii.Error, ValueError) as exc:
+            raise ValueError(f"title_image_dataurl base64 解码失败: {exc}") from exc
+        if not png_bytes:
+            raise ValueError("title_image_dataurl 解码后为空")
+
+        if dst_path:
+            out_path = dst_path
+            with open(out_path, "wb") as f:
+                f.write(png_bytes)
+            return out_path
+        suffix = f"_title_{job_id}.png" if job_id else "_title.png"
+        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
+            tmp.write(png_bytes)
+            return tmp.name
+
    @staticmethod
    def _probe_video_resolution(video_path: str) -> tuple[int, int]:
        """用 ffprobe 探测视频分辨率，返回 (width, height)；失败返回 (0, 0)。"""
@@ -444,6 +540,7 @@ class AiAvatarRenderService:
        *,
        input_video: str,
        b_roll_segments: list[dict[str, Any]],
+        extra_inputs: list[str] | None = None,
        filter_complex: str,
        final_label: Optional[str],
        output_path: str,
@@ -460,6 +557,9 @@ class AiAvatarRenderService:
            asset_url = seg.get("asset_url", "")
            if asset_url:
                cmd.extend(["-i", asset_url])
+        # 额外输入（例如前端 Canvas 渲染的标题 PNG）
+        for extra in extra_inputs or []:
+            cmd.extend(["-i", extra])

        if filter_complex and final_label:
            cmd.extend(
@@ -55,13 +55,17 @@ def _sign_media_url(url: str) -> str:


 def _split_script_into_sentences(script_text: str) -> list[str]:
-    """按句号/问号/感叹号/分号/换行分句（与前端 splitScriptIntoSentences 一致）."""
+    """按句号/问号/感叹号/分号/逗号/换行分句（与前端 SENTENCE_SPLIT_RE 一致）.
+
+    中文短视频文案习惯用「，」断小句（如"卖花的叫花无缺，卖姜的叫姜子牙"），
+    必须把逗号也纳入分隔符，否则多句文案会被识别成一整句，导致 B-roll 时间戳错位。
+    """
    import re

    text = (script_text or "").strip()
    if not text:
        return []
-    parts = re.split(r"[。！？!?；;\n\r]+", text)
+    parts = re.split(r"[。！？!?？！;；,，\n\r]+", text)
    return [p.strip() for p in parts if p.strip()]


@@ -32,6 +32,7 @@ import {
  buildTitleConfigPayload,
  buildCoverConfigPayload,
 } from "./utils/contract"
+import { renderTitleToPngDataUrl, getVideoResolution } from "./utils/titleCanvas"

 /** 面板折叠状态 */
 type PanelKey = "video" | "voice" | "script" | "lipsync" | "title" | "cover"
@@ -210,6 +211,23 @@ const AiAvatarPage: React.FC = () => {
    try {
      // 确保有 project_id（AI数字人入口独立，不在项目内，自动取默认项目；#1860 P0 bugfix）
      const defaultProject = await getOrCreateDefaultProject()
+
+      // 用 Canvas 预渲染标题为 PNG dataURL（所见即所得，后端用 overlay 直接叠加）
+      let titleImageDataUrl: string | null = null
+      if (state.titleConfig.title?.trim()) {
+        try {
+          const res = await getVideoResolution(state.lipsyncJob.output_video_url || "")
+          titleImageDataUrl = renderTitleToPngDataUrl({
+            titleConfig: state.titleConfig,
+            videoWidth: res.width,
+            videoHeight: res.height,
+          })
+        } catch (canvasErr) {
+          console.warn("[渲染] 标题 Canvas 渲染失败，降级 drawtext:", canvasErr)
+          titleImageDataUrl = null
+        }
+      }
+
      const job = await submitRender({
        lipsync_job_id: state.lipsyncJob.id,
        script_id: state.script?.id,
@@ -223,7 +241,7 @@ const AiAvatarPage: React.FC = () => {
          pip_position: seg.pip_position,
          pip_scale: seg.pip_scale,
        })) as never,
-        title_config: buildTitleConfigPayload(state.titleConfig),
+        title_config: buildTitleConfigPayload(state.titleConfig, titleImageDataUrl),
        // 封面不阻塞渲染：用户未选定封面时传空 dict，后端不生成封面；渲染完成后再单独抽帧
        cover_config:
          state.coverConfig.smart_cover_url ||
@@ -58,21 +58,6 @@ export const getLipsyncJob = async (id: string): Promise<LipsyncJob> => {
  return response.data
 }

-/* ── 智能封面（MediaKit 抽帧 + 质量评分选最佳帧 + 可选 drawtext 标题叠加） ── */
-export const generateSmartCover = async (
-  video_url: string,
-  title_config?: Record<string, unknown> | null,
-  max_frames = 5,
-): Promise<{ cover_url: string; status: string; message: string }> => {
-  const response = await apiClient.post<{ cover_url: string; status: string; message: string }>(
-    "/ai-avatar/render/smart-cover",
-    { video_url, max_frames, title_config: title_config ?? null },
-    // smart-cover 链路：下载视频+抽帧+drawtext 加标题+上传 OSS，需要较长时间，120s 超时
-    { timeout: 120000 },
-  )
-  return response.data
-}
-
 /* ── 渲染 ── */
 export const submitRender = async (data: {
  lipsync_job_id: string
@@ -82,6 +67,7 @@ export const submitRender = async (data: {
  cover_config?: Record<string, unknown>
  project_id?: string
 }): Promise<RenderJob> => {
+  // title_config 内可含 title_image_dataurl（前端 Canvas 渲染的 PNG dataURL）
  const response = await apiClient.post<RenderJob>("/ai-avatar/render", data)
  return response.data
 }
@@ -28,10 +28,13 @@ export function normalizeEmotion(raw: string | undefined | null): VoiceEmotion {
 * 后端真实字段：text(或content)、font(或font_preset)、font_size(或size)、
 * font_color(或color，可传 #RRGGBB)、position(top/center/bottom/custom)、
 * enabled、bold、stroke{enabled,width,color}、shadow{enabled,color,offset_x,offset_y}、
- * pos_x/pos_y(custom 时)。
+ * pos_x/pos_y(custom 时)、title_image_dataurl（前端 Canvas 渲染的 PNG dataURL，WYSIWYG 路径优先）。
 * 口播标题默认 position=bottom（不传后端会默认 top 跑到画面顶部）。
 */
-export function buildTitleConfigPayload(cfg: AiAvatarTitleConfig): Record<string, unknown> {
+export function buildTitleConfigPayload(
+  cfg: AiAvatarTitleConfig,
+  titleImageDataUrl?: string | null,
+): Record<string, unknown> {
  const text = (cfg.title || "").trim()
  if (!text) return {}
  const position = cfg.position || "bottom"
@@ -53,6 +56,10 @@ export function buildTitleConfigPayload(cfg: AiAvatarTitleConfig): Record<string
    payload.pos_x = cfg.pos_x
    payload.pos_y = cfg.pos_y
  }
+  // 前端 Canvas 渲染好的 PNG dataURL（所见即所得，后端优先 overlay 此图片图层）
+  if (titleImageDataUrl) {
+    payload.title_image_dataurl = titleImageDataUrl
+  }
  return payload
 }

@@ -1,8 +1,10 @@
 /**
 * AI数字人 — 文案分句 & B-roll 时间计算
 *
- * 优先使用后端基于 TTS 音频静音检测计算的精确 sentence_timings；
- * 后端未返回（如对口型还在生成中）时，降级为前端按字数比例估算。
+ * 数据来源优先级：
+ *   1. 后端 sentence_timings（基于 TTS 音频静音检测，精确到句子边界）—— 直接使用，不重新分句
+ *   2. 后端 output_duration（最终渲染视频时长） + 本地分句 —— 按字数比例估算
+ *   3. 两者都没有（对口型还在生成中）—— 返回分句文本但 startTime/endTime 全部 0，等数据到位重算
 */

 export interface ScriptSentence {
@@ -20,45 +22,47 @@ export interface ScriptSentence {
  endTime: number
 }

+/** 句子分隔符：中英文句号/问号/感叹号/分号/逗号/换行（覆盖中文短视频常用断句） */
+const SENTENCE_SPLIT_RE = /[。！？!?？！;；,，\n\r]+/
+
 /**
- * 按句号/问号/感叹号/分号/换行分句（兼容中英文标点）。
- * 空文案返回空数组。时间优先使用后端 sentence_timings；否则按字数线性估算。
+ * 分句并计算每句的起止时间。
 *
 * @param sentenceTimings 后端返回的精确句子时间戳（来自 lipsync_job.sentence_timings）。
- *   非空且有效时优先采用，跳过前端估算。
+ *   非空时直接按后端返回的句子列表渲染，不再本地分句（避免前后端分句不一致导致时间错位）。
+ * @param outputDuration  最终视频时长（秒）。对口型预览阶段可能为 0，此时降级估算只能给 0。
 */
 export function splitScriptIntoSentences(
  scriptText: string,
-  sentenceTimings?: { index: number; text: string; start_time: number; end_time: number }[] | null,
+  sentenceTimings?:
+    { index?: number; text?: string; start_time: number; end_time: number }[] | null,
  outputDuration: number = 0,
 ): ScriptSentence[] {
  const text = (scriptText || "").trim()
  if (!text) return []

-  // 1. 先做基础分句（仅用于降级估算 / 没有 sentenceTimings 时）
-  const rawParts = text
-    .split(/[。！？!?；;\n\r]+/)
-    .map((part) => part.trim())
-    .filter((part) => part.length > 0)
-
-  // 2. 优先使用后端精确时间戳
-  //    校验：必须是数组、条数一致、每条都有 start_time/end_time，否则降级估算
-  if (Array.isArray(sentenceTimings) && sentenceTimings.length === rawParts.length) {
+  // 1. 后端返回了 sentence_timings：校验通过就直接用，跳过本地分句
+  //    校验条件放宽：只要是数组、至少1条、每条 start_time/end_time 是数字即可
+  //    （不再强制要求条数相等——后端静音检测可能按停顿切出更多/更少边界，
+  //    比如文案用逗号连写时本地只分1句、后端按停顿切4句，后端的切法才是对的）
+  if (Array.isArray(sentenceTimings) && sentenceTimings.length > 0) {
    const valid = sentenceTimings.every(
      (t) =>
        t &&
        typeof t.start_time === "number" &&
        typeof t.end_time === "number" &&
+        isFinite(t.start_time) &&
+        isFinite(t.end_time) &&
        t.end_time >= t.start_time,
    )
    if (valid) {
      let accChar = 0
      return sentenceTimings.map((t, i) => {
-        const part = rawParts[i] ?? t.text ?? ""
-        const charCount = part.replace(/\s/g, "").length
+        const sentenceText = (t.text || "").trim() || `句子${i + 1}`
+        const charCount = sentenceText.replace(/\s/g, "").length
        const sentence: ScriptSentence = {
-          index: t.index ?? i,
-          text: part,
+          index: typeof t.index === "number" ? t.index : i,
+          text: sentenceText,
          charCount,
          startChar: accChar,
          startTime: round1(t.start_time),
@@ -70,7 +74,12 @@ export function splitScriptIntoSentences(
    }
  }

-  // 3. 降级：按字数比例线性估算
+  // 2. 本地分句 + 按字数比例估算（降级路径）
+  const rawParts = text
+    .split(SENTENCE_SPLIT_RE)
+    .map((part) => part.trim())
+    .filter((part) => part.length > 0)
+
  const totalChars = rawParts.reduce((sum, part) => sum + part.replace(/\s/g, "").length, 0)
  const duration = outputDuration > 0 ? outputDuration : 0

@@ -0,0 +1,174 @@
+/**
+ * AI数字人 — 标题 Canvas 渲染工具
+ *
+ * 把标题按前端预览的 HTML/CSS 效果画到透明背景 PNG 上（与视频同分辨率），
+ * 以 dataURL 形式传给后端，后端用 FFmpeg overlay 直接叠加图层，
+ * 彻底解决前端 HTML/CSS 预览 ≠ FFmpeg drawtext 成片的 WYSIWYG 问题。
+ */
+import type { AiAvatarTitleConfig } from "../types"
+
+export interface RenderTitlePngOptions {
+  /** 标题配置 */
+  titleConfig: AiAvatarTitleConfig
+  /** 视频宽度（像素），默认 720 */
+  videoWidth?: number
+  /** 视频高度（像素），默认 1280 */
+  videoHeight?: number
+}
+
+/**
+ * 将标题渲染为透明背景 PNG 的 dataURL（data:image/png;base64,...）
+ * Canvas 尺寸与视频一致，保证叠加时 1:1 像素对齐。
+ *
+ * 标题为空时返回 null。
+ */
+export function renderTitleToPngDataUrl(opts: RenderTitlePngOptions): string | null {
+  const { titleConfig, videoWidth = 720, videoHeight = 1280 } = opts
+  if (!titleConfig) return null
+  const rawTitle = (titleConfig.title || "").trim()
+  if (!rawTitle) return null
+
+  // 按 / 或 ／ 分割为多行
+  const lines = rawTitle
+    .split(/[/／]/)
+    .map((l) => l.trim())
+    .filter((l) => l.length > 0)
+  if (lines.length === 0) return null
+
+  const canvas = document.createElement("canvas")
+  canvas.width = videoWidth
+  canvas.height = videoHeight
+  const ctx = canvas.getContext("2d")
+  if (!ctx) return null
+
+  const size = Math.max(12, Math.round(titleConfig.size || 48))
+  const bold = !!titleConfig.bold
+  const italic = !!titleConfig.italic
+  const color = titleConfig.color || "#ffffff"
+  const stroke = !!titleConfig.stroke
+  const shadow = !!titleConfig.shadow
+
+  // 字体族 fallback 链：优先中文字体
+  const fontFamily =
+    '"Noto Sans CJK SC","Source Han Sans CN","PingFang SC","Microsoft YaHei",sans-serif'
+  const fontParts: string[] = []
+  if (italic) fontParts.push("italic")
+  if (bold) fontParts.push("bold")
+  fontParts.push(`${size}px`, fontFamily)
+  ctx.font = fontParts.join(" ")
+  ctx.fillStyle = color
+  ctx.textAlign = "center"
+  ctx.textBaseline = "middle"
+
+  // 阴影（shadow=true 时开启）
+  if (shadow) {
+    ctx.shadowColor = "rgba(0,0,0,0.8)"
+    ctx.shadowBlur = 4
+    ctx.shadowOffsetX = 0
+    ctx.shadowOffsetY = 2
+  }
+
+  // 位置计算：与 PanelLipsyncPreview 的 CSS 对齐
+  // 预览用 top/bottom 8px padding + transform translateX(-50%) 居中；
+  // 这里画到整尺寸 canvas，padding 按比例放大到全分辨率（预览缩放 0.35x 时 8px ≈ 23px 全尺寸，
+  // 为更贴近原 CSS 16px 安全边距，用 16px 作为内边距）。
+  const PAD = 16
+  let centerX = videoWidth / 2
+  const position = titleConfig.position || "bottom"
+  const lineGap = size * 1.2
+  const totalTextH = lines.length * lineGap - (lineGap - size) // 所有行的总高度
+  // 文本块顶部 y（textBaseline=middle 时首行基线）
+  let firstLineY: number
+  if (
+    position === "custom" &&
+    typeof titleConfig.pos_x === "number" &&
+    typeof titleConfig.pos_y === "number"
+  ) {
+    centerX = (Math.max(0, Math.min(100, titleConfig.pos_x)) / 100) * videoWidth
+    const centerY = (Math.max(0, Math.min(100, titleConfig.pos_y)) / 100) * videoHeight
+    firstLineY = centerY - totalTextH / 2 + size / 2
+  } else if (position === "top") {
+    // 顶部：y = size/2 + PAD
+    firstLineY = size / 2 + PAD
+  } else if (position === "center") {
+    firstLineY = videoHeight / 2 - totalTextH / 2 + size / 2
+  } else {
+    // bottom（默认）
+    firstLineY = videoHeight - totalTextH - PAD + size / 2
+  }
+
+  // 描边参数（stroke=true 或 bold 默认细描边模拟粗体时都画；
+  // 注意：浏览器原生 bold 已经是粗体 glyph，Canvas 这里对 stroke=true 才加黑描边，
+  // 与预览 CSS 的 WebkitTextStroke 保持一致，不对 bold 自动加描边避免双粗）。
+  const doStroke = stroke
+  // 逐行绘制
+  lines.forEach((line, idx) => {
+    const y = firstLineY + idx * lineGap
+    if (doStroke) {
+      const prevShadowColor = ctx.shadowColor
+      const prevShadowBlur = ctx.shadowBlur
+      // 描边不要带阴影（避免黑色描边发虚）
+      ctx.shadowColor = "rgba(0,0,0,0)"
+      ctx.shadowBlur = 0
+      ctx.lineWidth = Math.max(2, size * 0.06)
+      ctx.strokeStyle = "#000000"
+      ctx.lineJoin = "round"
+      ctx.strokeText(line, centerX, y)
+      // 恢复阴影
+      if (shadow) {
+        ctx.shadowColor = "rgba(0,0,0,0.8)"
+        ctx.shadowBlur = 4
+      } else {
+        ctx.shadowColor = prevShadowColor
+        ctx.shadowBlur = prevShadowBlur
+      }
+    }
+    ctx.fillText(line, centerX, y)
+  })
+
+  try {
+    return canvas.toDataURL("image/png")
+  } catch {
+    return null
+  }
+}
+
+/**
+ * 获取视频真实分辨率（HTMLVideoElement + loadedmetadata，超时 3 秒兜底 720×1280）。
+ */
+export function getVideoResolution(
+  videoUrl: string,
+  timeoutMs = 3000,
+): Promise<{ width: number; height: number }> {
+  return new Promise((resolve) => {
+    if (!videoUrl) {
+      resolve({ width: 720, height: 1280 })
+      return
+    }
+    const video = document.createElement("video")
+    video.preload = "metadata"
+    video.muted = true
+    video.playsInline = true
+    video.crossOrigin = "anonymous"
+    let settled = false
+    const done = (w: number, h: number) => {
+      if (settled) return
+      settled = true
+      video.removeAttribute("src")
+      video.load()
+      resolve({ width: w, height: h })
+    }
+    const timer = window.setTimeout(() => done(720, 1280), timeoutMs)
+    video.onloadedmetadata = () => {
+      window.clearTimeout(timer)
+      const w = video.videoWidth || 720
+      const h = video.videoHeight || 1280
+      done(w, h)
+    }
+    video.onerror = () => {
+      window.clearTimeout(timer)
+      done(720, 1280)
+    }
+    video.src = videoUrl
+  })
+}
@@ -420,44 +420,29 @@ def _escape_drawtext_text(text: str) -> str:
    return result


-# 粗体字体文件映射：服务器镜像只保留了 NotoSansSC-VF.ttf（可变字体，已删除
-# NotoSansCJK-Bold.ttc 以避免 Mono 变体问题，见 worker-base.Dockerfile），
-# 因此无法通过 fontfile 切换到 Bold 字重。这里保留路径列表作为未来扩展，
-# 实际加粗通过 borderw 黑色描边实现（见下）。
-DRAWTEXT_BOLD_FONT_SEARCH_PATHS: list[str] = [
-    "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
-    "/usr/share/fonts/noto-cjk/NotoSansCJK-Bold.ttc",
-    "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Bold.ttc",
-    "/usr/share/fonts/truetype/noto/NotoSansSC-Bold.ttf",
-    "/usr/share/fonts/noto/NotoSansSC-Bold.ttf",
-]
+# 粗体字体已由前端 Canvas 直接渲染（Canvas 使用浏览器原生粗体 glyph），
+# FFmpeg 侧不再需要查找 Bold 字体文件；drawtext 仅作为旧版前端的降级路径，
+# 通过 borderw 黑色细描边模拟粗体（见 build_title_drawtext_filter）。


-def _resolve_font_path(font_name: str, bold: bool = False) -> str:
+def _resolve_font_path(font_name: str) -> str:
    """解析字体名到服务器实际字体文件路径。

    查找策略：
    1. 通过 DRAWTEXT_FONT_MAP 映射前端字体名到服务器关键字
-    2. bold=True 时优先查找粗体变体；找不到回退常规字重
-    3. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径
-    4. 未找到则返回空字符串（drawtext 使用内置默认字体）
+    2. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径
+    3. 未找到则返回空字符串（drawtext 使用内置默认字体）
+
+    注：粗体已由前端 Canvas 渲染时直接用浏览器 bold glyph 绘制，
+    此处仅作为旧版前端降级路径，无需切换 Bold 字体文件。
    """
    keyword = DRAWTEXT_FONT_MAP.get(font_name, font_name)
    import os

-    if bold:
-        for path in DRAWTEXT_BOLD_FONT_SEARCH_PATHS:
-            if keyword.lower() in path.lower() and os.path.isfile(path):
-                return path
-        # 粗体文件找不到时，再查常规字重（后面会用描边兜底加粗）
    for path in DRAWTEXT_FONT_SEARCH_PATHS:
        if keyword.lower() in path.lower() and os.path.isfile(path):
            return path
    # fallback：遍历搜索任意可用字体
-    if bold:
-        for path in DRAWTEXT_BOLD_FONT_SEARCH_PATHS:
-            if os.path.isfile(path):
-                return path
    for path in DRAWTEXT_FONT_SEARCH_PATHS:
        if os.path.isfile(path):
            return path
@@ -493,8 +478,10 @@ def build_title_drawtext_filter(
    if not title_config or not isinstance(title_config, dict):
        return None

-    # 字段名归一化：兼容 content/text、font_preset/font 两套命名
-    text = (title_config.get("text") or title_config.get("content") or "").strip()
+    # 字段名归一化：兼容 content/text/title 三套命名
+    text = (
+        title_config.get("text") or title_config.get("content") or title_config.get("title") or ""
+    ).strip()
    if not text:
        return None

@@ -518,8 +505,8 @@ def build_title_drawtext_filter(
    # ── 构建 drawtext 参数 ──
    params: list[str] = []

-    # 字体文件：粗体优先使用 Bold 字体文件，避免同色描边造成字形偏移/重影
-    font_path = _resolve_font_path(font_name, bold=bold)
+    # 字体文件（drawtext 降级路径：粗体通过 borderw 黑色描边模拟）
+    font_path = _resolve_font_path(font_name)
    if font_path:
        escaped_path = font_path.replace("\\", "\\\\").replace(":", "\\\\:").replace("'", "\\\\'")
        params.append(f"fontfile='{escaped_path}'")
@@ -599,6 +586,43 @@ def build_title_drawtext_filter(
    return "drawtext=" + ":".join(params)


+def build_title_overlay_filter(
+    title_config: dict[str, Any],
+    output_width: int,  # noqa: ARG001 - 保留参数签名，PNG 已按视频分辨率绘制
+    output_height: int,  # noqa: ARG001
+    title_png_path: str,
+    *,
+    title_input_label: str = "[1:v]",
+    base_label: str = "[0:v]",
+    output_label: str = "vout_titled",
+) -> str | None:
+    """构建标题 PNG 图层 overlay 滤镜（WYSIWYG 路径）。
+
+    前端用 Canvas 把标题画成与视频同分辨率的透明 PNG（所见即所得），
+    后端直接 overlay=0:0 叠加即可，PNG 透明区域不遮挡视频。
+
+    Args:
+        title_config: 标题配置 dict（仅用来判断降级）
+        output_width: 输出宽度（未使用，PNG 已按该分辨率绘制）
+        output_height: 输出高度（未使用）
+        title_png_path: 已保存到本地的标题 PNG 文件路径
+        title_input_label: 标题 PNG 在 filter_complex 中的输入标签（默认 "[1:v]"）
+        base_label: 前序滤镜输出标签（如 B-roll 输出 "[vout]"）
+        output_label: overlay 输出标签名
+
+    Returns:
+        overlay 滤镜字符串；title_png_path 为空/文件不存在时返回 None（降级到 drawtext）
+    """
+    import os
+
+    if not title_png_path or not os.path.isfile(title_png_path):
+        return None
+    if not title_config or not isinstance(title_config, dict):
+        return None
+
+    return f"{base_label}{title_input_label}overlay=0:0[{output_label}]"
+
+
 # ── B-roll 叠加滤镜 ─────────────────────────────────────────────────────────


@@ -428,6 +428,9 @@ class TestPersistOutputVideoTask:
        storage = MagicMock()
        storage.public_url = "https://oss.example.com/"
        storage.upload_file.return_value = upload_url or "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4"
+        # _sign_media_url 内部会调 storage.get_download_url，必须mock返回字符串
+        _upload_url = upload_url or "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4"
+        storage.get_download_url.return_value = _upload_url + "?signed"

        fake_httpx = ModuleType("httpx")
        fake_httpx.Client = FakeHttpxClient
@@ -448,7 +451,7 @@ class TestPersistOutputVideoTask:

        job = self._make_persist_job(output_video_url="https://temp.mk/x.mp4")
        session, fake_client, storage, patches = self._persist_patches(
-            job=job, video_bytes=b"VIDEODATA", upload_url="https://oss.example.com/lipsync-outputs/u1/j1.mp4"
+            job=job, video_bytes=b"VIDEODATA", upload_url="https://oss.example.com/lipsync-outputs/user-1/job-1.mp4"
        )
        entered = [p.__enter__() for p in patches]
        try:
@@ -459,17 +462,11 @@ class TestPersistOutputVideoTask:

        fake_client.get.assert_called_once_with("https://temp.mk/x.mp4")
        storage.upload_file.assert_called_once()
-        # 上传的 key 必须是 lipsync-outputs/{user_id}/{job_id}.mp4
-        key_arg = (
-            storage.upload_file.call_args.args[1]
-            if storage.upload_file.call_args.args
-            else storage.upload_file.call_args.kwargs.get("key")
-        )
-        # upload_file(data, key, content_type=...)
        call_args = storage.upload_file.call_args.args
+        # 上传的 key 必须是 lipsync-outputs/{user_id}/{job_id}.mp4
        assert call_args[1] == "lipsync-outputs/user-1/job-1.mp4"
-        # output_video_url 被替换为签名后的永久 URL
-        assert job.output_video_url == "https://oss.example.com/lipsync-outputs/u1/j1.mp4?signed"
+        # upload_file 返回永久 URL，再被 _sign_media_url 追加 ?signed
+        assert job.output_video_url == "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4?signed"
        assert job.updated_at is not None
        session.commit.assert_called_once()
        session.close.assert_called_once()
@@ -13,6 +13,7 @@

 from __future__ import annotations

+import os
 import unittest
 from dataclasses import FrozenInstanceError
 from unittest.mock import patch
@@ -29,10 +30,12 @@ from packages.domain.video_filter_builder import (
    ClipFilterChain,
    _escape_drawtext_text,
    _resolve_font_path,
+    build_broll_overlay_filter,
    build_clip_filter,
    build_concat_filter,
    build_filter_complex,
    build_title_drawtext_filter,
+    build_title_overlay_filter,
    build_xfade_filter,
    chain_filters,
    has_audio,
@@ -1161,5 +1164,97 @@ class TestDrawtextNotDictConfig(unittest.TestCase):
        self.assertIsNone(build_title_drawtext_filter([1, 2, 3]))


+class TestTitleOverlay(unittest.TestCase):
+    """build_title_overlay_filter 单元测试（WYSIWYG PNG 叠加路径）。"""
+
+    def test_overlay_filter_format(self):
+        """PNG 文件存在时返回正确的 overlay 滤镜字符串。"""
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
+            tmp.write(b"\x89PNG\r\n\x1a\n")
+            tmp_path = tmp.name
+        try:
+            result = build_title_overlay_filter(
+                {"text": "标题"},
+                output_width=720,
+                output_height=1280,
+                title_png_path=tmp_path,
+                title_input_label="[2:v]",
+                base_label="[vout]",
+                output_label="vout_titled",
+            )
+            self.assertIsNotNone(result)
+            self.assertIn("[vout][2:v]overlay=0:0[vout_titled]", result)
+        finally:
+            os.unlink(tmp_path)
+
+    def test_overlay_default_labels(self):
+        """不传 label 参数时使用默认 [0:v] / [1:v] / vout_titled。"""
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
+            tmp.write(b"\x89PNG\r\n\x1a\n")
+            tmp_path = tmp.name
+        try:
+            result = build_title_overlay_filter(
+                {"text": "标题"},
+                output_width=720,
+                output_height=1280,
+                title_png_path=tmp_path,
+            )
+            self.assertEqual(result, "[0:v][1:v]overlay=0:0[vout_titled]")
+        finally:
+            os.unlink(tmp_path)
+
+    def test_overlay_returns_none_when_png_missing(self):
+        """PNG 文件不存在时返回 None，供调用方降级到 drawtext。"""
+        result = build_title_overlay_filter(
+            {"text": "标题"},
+            output_width=720,
+            output_height=1280,
+            title_png_path="/nonexistent/path/title.png",
+        )
+        self.assertIsNone(result)
+
+    def test_overlay_returns_none_for_empty_config(self):
+        """title_config 为空/非 dict 时返回 None。"""
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp:
+            tmp.write(b"\x89PNG\r\n\x1a\n")
+            tmp_path = tmp.name
+        try:
+            self.assertIsNone(
+                build_title_overlay_filter(
+                    None,
+                    output_width=720,
+                    output_height=1280,
+                    title_png_path=tmp_path,
+                )
+            )
+            self.assertIsNone(
+                build_title_overlay_filter(
+                    "not a dict",
+                    output_width=720,
+                    output_height=1280,
+                    title_png_path=tmp_path,
+                )
+            )
+        finally:
+            os.unlink(tmp_path)
+
+    def test_overlay_returns_none_for_empty_path(self):
+        """title_png_path 为空字符串时返回 None。"""
+        self.assertIsNone(
+            build_title_overlay_filter(
+                {"text": "标题"},
+                output_width=720,
+                output_height=1280,
+                title_png_path="",
+            )
+        )
+
+
 if __name__ == "__main__":
    unittest.main()