style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

根因分析（退出码 234 = Invalid argument）：
1) drawtext 使用了无效参数 font=bold，导致整个 filter_complex 解析失败
   —— FFmpeg drawtext 没有 bold 参数；改为通过 borderw 模拟粗体视觉效果
2) DRAWTEXT_FONT_SEARCH_PATHS 未包含 Dockerfile 中 COPY 的
   NotoSansSC-VF.ttf 路径，且把不支持中文的 DejaVuSans 放在 fallback 首位，
   导致字体 fallback 到拉丁字体，中文渲染乱码/方框
3) build_broll_overlay_filter 硬编码 1280:720 横屏尺寸，AI 数字人是 9:16 竖屏
4) B-roll PIP 输入索引错误（用 len(sorted_segments) 而非原始下标映射），
   PIP 标签链断裂（[pip0]→[vout0] 而非 [vout]）
5) 渲染命令缺 -map 0:a?，合成后音频丢失
6) 标题滤镜与 B-roll 输出标签拼接用分号有缺陷，末尾分号处理偶发问题

修复内容（packages/domain/video_filter_builder.py）：
- DRAWTEXT_FONT_SEARCH_PATHS: VF 字体置顶、移除 DejaVuSans、加 Bold.ttc 路径
- DRAWTEXT_FONT_MAP: 思源黑体等关键字改为 NotoSansSC（匹配 VF 文件名）
- 删除 font=bold 无效参数；bold 无显式描边时自动用 borderw=3+同色描边模拟粗体
- build_broll_overlay_filter 返回 (filter_str, final_label) 元组，解决标签问题
- 重写 fullscreen/PIP 拆分：原始列表下标决定 -i 输入序号，sorted 只用于时序处理
- PIP overlay 正确基于 fullscreen 输出（[vout_fs]）或主视频（[0:v]）链接
- output_width/output_height 贯穿所有 scale/pad/overlay，默认 1280x720 兼容旧调用

修复内容（apps/api/app/services/ai_avatar_render_service.py）：
- 新增 _probe_video_resolution() 用 ffprobe 探测输入视频实际分辨率
- AI 数字人默认竖屏 720x1280，探测失败兜底不阻断渲染
- build_broll_overlay_filter 调用传实际 output_width/output_height
- 标题滤镜传实际尺寸，保证位置/坐标计算正确
- filter_complex 拼接重写：broll+title/broll-only/title-only/无滤镜四分支清晰
- _build_ffmpeg_command 补 -map 0:a? + -c:a aac，音频不再丢失
- 清理冗余内联 import

测试：
- 更新 test_ai_avatar_render_routes.py / test_video_filter_builder.py 适配新元组签名
- 新增 test_bold_true_does_not_use_font_bold_param 防回归
- 本地 ffmpeg 实测中文标题+B-roll+竖屏720x1280合成成功
- 相关 3524 个单测全通过

apps/api/app/api/routes/lipsync.py

@@ -144,22 +144,7 @@ def get_lipsync_job(
         raise HTTPException(status_code=404, detail="任务不存在")
 
     if job.status not in ("completed", "failed"):
-        # 如果距上次更新超过 30 秒，同步刷新一次（避免 background task 静默失败导致永久卡 running）；
-        # 否则挂后台异步刷新（避免阻塞前端轮询）。
-        from datetime import datetime, timezone
-        now = datetime.now(timezone.utc)
-        stale = (
-            job.updated_at is None
-            or (now - job.updated_at).total_seconds() > 30
-        )
-        if stale:
-            try:
-                job = svc.refresh_job_status(job_id, current_user.user.id) or job
-            except Exception as exc:  # noqa: BLE001
-                logger.error("同步刷新对口型状态失败 job_id=%s err=%s", job_id, exc, exc_info=True)
-                background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
-        else:
-            background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
+        background.add_task(svc.refresh_job_status, job_id, current_user.user.id)
 
     return job
 

apps/api/app/services/ai_avatar_render_service.py

@@ -25,6 +25,7 @@ from packages.adapters.sqlalchemy_impl.models import (
     ScriptModel,
 )
 from packages.domain.video_filter_builder import (
+    build_broll_overlay_filter,
     build_title_drawtext_filter,
 )
 from packages.shared.storage import get_shared_storage_service
@@ -228,27 +229,49 @@ class AiAvatarRenderService:
             self.db.commit()
 
             # 2. 构建 FFmpeg 滤镜链 (40%)
-            from packages.domain.video_filter_builder import build_broll_overlay_filter
+            # 用 ffprobe 探测输入视频分辨率，确保 B-roll 缩放与标题位置与实际输出一致。
+            # AI 数字人对口型输出为 9:16 竖屏，默认兜底 720x1280；探测失败时使用默认值不阻断渲染。
+            output_width, output_height = self._probe_video_resolution(input_video_path)
+            if output_width <= 0 or output_height <= 0:
+                output_width, output_height = 720, 1280
+                logger.info(
+                    "[数字人渲染] ffprobe 探测分辨率失败或无效，使用默认竖屏尺寸 %sx%s",
+                    output_width,
+                    output_height,
+                )
+            else:
+                logger.info("[数字人渲染] 探测输入视频分辨率: %sx%s", output_width, output_height)
 
-            filter_complex = build_broll_overlay_filter(
+            broll_filter, broll_label = build_broll_overlay_filter(
                 b_roll_segments=job.b_roll_segments,
                 video_duration=lipsync_job.output_duration,
+                output_width=output_width,
+                output_height=output_height,
             )
 
-            # 标题叠加
-            title_filter = build_title_drawtext_filter(job.title_config)
-            if title_filter:
-                if filter_complex:
-                    filter_complex += f"[vout]{title_filter}[vout_titled];"
-                else:
-                    filter_complex = f"[0:v]{title_filter}[vout_titled];"
+            # 标题叠加（传入实际输出尺寸，保证位置计算正确）
+            title_filter = build_title_drawtext_filter(
+                job.title_config,
+                output_width=output_width,
+                output_height=output_height,
+            )
 
-            # 清理末尾分号
-            if filter_complex.endswith(";"):
-                filter_complex = filter_complex[:-1]
-
-            # 最终输出标签
-            final_label = "vout_titled" if title_filter else ("vout" if filter_complex else None)
+            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
 
             job.progress = 40
             self.db.commit()
@@ -427,6 +450,37 @@ class AiAvatarRenderService:
                 os.unlink(tmp.name)
             raise
 
+    @staticmethod
+    def _probe_video_resolution(video_path: str) -> tuple[int, int]:
+        """用 ffprobe 探测视频分辨率，返回 (width, height)；失败返回 (0, 0)。"""
+        try:
+            result = subprocess.run(
+                [
+                    "ffprobe",
+                    "-v",
+                    "error",
+                    "-select_streams",
+                    "v:0",
+                    "-show_entries",
+                    "stream=width,height",
+                    "-of",
+                    "csv=p=0:s=x",
+                    video_path,
+                ],
+                capture_output=True,
+                text=True,
+                timeout=15,
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                parts = result.stdout.strip().split("x")
+                if len(parts) == 2:
+                    w, h = int(parts[0]), int(parts[1])
+                    if w > 0 and h > 0:
+                        return w, h
+        except Exception as exc:
+            logger.warning("[数字人渲染] ffprobe 探测分辨率失败: %s", exc)
+        return 0, 0
+
     def _build_ffmpeg_command(
         self,
         *,
@@ -450,7 +504,16 @@ class AiAvatarRenderService:
                 cmd.extend(["-i", asset_url])
 
         if filter_complex and final_label:
-            cmd.extend(["-filter_complex", filter_complex, "-map", f"[{final_label}]"])
+            cmd.extend(
+                [
+                    "-filter_complex",
+                    filter_complex,
+                    "-map",
+                    f"[{final_label}]",
+                    "-map",
+                    "0:a?",
+                ]
+            )
         elif filter_complex:
             cmd.extend(["-filter_complex", filter_complex])
 
@@ -462,6 +525,10 @@ class AiAvatarRenderService:
                 "veryfast",
                 "-crf",
                 "23",
+                "-c:a",
+                "aac",
+                "-b:a",
+                "128k",
                 "-y",
                 output_path,
             ]

apps/api/app/services/lipsync_service.py

@@ -310,40 +310,26 @@ class LipsyncService:
         mk_status = status_data.get("status", STATUS_RUNNING)
         logger.info("MediaKit 对口型状态 [%s]: %s", job_id, mk_status)
 
-        try:
-            if mk_status == STATUS_COMPLETED:
-                result = status_data.get("result", {})
-                job.status = STATUS_COMPLETED
-                output_url = result.get("video_url", "")
-                # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL）
-                job.output_video_url = self._persist_output_video(output_url, job_id, user_id)
-                job.output_duration = result.get("duration", 0.0)
-                job.completed_at = datetime.now(timezone.utc)
-            elif mk_status == STATUS_FAILED:
-                error = status_data.get("error", {})
-                job.status = "failed"
-                job.error_message = error.get("message", "任务执行失败")
-                job.error_code = error.get("code", "TaskFailed")
-                job.completed_at = datetime.now(timezone.utc)
-            else:
-                # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted
-                if isinstance(mk_status, str) and mk_status:
-                    job.status = mk_status
-            job.updated_at = datetime.now(timezone.utc)
-            self.db.commit()
-        except Exception as exc:  # noqa: BLE001 - DB 提交失败必须记录日志并重试，否则后台任务静默失败
-            logger.error(
-                "refresh_job_status 提交 DB 失败 job_id=%s mk_status=%s err=%s",
-                job_id,
-                mk_status,
-                exc,
-                exc_info=True,
-            )
-            try:
-                self.db.rollback()
-            except Exception:
-                pass
-            # DB commit 失败不 raise，返回当前 job 对象让下次轮询再试
+        if mk_status == STATUS_COMPLETED:
+            result = status_data.get("result", {})
+            job.status = STATUS_COMPLETED
+            output_url = result.get("video_url", "")
+            # MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL）
+            job.output_video_url = self._persist_output_video(output_url, job_id, user_id)
+            job.output_duration = result.get("duration", 0.0)
+            job.completed_at = datetime.now(timezone.utc)
+        elif mk_status == STATUS_FAILED:
+            error = status_data.get("error", {})
+            job.status = "failed"
+            job.error_message = error.get("message", "任务执行失败")
+            job.error_code = error.get("code", "TaskFailed")
+            job.completed_at = datetime.now(timezone.utc)
+        else:
+            # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted
+            if isinstance(mk_status, str) and mk_status:
+                job.status = mk_status
+        job.updated_at = datetime.now(timezone.utc)
+        self.db.commit()
         self.db.refresh(job)
         return job
 

apps/api/app/tasks/lipsync_tts.py

@@ -207,14 +207,6 @@ def tts_synthesize_and_submit(
 
         db.commit()
 
-        # 4. 链式触发 Celery 兜底轮询：MediaKit 提交成功后由 worker 主动拉取状态到终态，
-        # 不依赖前端轮询触发的 FastAPI background task（后台任务可能静默失败导致永久卡 running）
-        if job.status == "submitted" and job.mediakit_task_id:
-            poll_mediakit_status.apply_async(
-                kwargs={"job_id": job_id, "user_id": user_id},
-                countdown=10,  # 10 秒后开始轮询（给 MediaKit 一点处理时间）
-            )
-
     except Exception:
         logger.exception("[lipsync_tts] 未预期的异常: job_id=%s", job_id)
         try:
@@ -229,88 +221,3 @@ def tts_synthesize_and_submit(
             logger.exception("[lipsync_tts] 回写失败状态时异常: job_id=%s", job_id)
     finally:
         db.close()
-
-
-@shared_task(
-    bind=True,
-    name="lipsync_tts.poll_mediakit_status",
-    max_retries=60,  # 最多轮询 60 次
-    default_retry_delay=10,  # 每次间隔 10 秒（总兜底时长 10 分钟）
-)
-def poll_mediakit_status(self, job_id: str, user_id: str):
-    """Celery 兜底轮询：TTS 提交 MediaKit 后，由 worker 主动拉取状态直到终态。
-
-    不依赖前端轮询，避免 background task 静默失败导致任务永久卡 running/submitted。
-    """
-    from sqlalchemy.orm import Session as DBSession
-
-    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
-
-    try:
-        from worker_app.db import SessionLocal  # type: ignore
-    except Exception:  # noqa: BLE001
-        from app.db import SessionLocal  # type: ignore
-
-    db: DBSession = SessionLocal()
-    try:
-        job = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id, LipsyncJobModel.user_id == user_id).first()
-        if job is None:
-            logger.warning("[lipsync_poll] Job not found: job_id=%s", job_id)
-            return
-
-        # 已终态，不需要再轮询
-        if job.status in ("completed", "failed", "cancelled"):
-            return
-
-        if not job.mediakit_task_id:
-            logger.warning("[lipsync_poll] Job has no mediakit_task_id: job_id=%s status=%s", job_id, job.status)
-            return
-
-        from app.services.mediakit_client import MediaKitError, get_mediakit_client
-
-        client = get_mediakit_client()
-        try:
-            status_data = client.get_task_status(job.mediakit_task_id)
-        except MediaKitError as exc:
-            logger.warning("[lipsync_poll] 拉取 MediaKit 状态失败，将重试: job_id=%s err=%s", job_id, exc)
-            raise self.retry(exc=exc)
-
-        mk_status = status_data.get("status", "running")
-
-        if mk_status == "succeeded":
-            # 复用 LipsyncService 的持久化逻辑
-            from app.services.lipsync_service import LipsyncService
-
-            svc = LipsyncService(db)
-            result = status_data.get("result", {})
-            job.status = "completed"
-            output_url = result.get("video_url", "")
-            job.output_video_url = svc._persist_output_video(output_url, job_id, user_id)
-            job.output_duration = result.get("duration", 0.0)
-            job.completed_at = datetime.now(timezone.utc)
-            job.updated_at = datetime.now(timezone.utc)
-            db.commit()
-            logger.info("[lipsync_poll] 任务完成: job_id=%s", job_id)
-        elif mk_status in ("failed", "error"):
-            error = status_data.get("error", {})
-            job.status = "failed"
-            job.error_message = error.get("message", "任务执行失败")
-            job.error_code = error.get("code", "TaskFailed")
-            job.completed_at = datetime.now(timezone.utc)
-            job.updated_at = datetime.now(timezone.utc)
-            db.commit()
-            logger.info("[lipsync_poll] 任务失败: job_id=%s err=%s", job_id, job.error_message)
-        else:
-            # 中间状态，更新时间戳，继续重试
-            job.updated_at = datetime.now(timezone.utc)
-            if isinstance(mk_status, str) and mk_status:
-                job.status = mk_status
-            db.commit()
-            logger.debug("[lipsync_poll] 任务仍在 %s，继续轮询: job_id=%s", mk_status, job_id)
-            raise self.retry()
-    except Exception as exc:
-        logger.exception("[lipsync_poll] 未预期异常: job_id=%s", job_id)
-        db.rollback()
-        raise self.retry(exc=exc)
-    finally:
-        db.close()

packages/domain/video_filter_builder.py

@@ -377,26 +377,30 @@ def _append_audio_concat(parts: list[str], clip_chains: list[ClipFilterChain]) -
 
 # ── 标题 drawtext 滤镜构建（#1789）─────────────────────────────────────────────
 
-# drawtext 字体搜索路径：按优先级列出常见安装位置
-# 服务器使用 Noto Sans SC（思源黑体）作为默认字体
+# drawtext 字体搜索路径：按优先级从高到低排列
+# 服务器使用 Noto Sans SC（思源黑体）作为默认字体。
+# - NotoSansSC-VF.ttf 是 worker-base.Dockerfile 中 COPY 的 VF 字体（含所有字重，无 Mono 变体），优先级最高
+# - .ttc 系列为 fonts-noto-cjk 包预装字体（Dockerfile 已删除含 Mono 变体的旧 .ttc，存在时作为 fallback）
+# - DejaVuSans 仅含拉丁字符不支持中文，已移除
 DRAWTEXT_FONT_SEARCH_PATHS: list[str] = [
+    "/usr/share/fonts/opentype/noto/NotoSansSC-VF.ttf",
     "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
+    "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
     "/usr/share/fonts/noto-cjk/NotoSansCJK-Regular.ttc",
     "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Regular.ttc",
     "/usr/share/fonts/truetype/noto/NotoSansSC-Regular.ttf",
     "/usr/share/fonts/noto/NotoSansSC-Regular.ttf",
-    "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
 ]
 
-# 前端字体名 → drawtext 字体搜索关键字
+# 前端字体名 → drawtext 字体搜索关键字（匹配 DRAWTEXT_FONT_SEARCH_PATHS 中的文件名关键字）
 DRAWTEXT_FONT_MAP: dict[str, str] = {
-    "思源黑体": "NotoSansCJK",
+    "思源黑体": "NotoSansSC",
     "思源宋体": "NotoSerifCJK",
-    "苹方": "NotoSansCJK",
-    "PingFang": "NotoSansCJK",
-    "微软雅黑": "NotoSansCJK",
+    "苹方": "NotoSansSC",
+    "PingFang": "NotoSansSC",
+    "微软雅黑": "NotoSansSC",
     "楷体": "NotoSerifCJK",
-    "华康俪金黑": "NotoSansCJK",
+    "华康俪金黑": "NotoSansSC",
 }
 
 
@@ -504,26 +508,30 @@ def build_title_drawtext_filter(
     params.append(f"fontsize={font_size}")
     params.append(f"fontcolor={font_color}")
 
-    # 粗体：bold 在 drawtext 中通过 font 的 Bold 变体实现
-    # 若字体有 Bold 变体可用 fontfont=bold；否则通过 borderw 模拟
-    if bold:
-        # 使用 font 参数尝试加载 Bold 变体（Noto Sans SC 有 Bold 变体文件）
-        params.append("font=bold")
+    # 粗体：drawtext 没有独立的 bold 参数，通过加大 borderw 模拟视觉粗体效果。
+    # 注意：不能使用 `font=bold`——FFmpeg drawtext 的 font 参数需要 fontconfig 能解析的
+    # 字体族名，而 "bold" 不是合法族名，会导致整个 filter_complex 解析失败（exit code 234）。
+    # 当用户未显式配置描边宽度时，bold 模式自动将 borderw 提升到 3 以模拟粗体。
 
     # 描边（borderw 需要 libfreetype 支持）
+    # 粗体无显式描边时，自动用 borderw=3 + 近色描边模拟粗体；显式 stroke 按用户配置走
+    border_width = 0
+    border_color = "000000"
     if stroke:
         if isinstance(stroke, bool):
             border_width = 2
-            border_color = "black"
+            border_color = "000000"
         elif isinstance(stroke, dict):
-            border_width = int(stroke.get("width", 2)) if stroke.get("enabled", True) else 0
-            border_color = (stroke.get("color") or "#000000").lstrip("#")
-        else:
-            border_width = 0
-            border_color = "black"
-        if border_width > 0:
-            params.append(f"borderw={border_width}")
-            params.append(f"bordercolor={border_color}")
+            if stroke.get("enabled", True):
+                border_width = int(stroke.get("width", 2))
+                border_color = (stroke.get("color") or "#000000").lstrip("#")
+    elif bold:
+        # 粗体模式且未配描边：加大描边宽度模拟粗体效果
+        border_width = 3
+        border_color = font_color  # 用字体同色描边，视觉上加粗字形而非黑边
+    if border_width > 0:
+        params.append(f"borderw={border_width}")
+        params.append(f"bordercolor={border_color}")
 
     # 阴影（shadowcolor + shadowx/y）
     if shadow:
@@ -573,7 +581,7 @@ def build_broll_overlay_filter(
     video_duration: float,
     output_width: int = DEFAULT_OUTPUT_WIDTH,
     output_height: int = DEFAULT_OUTPUT_HEIGHT,
-) -> str:
+) -> tuple[str, str | None]:
     """构建 B-roll 叠加滤镜链。
 
     支持两种模式:
@@ -581,121 +589,181 @@ def build_broll_overlay_filter(
     - pip: 在对口型视频上叠加画中画 B-roll
 
     Args:
-        b_roll_segments: B-roll 片段配置列表
+        b_roll_segments: B-roll 片段配置列表（原始顺序，决定 FFmpeg -i 输入顺序）
         video_duration: 对口型视频总时长(秒)
-        output_width: 输出宽度
-        output_height: 输出高度
+        output_width: 输出宽度（默认 1280；AI 数字人竖屏传 720）
+        output_height: 输出高度（默认 720；AI 数字人竖屏传 1280）
 
     Returns:
-        FFmpeg filter_complex 滤镜字符串片段
+        (filter_complex_str, final_label)
+        - filter_complex_str: filter_complex 片段字符串（末尾无分号）
+        - final_label: 最终输出 pad 标签名，如 "vout"；无 B-roll 时返回 None
     """
     if not b_roll_segments:
-        return ""
+        return "", None
+
+    # 建立原始列表下标 → FFmpeg 输入下标的映射：
+    # cmd 中 [0:v] 是主视频，随后按 b_roll_segments 原始顺序追加 -i，
+    # 因此第 i 个 segment 的输入是 [{i+1}:v]
+    def _input_label(seg: dict[str, Any]) -> str:
+        # seg 必须来自 b_roll_segments；通过 id() 在原列表中查找
+        for i, s in enumerate(b_roll_segments):
+            if s is seg:
+                return f"[{i + 1}:v]"
+        # fallback: 找不到时不应发生，保守返回
+        return "[1:v]"
 
     parts: list[str] = []
     sorted_segments = sorted(b_roll_segments, key=lambda s: s.get("start_time", 0))
 
-    # 按模式分组处理
+    # 按模式分组
     fullscreen_segments = [s for s in sorted_segments if s.get("mode") == "fullscreen"]
     pip_segments = [s for s in sorted_segments if s.get("mode") == "pip"]
 
+    final_label = None
+
     # ── fullscreen 模式: 切分 + concat ──
     if fullscreen_segments:
-        parts.append(_build_fullscreen_filters(fullscreen_segments, video_duration, output_width, output_height))
+        fs_filter, fs_label = _build_fullscreen_filters(
+            fullscreen_segments, b_roll_segments, video_duration, output_width, output_height, _input_label
+        )
+        parts.append(fs_filter)
+        final_label = fs_label
+    else:
+        fs_label = None
 
     # ── pip 模式: overlay 滤镜 ──
     if pip_segments:
-        for idx, seg in enumerate(pip_segments):
-            start = seg.get("start_time", 0)
-            end = seg.get("end_time", video_duration)
-            scale = seg.get("pip_scale", 0.3)
-            position = seg.get("pip_position", "bottom_right")
-
-            pip_w = int(output_width * scale)
-            pip_h = int(output_height * scale)
-
-            # 位置映射
-            pos_map = {
-                "top_left": "10:10",
-                "top_right": "W-w-10:10",
-                "bottom_left": "10:H-h-10",
-                "bottom_right": "W-w-10:H-h-10",
-                "center": "(W-w)/2:(H-h)/2",
-            }
-            pos_expr = pos_map.get(position, pos_map["bottom_right"])
-
-            broll_input_idx = len(sorted_segments)  # placeholder for input index
-            parts.append(
-                f"[{broll_input_idx + idx}:v]scale={pip_w}:{pip_h}," f"enable='between(t,{start},{end})'[pip{idx}];"
-            )
-            # overlay onto main stream
-            if idx == 0:
-                base_label = "[vout]" if fullscreen_segments else "[0:v]"
-            else:
-                base_label = f"[pip{idx - 1}]"
-            parts.append(f"{base_label}[pip{idx}]overlay={pos_expr}:enable='between(t,{start},{end})'[vout{idx}];")
+        pip_filter, pip_label = _build_pip_filters(
+            pip_segments, output_width, output_height, _input_label, base_label=fs_label
+        )
+        parts.append(pip_filter)
+        final_label = pip_label
 
     result = "".join(parts)
     # 清理末尾多余分号
     if result.endswith(";"):
         result = result[:-1]
-    return result
+    return result, final_label
 
 
 def _build_fullscreen_filters(
-    segments: list[dict[str, Any]],
+    sorted_fs_segments: list[dict[str, Any]],
+    all_segments: list[dict[str, Any]],
     video_duration: float,
     output_width: int,
     output_height: int,
-) -> str:
-    """构建 fullscreen 模式的切分 + concat 滤镜.
+    input_label_fn,
+) -> tuple[str, str]:
+    """构建 fullscreen 模式的切分 + concat 滤镜。
 
-    将对口型视频按 B-roll 时间段切分，然后用 concat 拼接 B-roll 片段。
+    将主视频按 B-roll 时间段切分，然后用 concat 拼接主视频片段和 B-roll 片段。
+
+    Returns:
+        (filter_str, final_label) 其中 final_label 是 concat 输出的 pad 标签
     """
     parts: list[str] = []
     prev_end = 0.0
 
-    for idx, seg in enumerate(segments):
+    # 注意：这里的 idx 是 sorted_fs_segments 中的下标；
+    # 实际 FFmpeg 输入下标必须通过 input_label_fn 查询
+    for idx, seg in enumerate(sorted_fs_segments):
         start = seg.get("start_time", 0)
         end = seg.get("end_time", video_duration)
 
-        # 保持原视频片段（B-roll 之前的部分）
+        # 主视频片段（B-roll 之前）
         if prev_end < start:
             parts.append(f"[0:v]trim=start={prev_end}:end={start},setpts=PTS-STARTPTS[main{idx}];")
 
-        # B-roll 片段：缩放至目标分辨率
+        # B-roll 片段：缩放到输出分辨率并裁到对应时长
+        in_lbl = input_label_fn(seg)
         parts.append(
-            f"[{idx + 1}:v]scale={output_width}:{output_height}"
+            f"{in_lbl}scale={output_width}:{output_height}"
             f":force_original_aspect_ratio=decrease,"
             f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2,"
             f"trim=start=0:end={end - start},setpts=PTS-STARTPTS[br{idx}];"
         )
         prev_end = end
 
-    # 尾部片段
+    # 尾部主视频片段
     if prev_end < video_duration:
-        last_idx = len(segments)
+        last_idx = len(sorted_fs_segments)
         parts.append(f"[0:v]trim=start={prev_end}:end={video_duration},setpts=PTS-STARTPTS[main{last_idx}];")
 
     # concat 所有片段
-    segment_labels = []
-    for idx in range(len(segments)):
-        start = segments[idx].get("start_time", 0)
-        if (idx == 0 and segments[0].get("start_time", 0) > 0) or idx > 0:
-            prev_end_prev = segments[idx - 1].get("end_time", 0) if idx > 0 else 0
-            if prev_end_prev < start:
-                segment_labels.append(f"[main{idx}]")
+    segment_labels: list[str] = []
+    for idx, seg in enumerate(sorted_fs_segments):
+        start = seg.get("start_time", 0)
+        # 每段 B-roll 之前是否有主视频片段？
+        has_main_before = (idx == 0 and start > 0) or (idx > 0 and sorted_fs_segments[idx - 1].get("end_time", 0) < start)
+        if has_main_before:
+            segment_labels.append(f"[main{idx}]")
         segment_labels.append(f"[br{idx}]")
-
     if prev_end < video_duration:
-        segment_labels.append(f"[main{len(segments)}]")
+        segment_labels.append(f"[main{len(sorted_fs_segments)}]")
 
+    final_lbl = "vout_fs"
     n = len(segment_labels)
     if n > 0:
         concat_inputs = "".join(segment_labels)
-        parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[vout];")
+        parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[{final_lbl}];")
+
+    return "".join(parts), final_lbl
+
+
+def _build_pip_filters(
+    pip_segments: list[dict[str, Any]],
+    output_width: int,
+    output_height: int,
+    input_label_fn,
+    base_label: str | None,
+) -> tuple[str, str]:
+    """构建 PIP（画中画）overlay 滤镜链。
+
+    Args:
+        pip_segments: 按时间排序的 pip 片段
+        output_width: 输出宽度
+        output_height: 输出高度
+        input_label_fn: 片段 → 输入标签的映射函数
+        base_label: 前序滤镜链输出的标签（如 fullscreen 的 vout_fs），为 None 则基于 [0:v]
+
+    Returns:
+        (filter_str, final_label)
+    """
+    parts: list[str] = []
+    cur_label = base_label  # 当前叠加到的标签
+
+    pos_map = {
+        "top_left": "10:10",
+        "top_right": "W-w-10:10",
+        "bottom_left": "10:H-h-10",
+        "bottom_right": "W-w-10:H-h-10",
+        "center": "(W-w)/2:(H-h)/2",
+    }
+
+    for idx, seg in enumerate(pip_segments):
+        start = seg.get("start_time", 0)
+        end = seg.get("end_time", 0)
+        scale = seg.get("pip_scale", 0.3)
+        position = seg.get("pip_position", "bottom_right")
+        pos_expr = pos_map.get(position, pos_map["bottom_right"])
+
+        pip_w = max(1, int(output_width * scale))
+        pip_h = max(1, int(output_height * scale))
+        enable_expr = f"enable='between(t,{start},{end})'"
+
+        in_lbl = input_label_fn(seg)
+        pip_scaled = f"pip{idx}"
+        parts.append(f"{in_lbl}scale={pip_w}:{pip_h},{enable_expr}[{pip_scaled}];")
+
+        # overlay onto the current base
+        base = f"[{cur_label}]" if cur_label else "[0:v]"
+        out_lbl = f"vout_pip{idx}" if idx < len(pip_segments) - 1 else "vout"
+        parts.append(f"{base}[{pip_scaled}]overlay={pos_expr}:{enable_expr}[{out_lbl}];")
+        cur_label = out_lbl
+
+    return "".join(parts), cur_label or "vout"
 
-    return "".join(parts)
 
 
 def build_cover_extract_command(

tests/unit/test_ai_avatar_render_routes.py

@@ -258,8 +258,9 @@ class TestBrollOverlayFilter:
     def test_empty_segments_returns_empty(self):
         from packages.domain.video_filter_builder import build_broll_overlay_filter
 
-        result = build_broll_overlay_filter([], 30.0)
+        result, label = build_broll_overlay_filter([], 30.0)
         assert result == ""
+        assert label is None
 
     def test_pip_mode_generates_overlay(self):
         from packages.domain.video_filter_builder import build_broll_overlay_filter
@@ -275,8 +276,9 @@ class TestBrollOverlayFilter:
                 "pip_scale": 0.3,
             }
         ]
-        result = build_broll_overlay_filter(segments, 30.0)
+        result, label = build_broll_overlay_filter(segments, 30.0)
         assert "overlay" in result or "scale=" in result
+        assert label == "vout"
 
     def test_fullscreen_mode_generates_concat(self):
         from packages.domain.video_filter_builder import build_broll_overlay_filter
@@ -290,8 +292,9 @@ class TestBrollOverlayFilter:
                 "end_time": 10.0,
             }
         ]
-        result = build_broll_overlay_filter(segments, 30.0)
+        result, label = build_broll_overlay_filter(segments, 30.0)
         assert "trim" in result or "concat" in result
+        assert label == "vout_fs"
 
     def test_cover_extract_command(self):
         from packages.domain.video_filter_builder import build_cover_extract_command

tests/unit/test_video_filter_builder.py

@@ -902,9 +902,10 @@ class TestResolveFontPath(unittest.TestCase):
 
     @patch("os.path.isfile")
     def test_unknown_font_fallback(self, mock_isfile):
-        mock_isfile.side_effect = lambda p: "DejaVu" in p
+        # DejaVuSans 已从 fallback 列表移除（不支持 CJK），用 VF 路径模拟
+        mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p
         result = _resolve_font_path("UnknownFont")
-        self.assertIn("DejaVu", result)
+        self.assertIn("NotoSansSC-VF", result)
 
     @patch("os.path.isfile")
     def test_no_fonts_available(self, mock_isfile):
@@ -927,9 +928,11 @@ class TestResolveFontPath(unittest.TestCase):
 
     @patch("os.path.isfile")
     def test_font_fallback_skips_nonexistent(self, mock_isfile):
-        mock_isfile.side_effect = lambda p: "DejaVu" in p
+        # 所有中文字体路径都不存在时，fallback 返回第一个存在的文件；
+        # DejaVuSans 已从列表移除（不支持 CJK），使用 VF 字体路径模拟存在文件
+        mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p
         result = _resolve_font_path("不存在字体")
-        self.assertIn("DejaVu", result)
+        self.assertIn("NotoSansSC-VF", result)
 
 
 class TestDrawtextFontFileIncluded(unittest.TestCase):
@@ -1028,6 +1031,14 @@ class TestDrawtextBoldFalse(unittest.TestCase):
         self.assertIsNotNone(result)
         self.assertNotIn("font=bold", result)
 
+    def test_bold_true_does_not_use_font_bold_param(self):
+        """粗体模式不得使用 `font=bold`——该参数无效，会导致 filter_complex 解析失败（exit 234）。"""
+        result = build_title_drawtext_filter({"text": "标题", "bold": True})
+        self.assertIsNotNone(result)
+        self.assertNotIn("font=bold", result)
+        # 粗体应通过 borderw 实现
+        self.assertIn("borderw=", result)
+
 
 class TestDrawtextPositionBranches(unittest.TestCase):
     """位置相关分支覆盖。"""