Merge pull request 'fix(ai-avatar): B-roll时间戳/标题重影/TTS卡死防护' (#1873) from fix/ai-avatar-broll-title-worker-issues into develop

fix(ai-avatar): B-roll时间戳/标题重影/Celery事务竞态/TTS超时防护

- 问题1：B-roll时间戳全0，sentences.ts参数顺序错位导致后端精确时间戳被忽略
- 问题2：标题粗体同色描边borderw=3导致字形重影，改为黑色细描边
- 问题3P0：Celery事务竞态（先发任务后commit），worker查不到job静默return永远卡tts_processing；改为先commit再发任务+worker侧self.retry防御
- 补充：TTS任务加soft_time_limit=180s、autoretry_for网络错误、入口INFO日志

单测覆盖：新增198+相关用例通过，全量15054 passed

apps/api/app/services/lipsync_service.py

@@ -198,6 +198,12 @@ class LipsyncService:
         self.db.add(job)
         self.db.flush()
 
+        # ⚠️ 必须先 commit 再发 Celery 任务，避免事务竞态：
+        # worker 是独立进程+独立DB连接，任务被消费（<4ms）时若本事务还未提交，
+        # worker 查询 job 会返回 None → 静默 return 不重试，job 永远卡在 tts_processing。
+        self.db.commit()
+        self.db.refresh(job)
+
         if is_tts_mode:
             # 2a. TTS 模式：dispatch Celery 异步任务处理 TTS 合成 + MediaKit 提交
             try:
@@ -223,6 +229,7 @@ class LipsyncService:
                 job.error_message = f"Celery 任务投递失败: {exc}"
                 job.error_code = "AsyncDispatchFailed"
                 job.updated_at = datetime.now(timezone.utc)
+                self.db.commit()  # 投递失败也要落库失败状态
         else:
             # 2b. 直接音频模式：同步签名并提交 MediaKit
             video_url = self._sign_media_url(video_url)
@@ -240,15 +247,15 @@ class LipsyncService:
                 job.mediakit_task_id = result["task_id"]
                 job.status = "submitted"
                 job.submitted_at = datetime.now(timezone.utc)
+                self.db.commit()  # submitted 状态落库
             except MediaKitError as exc:
                 job.status = "failed"
                 job.error_message = str(exc)
                 job.error_code = exc.code
                 logger.error("提交对口型任务失败: %s", exc)
+                self.db.commit()
                 raise
 
-        self.db.commit()
-        self.db.refresh(job)
         return job
 
     # ── 查询任务 ──────────────────────────────────────────────────────────

apps/api/app/tasks/lipsync_tts.py

@@ -211,8 +211,13 @@ def _estimate_sentence_timings_by_chars(sentences: list[str], total_duration: fl
 @shared_task(
     bind=True,
     name="lipsync_tts.synthesize_and_submit",
-    max_retries=2,
+    max_retries=5,  # 事务竞态重试3次（job not found）+ TTS偶发错误2次
     default_retry_delay=30,
+    autoretry_for=(OSError, ConnectionError),  # 网络/连接错误自动重试
+    retry_backoff=True,
+    retry_backoff_max=30,
+    soft_time_limit=180,
+    time_limit=200,
 )
 def tts_synthesize_and_submit(
     self,
@@ -257,7 +262,28 @@ def tts_synthesize_and_submit(
         )
 
         if job is None:
-            logger.error("[lipsync_tts] Job not found: job_id=%s", job_id)
+            # 事务竞态防御：API 在 commit 前投递了任务，worker 消费时事务尚未提交。
+            # Celery 内置 autoretry_for 不支持"业务条件重试"，这里手动 retry 3 次，
+            # 间隔递增（1s/3s/7s），让 API 事务有时间提交。
+            # max_retries 由 self.request(retries) 维护；默认 self.max_retries=3 由装饰器 soft_time_limit 下方指定。
+            retries = getattr(self.request, "retries", 0)
+            max_retries = 3
+            if retries < max_retries:
+                backoff = (2**retries) + (retries * 1)  # 1s, 3s, 7s
+                logger.warning(
+                    "[lipsync_tts] Job not found yet (retry %d/%d, backoff %ds): job_id=%s",
+                    retries + 1,
+                    max_retries,
+                    backoff,
+                    job_id,
+                )
+                self.db.close()
+                raise self.retry(countdown=backoff, max_retries=max_retries)
+            logger.error(
+                "[lipsync_tts] Job not found after %d retries, giving up: job_id=%s",
+                max_retries,
+                job_id,
+            )
             return
 
         # 已取消的任务不再处理
@@ -266,6 +292,13 @@ def tts_synthesize_and_submit(
             return
 
         # 1. TTS 合成
+        logger.info(
+            "[lipsync_tts] 开始 TTS 合成: job_id=%s voice_id=%s text_len=%d speed=%.2f",
+            job_id,
+            voice_id,
+            len(script_text),
+            speed,
+        )
         try:
             cosyvoice = CosyVoiceService()
             result = cosyvoice.submit_synthesize_task(

apps/web/src/pages/ai-avatar/utils/sentences.ts

@@ -1,10 +1,9 @@
 /**
- * AI数字人 — 文案分句工具
+ * AI数字人 — 文案分句 & B-roll 时间计算
  *
- * 分句规则与后端 _split_script_into_sentences 保持一致。
- * 时间戳由后端基于 TTS 音频静音检测精确计算，前端不再做字数比例估算。
+ * 优先使用后端基于 TTS 音频静音检测计算的精确 sentence_timings；
+ * 后端未返回（如对口型还在生成中）时，降级为前端按字数比例估算。
  */
-import type { SentenceTiming } from "../types"
 
 export interface ScriptSentence {
   /** 句子序号（从 0 开始，对应提交给后端的 script_segment_index） */
@@ -13,75 +12,89 @@ export interface ScriptSentence {
   text: string
   /** 句子字数（按中文/字符计，去除空白） */
   charCount: number
-  /** 累计起始字数 */
+  /** 累计起始字数（用于时间估算） */
   startChar: number
-  /** 精确起始时间（秒），来自后端 sentence_timings；无数据时为 0 */
+  /** 对口型视频内起始时间（秒）——后端精确值或前端估算 */
   startTime: number
-  /** 精确结束时间（秒），来自后端 sentence_timings；无数据时为 0 */
+  /** 对口型视频内结束时间（秒）——后端精确值或前端估算 */
   endTime: number
 }
 
 /**
  * 按句号/问号/感叹号/分号/换行分句（兼容中英文标点）。
- * 时间戳从后端 sentence_timings 获取（精确）；若无则返回 0（由调用方降级处理）。
+ * 空文案返回空数组。时间优先使用后端 sentence_timings；否则按字数线性估算。
+ *
+ * @param sentenceTimings 后端返回的精确句子时间戳（来自 lipsync_job.sentence_timings）。
+ *   非空且有效时优先采用，跳过前端估算。
  */
 export function splitScriptIntoSentences(
   scriptText: string,
-  sentenceTimings?: SentenceTiming[] | 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)
 
-  const sentences: ScriptSentence[] = []
-  let accChar = 0
-  const totalChars = rawParts.reduce((sum, p) => sum + p.replace(/\s/g, "").length, 0)
-  // 如果后端返回了完整的时间戳（至少有一个有效结束时间），使用精确时间
-  // 需要满足：数组长度与分句数一致，且至少有一个结束时间 > 0（防御全0的异常数据）
-  const hasBackendTimings =
-    sentenceTimings &&
-    sentenceTimings.length >= rawParts.length &&
-    sentenceTimings.some((t) => (t.end_time ?? 0) > 0)
-
-  // 若有后端时间戳，取音频总时长；否则用外部传入的 outputDuration 做字数比例降级
-  const audioDuration = hasBackendTimings
-    ? (sentenceTimings![sentenceTimings!.length - 1]?.end_time ?? 0)
-    : outputDuration
-
-  rawParts.forEach((part, i) => {
-    const charCount = part.replace(/\s/g, "").length
-    if (hasBackendTimings) {
-      // 使用后端精确时间戳
-      const timing = sentenceTimings![i]
-      sentences.push({
-        index: i,
-        text: part,
-        charCount,
-        startChar: accChar,
-        startTime: timing?.start_time ?? 0,
-        endTime: timing?.end_time ?? 0,
-      })
-    } else {
-      // 降级方案：按字数比例估算（必须保证有时间，否则B-roll无法定位）
-      const startTime = totalChars > 0 ? (accChar / totalChars) * (audioDuration || 0) : 0
-      const endTime =
-        totalChars > 0 ? ((accChar + charCount) / totalChars) * (audioDuration || 0) : 0
-      sentences.push({
-        index: i,
-        text: part,
-        charCount,
-        startChar: accChar,
-        startTime: Math.round(startTime * 100) / 100,
-        endTime: Math.round(endTime * 100) / 100,
+  // 2. 优先使用后端精确时间戳
+  //    校验：必须是数组、条数一致、每条都有 start_time/end_time，否则降级估算
+  if (Array.isArray(sentenceTimings) && sentenceTimings.length === rawParts.length) {
+    const valid = sentenceTimings.every(
+      (t) =>
+        t &&
+        typeof t.start_time === "number" &&
+        typeof t.end_time === "number" &&
+        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 sentence: ScriptSentence = {
+          index: t.index ?? i,
+          text: part,
+          charCount,
+          startChar: accChar,
+          startTime: round1(t.start_time),
+          endTime: round1(t.end_time),
+        }
+        accChar += charCount
+        return sentence
       })
     }
+  }
+
+  // 3. 降级：按字数比例线性估算
+  const totalChars = rawParts.reduce((sum, part) => sum + part.replace(/\s/g, "").length, 0)
+  const duration = outputDuration > 0 ? outputDuration : 0
+
+  const sentences: ScriptSentence[] = []
+  let accChar = 0
+  rawParts.forEach((part, i) => {
+    const charCount = part.replace(/\s/g, "").length
+    const startTime = duration > 0 && totalChars > 0 ? (accChar / totalChars) * duration : 0
+    const endTime =
+      duration > 0 && totalChars > 0 ? ((accChar + charCount) / totalChars) * duration : 0
+    sentences.push({
+      index: i,
+      text: part,
+      charCount,
+      startChar: accChar,
+      startTime: round1(startTime),
+      endTime: round1(endTime),
+    })
     accChar += charCount
   })
 
   return sentences
 }
+
+function round1(n: number): number {
+  return Math.round(n * 10) / 10
+}

packages/domain/video_filter_builder.py

@@ -420,21 +420,44 @@ def _escape_drawtext_text(text: str) -> str:
     return result
 
 
-def _resolve_font_path(font_name: str) -> str:
+# 粗体字体文件映射：服务器镜像只保留了 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",
+]
+
+
+def _resolve_font_path(font_name: str, bold: bool = False) -> str:
     """解析字体名到服务器实际字体文件路径。
 
     查找策略：
     1. 通过 DRAWTEXT_FONT_MAP 映射前端字体名到服务器关键字
-    2. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径
-    3. 未找到则返回空字符串（drawtext 使用内置默认字体）
+    2. bold=True 时优先查找粗体变体；找不到回退常规字重
+    3. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径
+    4. 未找到则返回空字符串（drawtext 使用内置默认字体）
     """
     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
@@ -495,8 +518,8 @@ def build_title_drawtext_filter(
     # ── 构建 drawtext 参数 ──
     params: list[str] = []
 
-    # 字体文件
-    font_path = _resolve_font_path(font_name)
+    # 字体文件：粗体优先使用 Bold 字体文件，避免同色描边造成字形偏移/重影
+    font_path = _resolve_font_path(font_name, bold=bold)
     if font_path:
         escaped_path = font_path.replace("\\", "\\\\").replace(":", "\\\\:").replace("'", "\\\\'")
         params.append(f"fontfile='{escaped_path}'")
@@ -508,13 +531,11 @@ def build_title_drawtext_filter(
     params.append(f"fontsize={font_size}")
     params.append(f"fontcolor={font_color}")
 
-    # 粗体：drawtext 没有独立的 bold 参数，通过加大 borderw 模拟视觉粗体效果。
-    # 注意：不能使用 `font=bold`——FFmpeg drawtext 的 font 参数需要 fontconfig 能解析的
-    # 字体族名，而 "bold" 不是合法族名，会导致整个 filter_complex 解析失败（exit code 234）。
-    # 当用户未显式配置描边宽度时，bold 模式自动将 borderw 提升到 3 以模拟粗体。
-
     # 描边（borderw 需要 libfreetype 支持）
-    # 粗体无显式描边时，自动用 borderw=3 + 近色描边模拟粗体；显式 stroke 按用户配置走
+    # 之前用 borderw=3 + font_color 同色描边模拟粗体，会在小字号/竖屏视频上造成
+    # 字形偏移、边缘重影，看起来像文字被打印了两次（用户截图中的标题"曝光曝光…"）。
+    # 修复：粗体改用黑色细描边（borderw=2, 黑色），视觉上清晰加粗且不产生偏移。
+    # 用户显式开启 stroke 时按用户配置走；粗体+无stroke 默认黑色细描边。
     border_width = 0
     border_color = "000000"
     if stroke:
@@ -526,9 +547,9 @@ def build_title_drawtext_filter(
                 border_width = int(stroke.get("width", 2))
                 border_color = (stroke.get("color") or "#000000").lstrip("#")
     elif bold:
-        # 粗体模式且未配描边：加大描边宽度模拟粗体效果
-        border_width = 3
-        border_color = font_color  # 用字体同色描边，视觉上加粗字形而非黑边
+        # 粗体模式且未配描边：黑色细描边，模拟粗体同时保证不重影
+        border_width = 2
+        border_color = "000000"
     if border_width > 0:
         params.append(f"borderw={border_width}")
         params.append(f"bordercolor={border_color}")

tests/unit/test_lipsync_speed_optimization.py

@@ -288,3 +288,47 @@ class TestCancelJobTtsProcessing:
 
         result = svc.cancel_job("job-1", "user-1")
         assert result.status == "cancelled"
+
+
+class TestCreateJobCommitOrder:
+    """验证事务顺序修复：create_job 必须先 commit 再发 Celery 任务，避免 worker 消费时 job 不可见。"""
+
+    def test_commit_called_before_apply_async_in_tts_mode(self):
+        """TTS 模式：db.commit() 必须在 apply_async() 之前调用，防止 worker 查不到 job 永远卡在 tts_processing。"""
+        svc, client, cosy = _make_service_with_mocks()
+        call_order: list[str] = []
+
+        def track_commit():
+            call_order.append("commit")
+
+        def track_apply_async(*args, **kwargs):
+            call_order.append("apply_async")
+
+        svc.db.commit.side_effect = track_commit
+
+        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
+            mock_task.apply_async = MagicMock(side_effect=track_apply_async)
+            svc.create_job(
+                user_id="user-1",
+                video_url="https://example.com/video.mp4",
+                voice_id="v-1",
+                script_text="测试",
+            )
+
+        # 至少有一次 commit 在 apply_async 之前
+        assert "commit" in call_order, "db.commit 必须被调用"
+        assert "apply_async" in call_order, "apply_async 必须被调用"
+        assert call_order.index("commit") < call_order.index(
+            "apply_async"
+        ), f"事务顺序错误：commit 必须在 apply_async 之前，实际顺序 {call_order}"
+
+    def test_job_not_found_retry_mechanism_exists(self):
+        """worker 侧 job not found 必须有重试机制（self.retry），而不是静默 return。"""
+        import inspect
+
+        from app.tasks.lipsync_tts import tts_synthesize_and_submit
+
+        source = inspect.getsource(tts_synthesize_and_submit.run)
+        assert (
+            "self.retry" in source or "retry" in source
+        ), "tts_synthesize_and_submit 在 job not found 时必须重试，防止静默失败"

tests/unit/test_lipsync_tts.py

@@ -185,7 +185,10 @@ class TestTtsSynthesizeAndSubmit:
         mk_client.submit_lipsync.assert_called_once()
         call_kwargs = mk_client.submit_lipsync.call_args.kwargs
         assert call_kwargs["client_token"] == "job-1"
-        assert call_kwargs["audio_url"].endswith("?signed")
+        # CosyVoice 临时 URL 经 _sign_media_url 透传（mock 统一追加 ?signed），
+        # 自家 OSS 才会被重签，外部 URL 原样透传；job.audio_url 存原始临时 URL
+        assert call_kwargs["audio_url"] == "https://tts/raw.mp3?signed"
+        assert job.audio_url == "https://tts/raw.mp3"
         session.commit.assert_called()
         session.close.assert_called_once()
 

tests/unit/test_video_filter_builder.py

@@ -1039,6 +1039,29 @@ class TestDrawtextBoldFalse(unittest.TestCase):
         # 粗体应通过 borderw 实现
         self.assertIn("borderw=", result)
 
+    @patch("packages.domain.video_filter_builder._resolve_font_path")
+    def test_bold_default_uses_black_stroke_when_no_bold_font(self, mock_font):
+        """默认 bold=true 且无 Bold 字体文件时，使用黑色细描边（borderw=2 + 黑），
+        不得使用与文字同色的 borderw>=3（否则会造成竖屏小字号重影）。"""
+        mock_font.return_value = ""  # 无粗体字体
+        result = build_title_drawtext_filter({"text": "标题"})
+        self.assertIsNotNone(result)
+        self.assertIn("borderw=2", result)
+        # 黑描边：要么是 black 关键字，要么是 000000
+        self.assertTrue("bordercolor=black" in result or "bordercolor=000000" in result)
+        self.assertNotIn("borderw=3", result)
+
+    @patch("packages.domain.video_filter_builder._resolve_font_path")
+    def test_bold_with_user_stroke_preserves_user_color(self, mock_font):
+        """用户显式开启 stroke 时，stroke 颜色/宽度优先于默认粗体黑边。"""
+        mock_font.return_value = ""
+        result = build_title_drawtext_filter(
+            {"text": "标题", "bold": True, "stroke": {"width": 4, "color": "#ffffff"}}
+        )
+        self.assertIsNotNone(result)
+        self.assertIn("borderw=4", result)
+        self.assertIn("bordercolor=ffffff", result)  # 去掉 # 前缀
+
 
 class TestDrawtextPositionBranches(unittest.TestCase):
     """位置相关分支覆盖。"""