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

fix(lipsync): #2076 预合成模式补齐 script_text，修复 Ditto 任务 script_len=0
根因：前端 #1875(053b0063) 起预合成 payload 不带 script_text， job.script_text 落库为空 → Ditto 任务 script_len=0 → 情绪分析被跳过。修复（双保险）： - 前端 AiAvatarPage 预合成 payload 补传 script_text: state.scriptText - 后端 create_job 预合成模式 script_text 为空时从 sentence_timings 拼接兜底（覆盖所有调用方；timings 每项自带 text 字段）与 PR#2292 无关（2292 只动 build_timeline/intensity 内部逻辑）。测试：新增 3 用例（空时拼接兜底/显式传入保留原文/空 timings 退化直接音频）， test_lipsync_routes.py 45 passed
2026-10-15 09:32:12 +00:00 · 2026-10-15 17:19:45 +08:00 · 2026-10-15 16:58:49 +08:00 · 2026-10-15 14:19:02 +08:00 · 2026-10-15 13:15:59 +08:00 · 2026-10-15 12:22:28 +08:00
15 changed files with 1244 additions and 272 deletions
@@ -623,6 +623,12 @@ class LipsyncService:
            # 预合成模式：script_text 可空（因为 timings 已自带句子文本），但仍建议传
            if not isinstance(sentence_timings, list) or len(sentence_timings) == 0:
                raise MediaKitError("预合成模式 sentence_timings 不能为空", code="InvalidInput")
+            # 文案为空时从 sentence_timings 拼接兜底（Ditto 情绪分析依赖 script_text；
+            # 前端 #1875 起预合成 payload 不带 script_text，导致 job.script_text 落库为空）
+            if not (script_text or "").strip():
+                script_text = "".join(
+                    str(item.get("text") or "") for item in sentence_timings if isinstance(item, dict)
+                ).strip()

        # 0.5 lip_sync 计费预扣（全局 points 开关 + 功能开关均开才扣）
        prepaid_credits = 0.0
@@ -4,17 +4,57 @@
    <meta charset="UTF-8" />
    <link rel="icon" type="image/svg+xml" href="/vite.svg" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
-    <!-- 标题字体（#2001 / #font-selection 修复）：Google Fonts CDN 引入中文字体，保证优设标题黑/抖音美好体/阿里普惠体等fallback可用 -->
-    <link rel="preconnect" href="https://fonts.googleapis.com" />
-    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin />
-    <link
-      href="https://fonts.googleapis.com/css2?family=Noto+Sans+SC:wght@400;500;700;900&family=Noto+Serif+SC:wght@400;700;900&family=ZCOOL+KuaiLe&family=ZCOOL+XiaoWei&family=ZCOOL+QingKe+HuangYou&family=Ma+Shan+Zheng&family=Long+Cang&family=Liu+Jian+Mao+Cao&family=Zhi+Mang+Xing&display=swap"
-      rel="stylesheet"
-    />
+    <!-- 标题字体（#2001 / #font-selection 修复）：Google Fonts 中文字体完全延迟加载
+         关键设计：所有 Google Fonts 子资源（CSS + 字体文件）必须在 window load 之后才发起请求。
+         - 原因：fonts.googleapis.com/gstatic.com 在境内访问间歇性挂起，若在关键路径内
+           （<link rel=stylesheet>/preload/preconnect 等任何会被计入文档子资源完成的形式），
+           会拖死 window.load 事件，导致 Playwright page.goto 默认 waitUntil:'load' 90s 超时，
+           同时真实用户首屏也可能因等字体而白屏。
+         - 实现：head 里完全不出现 Google 域的 <link>，首屏用系统字体兜底
+           （font-family fallback 链：Noto Sans SC → PingFang SC → Microsoft YaHei → sans-serif）；
+           页面 load 后通过 requestIdleCallback 动态注入 <link rel=stylesheet>，字体异步加载生效。
+         - noscript 降级：JS 禁用时仍然阻塞加载（可接受，这类用户极少）。
+         - 参考：运维 2026-10-14 E2E 失败根因分析，#2300 preload/media=print 方案不治本，
+           preload 与 media=print 样式表仍计入子资源完成，会阻塞 load 事件。 -->
    <title>小虾 SaaS - 自动化视频剪辑平台</title>
  </head>
  <body>
    <div id="root"></div>
    <script type="module" src="/src/main.tsx"></script>
+    <!-- Google Fonts 延迟注入：必须在 main.tsx 之后，用独立 <script>（非 module）在页面解析
+         完成时注册 load 回调；window.load 触发后再 requestIdleCallback 插 <link>，
+         确保字体请求不进入首屏关键路径，也不阻塞 Playwright load 等待 -->
+    <script>
+      ;(function () {
+        var FONT_HREF =
+          "https://fonts.googleapis.com/css2?family=Noto+Sans+SC:wght@400;500;700;900&family=Noto+Serif+SC:wght@400;700;900&family=ZCOOL+KuaiLe&family=ZCOOL+XiaoWei&family=ZCOOL+QingKe+HuangYou&family=Ma+Shan+Zheng&family=Long+Cang&family=Liu+Jian+Mao+Cao&family=Zhi+Mang+Xing&display=swap"
+        function inject() {
+          if (document.getElementById("__gfonts_link")) return
+          var l = document.createElement("link")
+          l.id = "__gfonts_link"
+          l.rel = "stylesheet"
+          l.href = FONT_HREF
+          l.setAttribute("data-lazy", "1")
+          document.head.appendChild(l)
+        }
+        function schedule() {
+          if ("requestIdleCallback" in window) {
+            requestIdleCallback(inject, { timeout: 2000 })
+          } else {
+            setTimeout(inject, 500)
+          }
+        }
+        if (document.readyState === "complete") {
+          schedule()
+        } else {
+          window.addEventListener("load", schedule, { once: true })
+        }
+      })()
+    </script>
+    <noscript
+      ><link
+        rel="stylesheet"
+        href="https://fonts.googleapis.com/css2?family=Noto+Sans+SC:wght@400;500;700;900&family=Noto+Serif+SC:wght@400;700;900&family=ZCOOL+KuaiLe&family=ZCOOL+XiaoWei&family=ZCOOL+QingKe+HuangYou&family=Ma+Shan+Zheng&family=Long+Cang&family=Liu+Jian+Mao+Cao&family=Zhi+Mang+Xing&display=swap"
+    /></noscript>
  </body>
 </html>
@@ -255,6 +255,8 @@ const AiAvatarPage: React.FC = () => {
          audio_url: state.ttsPreview.audioUrl!,
          audio_duration: state.ttsPreview.duration,
          sentence_timings: state.ttsPreview.sentenceTimings,
+          // 预合成也必须带文案：后端落库 job.script_text，Ditto 情绪分析依赖它
+          script_text: state.scriptText,
          enable_video_loop: true,
        }
      } else {
@@ -108,7 +108,7 @@ export const FALLBACK_VIDEO_MODELS: ViralVideoModel[] = [
    display_name: "Wan 3.0 — 通义万相全能",
    supports_audio: true,
    supported_resolutions: ["480p", "720p", "1080p"],
-    max_duration: 15,
+    max_duration: 30,
  },
 ]
 export const RESOLUTION_ORDER = ["480p", "720p", "1080p", "4k"]
@@ -79,7 +79,7 @@ export const StoryboardCard: React.FC<StoryboardCardProps> = ({
        <div className="vv-copy-box vv-copy-loading">
          <VVSpinner />
          <span className="vv-copy-loading-text">{stageMsg}</span>
-          <span className="vv-copy-loading-hint">预计 10-30 秒，请稍候</span>
+          <span className="vv-copy-loading-hint">预计 5-15 秒，请稍候</span>
        </div>
      )
    }
@@ -397,6 +397,7 @@ def _call_llm_once(
    max_tokens: int,
    timeout: int,
    on_delta: Callable[[str, str], None] | None = None,
+    enable_thinking: bool | None = None,
 ) -> str | None:
    """统一封装同步/流式 LLM 调用；返回 raw 文本，失败返回 None。

@@ -404,6 +405,7 @@ def _call_llm_once(
    - 兼容旧签名：传 system_prompt+user_prompt 自动拼两消息；
    - on_delta=None：走同步 chat_completion；
    - on_delta 非 None 且 client 有 chat_completion_stream：走流式，节流推送 delta；
+    - enable_thinking=False：关豆包 reasoning（分镜模板填充不需要长思考，实测 TTFT 从 100s+ 降到 2s）；
    - 流式失败自动降级同步。
    """
    if not client or not getattr(client, "is_available", False):
@@ -413,7 +415,12 @@ def _call_llm_once(
    # 同步路径
    if on_delta is None or not hasattr(client, "chat_completion_stream"):
        try:
-            return client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout)
+            _kw: dict = {"max_retries": 0}  # 禁 SDK 内重试，外层 attempt 机制兜底
+            if enable_thinking is not None:
+                _kw["enable_thinking"] = enable_thinking
+            return client.chat_completion(
+                messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout, **_kw
+            )
        except Exception as e:
            logger.warning("[爆款视频] LLM同步调用失败: %s", e)
            return None
@@ -444,10 +451,15 @@ def _call_llm_once(
    # 这里用外层 deadline + 迭代超时双重保护。
    _stream_deadline = time.time() + max(timeout or 120, 30)
    stream_ok = False
-    _length_truncated = False  # finish_reason=length 时标记为被截断（raw 不完整，parser 必败）
+    # #2305: 3 态——None=未截断; True=截断但内容过短(<200字, 视为异常走下一attempt);
+    # False=截断但有可用内容(≥200字, 直接交给 parser 救回前N个镜头, 不走同步降级)
+    _length_truncated: bool | None = None
    try:
+        _stream_kw: dict = {}
+        if enable_thinking is not None:
+            _stream_kw["enable_thinking"] = enable_thinking
        stream_iter = client.chat_completion_stream(
-            messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout or 120
+            messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout or 120, **_stream_kw
        )
        while True:
            remaining = _stream_deadline - time.time()
@@ -458,13 +470,25 @@ def _call_llm_once(
                chunk = next(stream_iter)
            except StopIteration:
                stream_ok = True
-                # 流结束时再检查一次 finish_reason（最后一个 chunk 可能带 finish_reason 但无 content）
-                if getattr(client, "last_finish_reason", "") == "length":
-                    _length_truncated = True
-                    logger.warning(
-                        "[爆款视频] 流式结束时检测到 finish_reason=length（received=%d chars），丢弃走下一 attempt",
-                        sum(len(p) for p in full_parts),
-                    )
+                # 流自然结束时统一做 finish_reason=length 三态判断（#2305 追加：
+                # ark 流式把 finish_reason 挂在最后一个无 content 的 chunk 上，ai_client yield 完
+                # content chunks 后置 last_finish_reason 然后 return，下一次 next() 直接 StopIteration，
+                # 所以 chunk 分支里的三态判断是死代码，真正的截断检测必须在这里做）。
+                fr = getattr(client, "last_finish_reason", "")
+                if fr == "length":
+                    _received_len = sum(len(p) for p in full_parts)
+                    if _received_len < 200:
+                        _length_truncated = True
+                        logger.warning(
+                            "[爆款视频] 流式结束 finish=length（received=%d chars，过短），丢弃走下一 attempt",
+                            _received_len,
+                        )
+                    else:
+                        _length_truncated = False
+                        logger.warning(
+                            "[爆款视频] 流式结束 finish=length（received=%d chars），尝试解析前N镜头救回",
+                            _received_len,
+                        )
                break
            except Exception as inner_e:
                err_msg = str(inner_e).lower()
@@ -481,8 +505,9 @@ def _call_llm_once(
                )
                if should:
                    _flush(force=False)
-                # 早期格式预判：>800字仍无任何结构化标记（<script/<clips/<clip/镜头1/镜头一/第1镜），
-                # 基本可以判定 LLM 跑题输出自然语言，parser 必败，早停省 15~30s
+                # 早期格式预判：>800字仍无任何结构化标记（<script/<clips/<clip/镜头1/镜头一/第1镜/0-N秒开头），
+                # 基本可以判定 LLM 跑题输出自然语言，parser 必败，早停省 15~30s。
+                # #2304 追加修复：必须 >800 字才检查（前 800 字是 LLM 热身/概述区，不能误杀正常开头）。
                _acc_len = sum(len(p) for p in full_parts)
                if _acc_len > 800:
                    _full_so_far = "".join(full_parts)
@@ -497,41 +522,33 @@ def _call_llm_once(
                            "镜头一",
                            "第1镜",
                            "第 1 镜",
-                            "0-",
-                            "0~",
-                            "1-",
-                            "2-",
-                            "3-",
-                            "4-",
-                            "5-",
+                            "秒",  # v6 格式包含时间范围"0-4秒""4~8秒"
+                            "min",  # 英文格式 "0-4s"
                        )
                    ):
                        logger.warning(
                            "[爆款视频] 流式 early-break：前%d字无结构化标记 head=%r",
                            _acc_len,
-                            _full_so_far[:80].replace("\n", "\\n"),
+                            _full_so_far[:80].replace("\n", "\\"),
                        )
                        break
-            # 检测 finish_reason=length（max_tokens 截断）——chunk 是 text，finish_reason 由
-            # DoubaoClient 在 yield 后写入 self.last_finish_reason。截断时 XML/双段式缺闭合标签，
-            # parser 必败，直接丢 raw 走下一 attempt（自带更大 max_tokens）。
-            if getattr(client, "last_finish_reason", "") == "length":
-                _length_truncated = True
-                logger.warning(
-                    "[爆款视频] 流式输出被 max_tokens=%d 截断（received=%d chars），丢弃走下一 attempt",
-                    max_tokens,
-                    sum(len(p) for p in full_parts),
-                )
-                break
+            # 注：finish_reason=length 的三态判断统一在 StopIteration 分支处理
+            # （ark 把 finish 挂在最后一个无 content 的 chunk，这里带 content 的 chunk 几乎不会触发）。
    except Exception as e:
        logger.warning("[爆款视频] 流式调用失败，降级同步: %s", e, exc_info=True)
        full_parts = []
-    if _length_truncated:
-        # max_tokens 截断：raw 不完整，不降级同步（同步也同样 max_tokens 照样截断）
+    if _length_truncated is True:
+        # 截断且内容过短（疑似异常/误判）：不降级同步（同步也同样 max_tokens 照样截断），返回 None 走下一 attempt
        return None
+    # _length_truncated is False: 截断但有可用内容，stream_ok 已置 True，跳过同步降级直接过 parser
    if not stream_ok:
        try:
-            raw = client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout)
+            _fb_kw: dict = {"max_retries": 0}  # 禁 SDK 内重试，避免单 attempt 被重试拖到 300s+
+            if enable_thinking is not None:
+                _fb_kw["enable_thinking"] = enable_thinking
+            raw = client.chat_completion(
+                messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout, **_fb_kw
+            )
        except Exception as e:
            logger.warning("[爆款视频] LLM同步降级调用失败: %s", e)
            return None
@@ -1157,6 +1174,78 @@ def _parse_script(raw: str, job: ViralVideoJob) -> dict | None:
    return _parse_script_markdown(raw, job, dur, ratio, default_theme)


+# #2308: 从旧schema混写的<visual>字段中拆分出 摄影参数 / 画面描述 / 动作细节
+_CAMERA_KEYWORDS = re.compile(
+    r"(?:景别|景|远景|全景|中景|近景|特写|大特写|俯拍|仰拍|平视|侧拍|"
+    r"机位|高度|机位高度|鸟瞰|斜拍|跟拍|固定镜头|手持|稳定器|"
+    r"推镜|拉镜|摇镜|移镜|平移|横移|升降|环绕|旋转|变焦|"
+    r"缓慢推|快速推|缓推|急推|慢推|匀速|"
+    r"[\d]+mm|f/[\d.]+|光圈|焦段|焦距|景深|浅景深)"
+)
+_ACTION_VERBS = re.compile(
+    r"(?:表情|面带|露出|微笑|皱眉|闭眼|睁眼|眼神|疲惫|放松|紧张|看向|望着|注视|凝视|"
+    r"伸手|抬手|挥手|举起|放下|拿起|握住|递|接|手扶|轻抚|抚摸|触碰|轻触|触摸|翻看|"
+    r"转身|转头|回头|点头|摇头|低头|抬头|歪头|扭头|"
+    r"走|跑|站|坐|躺|靠|倚|弯腰|侧身|前倾|后仰|走进|走出|推门|开门|关门|进入|离开|"
+    r"说话|开口|张嘴|嘴角上扬|深呼吸|吐气|流泪|"
+    r"动作|动作细节|操作|演示|接过|递上|展示|做出|比划)"
+)
+
+
+def _split_visual_field(visual: str) -> tuple[str, str, str]:
+    """把旧schema混写的<visual>文本拆成 (摄影参数, 画面描述, 动作细节)。
+
+    策略：按逗号/顿号/分号切分小片段，含摄影关键词→camera，含动作动词→action，
+    其余（主体/场景/光线/氛围）→visual。切不动时按第一个逗号前归camera、
+    最后一句含动词归action、中间归visual 兜底。
+    """
+    text = (visual or "").strip()
+    if not text:
+        return "", "", ""
+    # 去掉首尾的"镜头："/"画面："等前缀
+    text = re.sub(r"^(?:镜头|画面|视觉)\s*[:：]\s*", "", text)
+    cam_parts: list[str] = []
+    act_parts: list[str] = []
+    vis_parts: list[str] = []
+    # 按中英文逗号、顿号、分号切片
+    segs = [s.strip() for s in re.split(r"[，,、;；]\s*", text) if s.strip()]
+    if len(segs) <= 1:
+        return text, text, ""  # 切不开，整体当画面+摄影
+    for seg in segs:
+        if _CAMERA_KEYWORDS.search(seg) and not _ACTION_VERBS.search(seg):
+            cam_parts.append(seg)
+        elif _ACTION_VERBS.search(seg) and not _CAMERA_KEYWORDS.search(seg):
+            act_parts.append(seg)
+        else:
+            vis_parts.append(seg)
+    # 兜底：如果 camera 为空，取第一段
+    if not cam_parts and segs:
+        cam_parts.append(segs[0])
+        vis_parts = [s for s in vis_parts if s != segs[0]]
+    # 兜底1：最后一段含动作动词
+    if not act_parts and segs and _ACTION_VERBS.search(segs[-1]):
+        act_parts.append(segs[-1])
+        vis_parts = [s for s in vis_parts if s != segs[-1]]
+    # 兜底2：从后往前扫，找到第一个含动作动词且不含摄影关键词的段
+    if not act_parts:
+        for seg in reversed(segs):
+            if _ACTION_VERBS.search(seg) and seg not in cam_parts:
+                act_parts.append(seg)
+                vis_parts = [s for s in vis_parts if s != seg]
+                break
+    # 兜底3：包含人物主语（妈妈/店员/女生/男生/她/他/人物/模特）+动词的段视为动作
+    if not act_parts:
+        _person_re = re.compile(
+            r"(?:妈妈|店员|女生|男生|女士|男士|女孩|男孩|人物|模特|主播|她|他|主角|孩子|宝宝|老人|顾客|客人|顾客)"
+        )
+        for seg in reversed(segs):
+            if _person_re.search(seg) and seg not in cam_parts and not _CAMERA_KEYWORDS.search(seg):
+                act_parts.append(seg)
+                vis_parts = [s for s in vis_parts if s != seg]
+                break
+    return "，".join(cam_parts), "，".join(vis_parts), "，".join(act_parts)
+
+
 def _parse_script_xml(
    xml_body: str,
    raw_full: str,
@@ -1182,7 +1271,22 @@ def _parse_script_xml(
        a = c.get("attrs", {})
        body = c.get("text", "") or ""
        voice = xp.text_of(body, "voiceover") or xp.text_of(body, "voice_text")
-        visual = xp.text_of(body, "visual") or xp.text_of(body, "shot_type_angle_movement")
+        # #2308: 优先读新 <shot>（纯6项摄影参数），fallback 到 <visual>（旧schema混写）
+        shot_meta = xp.text_of(body, "shot") or ""
+        visual_full = xp.text_of(body, "visual") or xp.text_of(body, "shot_type_angle_movement") or ""
+        action_raw = xp.text_of(body, "action_details") or ""
+        if shot_meta:
+            # 新schema：<shot>是摄影参数，<visual>是纯画面描述，<action_details>是动作细节
+            cam_text = shot_meta.strip()
+            visual_text = visual_full.strip()
+            action_text = (action_raw or "").strip()
+            if action_text in ("-", "—", "无", "无。"):
+                action_text = ""
+        else:
+            # 旧schema兼容：<visual> 混写了"景别/运镜+画面+动作"，用正则拆分
+            cam_text, visual_text, action_text = _split_visual_field(visual_full)
+            if not action_text and action_raw:
+                action_text = action_raw.strip()
        ref_raw = xp.text_of(body, "reference_image_index")
        ref_idx = xp.attr_int(ref_raw, -1) if ref_raw not in (None, "") else -1
        if not isinstance(ref_idx, int) or ref_idx < 0:
@@ -1192,9 +1296,13 @@ def _parse_script_xml(
            voice_parts.append(f"[speaker:{speaker}]{voice}")
        shot = {
            "time_range": a.get("time_range") or xp.text_of(body, "time_range") or f"{i * 3}-{(i + 1) * 3}秒",
-            "shot_type_angle_movement": visual or "中景平视，固定镜头",
-            "scene_and_dialogue": (voice or "") if not visual else f"{visual}｜{voice or ''}",
-            "action_details": xp.text_of(body, "action_details") or "",
+            # shot_type_angle_movement：纯6项摄影参数，不塞画面动作
+            "shot_type_angle_movement": cam_text or "中景平视，固定镜头",
+            # scene_and_dialogue：纯口播对白（不再拼接visual，避免重复）
+            "scene_and_dialogue": voice or "",
+            # 画面描述（拆出来的纯画面）放到 shot 里供视频prompt使用
+            "visual": visual_text,
+            "action_details": action_text,
            "audio_bgm": xp.text_of(body, "audio_bgm") or "轻快BGM",
            "transition": xp.text_of(body, "transition") or ("硬切" if i < len(clips) - 1 else "结束"),
            "reference_image_index": ref_idx,
@@ -1228,11 +1336,13 @@ _MD_TIME_SPLIT_RE = re.compile(
 )
 _MD_TIME_RE = re.compile(r"(\d+)\s*[-~到至]\s*(\d+)\s*秒?")
 _MD_FIELD_RE = {
-    "visual": re.compile(r"(?:画面|视觉|镜头画面|shot_type_angle_movement)\s*[:：]\s*([^\n]+)"),
-    "camera": re.compile(r"(?:景别|运镜|镜头运动)\s*[:：]\s*([^\n]+)"),
+    "visual": re.compile(
+        r"(?:画面(?:内容|描述)?|视觉|镜头画面|场景(?:描述)?|shot_type_angle_movement)\s*[:：]\s*([^\n]+)"
+    ),
+    "camera": re.compile(r"(?:景别(?:/角度)?(?:与运镜)?|运镜|镜头运动|镜头|shot|camera)\s*[:：]\s*([^\n]+)"),
    "voiceover": re.compile(r"""(?:口播|对白|台词|配音|旁白|voiceover)\s*[:：]\s*""" r"""(?:"([^"]+)"|'([^']+)'|(\S[^
 ]*?)(?=\n\s*(?:镜头|#|$)|\s*$))"""),
-    "action": re.compile(r"(?:动作|动作细节|action)\s*[:：]\s*([^\n]+)"),
+    "action": re.compile(r"(?:动作(?:与真人细节)?|动作细节|人物动作|action)\s*[:：]\s*([^\n]+)"),
    "audio_bgm": re.compile(r"(?:音效|BGM|背景音乐|audio)\s*[:：]\s*([^\n]+)"),
    "transition": re.compile(r"(?:转场|过渡)\s*[:：]\s*([^\n]+)"),
    "reference": re.compile(r"参考@图片(\d+)|参考图[:：]?\s*(\d+)"),
@@ -1248,7 +1358,7 @@ def _parse_script_markdown(
    ratio: str,
    default_theme: str,
 ) -> dict | None:
-    """中文 markdown 兜底解析（格式C）：按"镜头1/镜头一/第1镜"等分割。"""
+    """中文 markdown 兜底解析（格式C/D）：按"镜头1/镜头一/第1镜"或"0-4秒/4-8秒"时间范围分割。"""
    # 提取 theme
    theme = default_theme
    theme_match = _MD_FIELD_RE["theme"].search(raw)
@@ -1355,14 +1465,15 @@ def _parse_script_markdown(
        if voice:
            voice_parts.append(voice)

-        # scene_and_dialogue 保持原有拼接格式
-        visual_desc = visual or "中景平视，固定镜头"
-        scene_and_dialogue = (voice or "") if not visual_desc else f"{visual_desc}｜{voice or ''}"
+        # #2308: scene_and_dialogue 只放口播对白，画面/景别已在独立字段
+        _ = visual or camera  # 兼容保留
+        scene_and_dialogue = voice or ""

        shot = {
            "time_range": time_range,
            "shot_type_angle_movement": shot_type,
            "scene_and_dialogue": scene_and_dialogue,
+            "visual": visual or "",
            "action_details": action,
            "audio_bgm": audio or "轻快BGM",
            "transition": trans or ("硬切" if i < len(shot_blocks) - 1 else "结束"),
@@ -1421,6 +1532,27 @@ def _step_script_generation(
    dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))

    template = get_template("storyboard")
+    # P0 观测：模板来自代码默认（DB 加载失败）时打 ERROR，避免静默回落
+    # 判定方式：system_prompt 与 DEFAULT_TEMPLATES storyboard 完全一致
+    try:
+        from packages.application.viral_video.prompts import DEFAULT_TEMPLATES as _DEF_T
+
+        _def_story = next((t for t in _DEF_T if t["prompt_type"] == "storyboard"), None)
+        if (
+            template is not None
+            and _def_story is not None
+            and template.system_prompt == _def_story["system_prompt"]
+            and (template.example_output or "") == (_def_story.get("example_output") or "")
+        ):
+            logger.error(
+                "[爆款视频] storyboard 模板使用代码默认 v%s/%s（DB 加载失败或无激活模板），"
+                "检查 DB 连接与 is_active——本次生成质量无保障，job_id=%s",
+                template.version,
+                template.name,
+                job.id,
+            )
+    except Exception:  # noqa: BLE001
+        pass
    marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
    theme_hint = _determine_theme(image_analysis, marketing_purpose)

@@ -1491,23 +1623,22 @@ def _step_script_generation(
    client_fast = ai_router.get_llm_client("storyboard", variant="lite")
    if not client_fast or not getattr(client_fast, "is_available", False):
        client_fast = client_primary
-    # 单档超时 + 总 deadline：doubao 流式偶发 90~120s，总 deadline 必须覆盖 2 次 fast + 1 次 pro
-    # （原 180s 太短，两次 fast 调用就耗尽，pro-0.7 永远跑不到）
-    fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "90"))
-    pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "120"))
-    deadline = time.time() + int(os.environ.get("VIRAL_VIDEO_SCRIPT_TOTAL_DEADLINE", "300"))
+    # 单档超时 + 总 deadline：豆包 seed-2-1-pro 关思考后 2500~3500 tokens 完整流式约 60~90s；
+    # 开思考时 reasoning 可能 100s+ 才出第一个 content，分镜模板填充式生成不需要长推理，
+    # 统一 enable_thinking=False 后按关思考的耗时设超时（10-15 实测 probe：fast 90s 内可完成，留 margin）。
+    # 若环境变量覆盖则优先使用环境变量，便于线上应急调参。
+    fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "120"))
+    pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "150"))
+    deadline = time.time() + int(os.environ.get("VIRAL_VIDEO_SCRIPT_TOTAL_DEADLINE", "450"))

    # ── 重试 for 循环（替代原嵌套 if + 两个重复 _try_gen） ─────────────────
    attempts: list[tuple] = [
        (client_fast, 0.8, 2500, "fast-0.8", fast_tmo),
        (client_fast, 0.6, 3200, "fast-0.6", fast_tmo),
    ]
-    if (
-        client_primary
-        and client_primary.is_available
-        and (not client_fast or client_primary.model != client_fast.model)
-    ):
-        attempts.append((client_primary, 0.7, 3500, "pro-0.7", pro_tmo))
+    # pro-0.7 第三档：primary 模型（可能与 fast 同一模型，但不同 temp/max_tokens 值得再试）
+    if client_primary and client_primary.is_available:
+        attempts.append((client_primary, 0.7, 4000, "pro-0.7", pro_tmo))

    try:
        for client, temp, max_tok, label, tmo in attempts:
@@ -1522,6 +1653,7 @@ def _step_script_generation(
                max_tokens=max_tok,
                timeout=tmo,
                on_delta=on_delta,
+                enable_thinking=False,  # 分镜模板填充式生成不需要长推理，关思考 TTFT<3s 实测
            )
            if not raw:
                logger.warning("[爆款视频] 分镜生成返回空 label=%s", label)
@@ -1724,9 +1856,247 @@ def _get_speaker_voice_id(speaker: str, main_voice_id: str, is_main_voice_male:
        return "longsanshu"  # 男声


-def _step_tts(job: ViralVideoJob, voiceover_script: str):
-    """步骤 5: CosyVoice 整段配音 → 返回本地 MP3 Path；失败返回 None。
+def _get_voice_sample_script(language: str = "zh-CN") -> str:
+    """#2311: 返回固定音色样本文案（约8秒，覆盖常见音素）。

+    短音色样本模式：TTS不再合成完整口播稿，而是合成一段固定短文案
+    作为 reference_audio 传给视频模型，由模型从短样本提取speaker embedding
+    后原生音画同生，口播文案全部放在prompt里。
+    """
+    lang = (language or "zh-CN").lower().replace("_", "-")
+    if lang.startswith("zh"):
+        return VOICE_SAMPLE_SCRIPT_ZH
+    return VOICE_SAMPLE_SCRIPT_EN
+
+
+# #2311: 预置音色静态样本 OSS 路径（部署时由 scripts/seed_preset_voice_samples.py 预生成上传，公共读）
+_PRESET_VOICE_SAMPLE_OSS_KEY_FMT = "static/voice-samples/preset/{voice_id}_sample.mp3"
+# #2311: 克隆音色预处理样本 OSS 路径（首次出片时用原音频ffmpeg截8s上传，后续永久复用）
+_CLONE_VOICE_SAMPLE_OSS_KEY_FMT = "static/voice-samples/clone/{profile_id}_sample.mp3"
+# #2311: 样本目标时长（秒）
+_VOICE_SAMPLE_TARGET_SECS = 8
+
+
+def _trim_audio_to_sample(local_path: Path, target_secs: int = _VOICE_SAMPLE_TARGET_SECS) -> Path | None:
+    """#2311: 用ffmpeg把音频裁剪/标准化为target_secs秒的单声道16kHz参考样本。
+    取音频中段（避开开头呼吸/结尾尾音/可能的静音段），输出到临时目录mp3。
+    失败返回None。
+    """
+    import subprocess
+    import tempfile
+
+    if not local_path or not local_path.exists():
+        return None
+    try:
+        probe = subprocess.run(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                str(local_path),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=15,
+        )
+        dur = float(probe.stdout.strip() or "0")
+        if dur <= 0:
+            return None
+        # 取中段；若比目标短则从头取
+        start = max(0.0, (dur - target_secs) / 2.0) if dur > target_secs + 1 else 0.0
+        out_dir = Path(tempfile.mkdtemp(prefix="vv_voice_sample_"))
+        out = out_dir / "sample.mp3"
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-ss",
+            f"{start:.2f}",
+            "-t",
+            str(target_secs),
+            "-i",
+            str(local_path),
+            "-ac",
+            "1",
+            "-ar",
+            "16000",
+            "-b:a",
+            "96k",
+            str(out),
+        ]
+        subprocess.run(cmd, capture_output=True, timeout=30, check=True)
+        if out.exists() and out.stat().st_size > 4096:
+            return out
+        return None
+    except Exception as e:
+        logger.warning("[爆款视频] 音频样本ffmpeg预处理失败: %s", e, exc_info=True)
+        return None
+
+
+def _download_audio_to_local(url_or_key: str, local_path: Path) -> bool:
+    """#2311: 从公网URL或storage_key下载音频到本地路径。成功返回True。"""
+    from video_processing.oss_helpers import download_asset
+
+    local_path.parent.mkdir(parents=True, exist_ok=True)
+    # 1. storage_key 路径（不带 :// 且不含 .com/.net 等域名特征）：用 download_asset
+    if "://" not in url_or_key and not any(d in url_or_key for d in (".com", ".net", ".org")):
+        try:
+            return download_asset(url_or_key, local_path)
+        except Exception as e:
+            logger.warning("[爆款视频] storage_key下载失败 key=%s err=%s", url_or_key[:80], e)
+            return False
+    # 2. 公网URL：urllib下载
+    try:
+        import urllib.request
+
+        urllib.request.urlretrieve(url_or_key, str(local_path))
+        return local_path.exists() and local_path.stat().st_size > 1024
+    except Exception as e:
+        logger.warning("[爆款视频] 音频HTTP下载失败: %s", e)
+        return False
+
+
+def _resolve_voice_sample_audio(job: ViralVideoJob) -> str | None:
+    """#2311: 热点路径零TTS — 为本次任务获取reference_audio用的~8s音色样本公网URL。
+
+    三层策略（正常路径零TTS）：
+    1. 无自定义音色 → None（模型原生音频）
+    2. 预置音色 → 直接读 PresetVoice.sample_audio_url（部署时seed脚本预生成上传OSS）
+    3. 克隆音色 → 从profile取原音频 → ffmpeg截中段8s → OSS缓存永久复用（仅首次预处理）
+    4. 兜底：以上失败则TTS合成短样本，保证任务不崩（正常路径不触发）
+    """
+    import tempfile
+
+    raw_voice_id = (getattr(job, "voice_id", "") or "").strip()
+    voice_source = (getattr(job, "voice_source", "") or "").strip().lower()
+    if not raw_voice_id:
+        return None
+
+    try:
+        from video_processing.oss_helpers import file_exists, get_public_url, upload_to_oss
+    except Exception as e:
+        logger.warning("[爆款视频] OSS工具导入失败，降级TTS: %s", e)
+        return _tts_voice_sample_fallback(job)
+
+    # --- 预置音色：直接读预生成sample_audio_url ---
+    if voice_source not in ("my_voice", "clone"):
+        from packages.domain.preset_voices import get_preset_voice_by_id
+
+        preset = get_preset_voice_by_id(raw_voice_id)
+        sample_url = (preset.sample_audio_url if preset else "").strip()
+        if not sample_url:
+            # 部署脚本没跑/未回填：警告+兜底TTS，保证不崩
+            logger.warning(
+                "[爆款视频][音色样本] 预置音色 %s 无sample_audio_url（seed_preset_voice_samples未跑？），降级TTS",
+                raw_voice_id,
+            )
+            return _tts_voice_sample_fallback(job)
+        logger.info("[爆款视频][音色样本] 命中预置静态样本: %s", raw_voice_id)
+        return sample_url
+
+    # --- 克隆音色：原音频ffmpeg预处理+OSS缓存 ---
+    profile_id = raw_voice_id
+    oss_key = _CLONE_VOICE_SAMPLE_OSS_KEY_FMT.format(profile_id=profile_id)
+    # 先查缓存
+    try:
+        if file_exists(oss_key):
+            cached = get_public_url(oss_key)
+            if cached:
+                logger.info("[爆款视频][音色样本] 命中克隆静态缓存: %s", profile_id)
+                return cached
+    except Exception:
+        pass
+
+    # 下载原始音频
+    src_url = None
+    session = None
+    try:
+        from packages.adapters.sqlalchemy_impl.asset_repository import (
+            SQLAlchemyAssetRepository,
+        )
+        from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
+            SQLAlchemyVoiceCloneProfileRepository,
+        )
+
+        session = SessionLocal()
+        _repo = SQLAlchemyVoiceCloneProfileRepository(session)
+        _profile = _repo.get(profile_id)
+        if not _profile or not _profile.is_ready:
+            logger.warning("[爆款视频][音色样本] 克隆profile %s 不存在/未就绪，降级TTS", profile_id)
+            return _tts_voice_sample_fallback(job)
+        # 优先走source_asset_id拿永久storage_key重签下载
+        source_asset_id = (_profile.metadata or {}).get("source_asset_id", "")
+        if source_asset_id:
+            try:
+                _asset_repo = SQLAlchemyAssetRepository(session)
+                asset = _asset_repo.find_by_id(source_asset_id)
+                if asset and asset.storage_key:
+                    src_url = asset.storage_key  # download_asset 走 storage_key
+            except Exception as e:
+                logger.warning("[爆款视频][音色样本] asset.storage_key获取失败，回退source_audio_url: %s", e)
+        if not src_url:
+            src_url = (_profile.source_audio_url or "").strip()
+    except Exception as e:
+        logger.warning("[爆款视频][音色样本] 克隆profile查询失败 %s，降级TTS: %s", profile_id, e)
+        return _tts_voice_sample_fallback(job)
+    finally:
+        if session is not None:
+            try:
+                session.close()
+            except Exception:
+                pass
+
+    if not src_url:
+        logger.warning("[爆款视频][音色样本] 克隆profile %s 无源音频，降级TTS", profile_id)
+        return _tts_voice_sample_fallback(job)
+
+    tmpdir = Path(tempfile.mkdtemp(prefix="vv_clone_src_"))
+    local_src = tmpdir / "source_audio"
+    if not _download_audio_to_local(src_url, local_src):
+        logger.warning("[爆款视频][音色样本] 克隆原音频下载失败，降级TTS profile=%s", profile_id)
+        return _tts_voice_sample_fallback(job)
+
+    # ffmpeg截中段8s
+    sample_path = _trim_audio_to_sample(local_src, _VOICE_SAMPLE_TARGET_SECS)
+    if not sample_path:
+        logger.warning("[爆款视频][音色样本] ffmpeg预处理失败，降级TTS profile=%s", profile_id)
+        return _tts_voice_sample_fallback(job)
+
+    # 上传OSS永久缓存
+    try:
+        url = upload_to_oss(sample_path, oss_key)
+        if url:
+            logger.info("[爆款视频][音色样本] 克隆样本已缓存OSS: %s", profile_id)
+            return url
+    except Exception as e:
+        logger.warning("[爆款视频][音色样本] 克隆样本上传OSS失败: %s", e)
+    # 临时上传兜底
+    try:
+        return _upload_tts_to_oss(job, sample_path)
+    except Exception:
+        return _tts_voice_sample_fallback(job)
+
+
+def _tts_voice_sample_fallback(job: ViralVideoJob) -> str | None:
+    """#2311: TTS兜底 — 静态样本路径异常时，合成一份8s短样本上传返回URL。
+    正常路径永不触发，仅作为最后防线保证任务不崩。
+    """
+    logger.warning("[爆款视频][音色样本] 走TTS兜底路径（正常不应触发）")
+    _sample_text = _get_voice_sample_script(getattr(job, "language", "zh-CN") or "zh-CN")
+    tts_path = _step_tts(job, _sample_text)
+    if not tts_path:
+        return None
+    return _upload_tts_to_oss(job, tts_path)
+
+
+def _step_tts(job: ViralVideoJob, voiceover_script: str):
+    """步骤 5: CosyVoice 配音 → 返回本地 MP3 Path；失败返回 None。
+
+    #2311 短音色样本模式：voiceover_script 由调用方决定传入内容
+    （完整口播稿 / 短音色样本文案），本函数只负责把传入文本合成为MP3。
    #2188: 支持 voice_source='my_voice'/'clone'，先把前端传的 profile.id
    解析成 CosyVoice 真正的克隆音色 voice_id，再走同一条 synthesize 链路。
    """
@@ -1736,7 +2106,6 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str):
        from apps.worker.services.tts_service_factory import get_tts_service

        tts_service = get_tts_service()
-        # #2188: 根据 voice_source 解析实际传给 CosyVoice 的 voice_id
        voice_id = _resolve_tts_voice_id(job)
        text = (voiceover_script or "").strip()
        if not text:
@@ -1760,11 +2129,11 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str):
                    result = tts_service.synthesize(text=text)
        if result is None:
            return None
-        p = _Path(result) if not isinstance(result, _Path) else result
-        if p.exists() and p.stat().st_size > 0:
-            logger.info("[爆款视频] TTS 合成完成: voice=%s path=%s size=%d", voice_id, p, p.stat().st_size)
-            return p
-        logger.warning("[爆款视频] TTS 返回路径不存在或空文件: %s", p)
+        pth = _Path(result) if not isinstance(result, _Path) else result
+        if pth.exists() and pth.stat().st_size > 0:
+            logger.info("[爆款视频] TTS 合成完成: voice=%s path=%s size=%d", voice_id, pth, pth.stat().st_size)
+            return pth
+        logger.warning("[爆款视频] TTS 返回路径不存在或空文件: %s", pth)
        return None
    except Exception as e:
        logger.warning("[爆款视频] TTS 配音失败: %s", e, exc_info=True)
@@ -1772,7 +2141,7 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str):


 def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None:
-    """把 TTS 本地 mp3 上传到 OSS，返回公网 URL（供 Seedance 做 reference_audios 口型驱动用）。"""
+    """把本地 mp3 上传到 OSS，返回公网 URL。"""
    if tts_path is None:
        return None
    try:
@@ -1791,144 +2160,6 @@ def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None:
        return None


-def _check_and_fix_tts_duration(tts_path, target_duration: int):
-    """Bug3 fix: TTS 合成后校验音频时长，超限则加速/截断。
-
-    - 音频时长 > max(target+2, 30) → 加速到目标时长
-    - 音频时长 > 30s（Seedance 硬限制）→ 必须加速/截断到 30s 以内
-    返回处理后的路径（可能覆盖原文件），失败返回原路径不阻塞。
-    """
-    if tts_path is None:
-        return None
-    try:
-        import subprocess
-        from pathlib import Path as _Path
-
-        local = _Path(tts_path) if not isinstance(tts_path, _Path) else tts_path
-        if not local.exists():
-            return tts_path
-
-        # 用 ffprobe 检查音频时长
-        probe_cmd = [
-            "ffprobe",
-            "-v",
-            "quiet",
-            "-show_entries",
-            "format=duration",
-            "-of",
-            "default=noprint_wrappers=1:nokey=1",
-            str(local),
-        ]
-        result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=15)
-        if result.returncode != 0:
-            logger.warning("[爆款视频] ffprobe 检查音频时长失败: %s", result.stderr[:200])
-            return tts_path
-
-        audio_dur = float(result.stdout.strip())
-        hard_limit = 30.0  # Seedance 硬限制
-        soft_limit = float(target_duration) + 2.0
-        effective_limit = min(soft_limit, hard_limit)
-
-        logger.info(
-            "[爆款视频] TTS 音频时长检查: audio=%.1fs target=%ds limit=%.1fs",
-            audio_dur,
-            target_duration,
-            effective_limit,
-        )
-
-        if audio_dur <= effective_limit:
-            return tts_path  # 时长合理
-
-        # 需要处理：优先用 ffmpeg 加速（atempo）保持内容完整
-        if audio_dur > hard_limit:
-            target_sec = 29.0  # 必须压缩到 30s 以内
-        else:
-            target_sec = float(target_duration)
-
-        speed_factor = audio_dur / target_sec
-        if speed_factor > 2.0:
-            # atempo 最大 2.0x，超过则先加速到 2x 再截断
-            logger.warning(
-                "[爆款视频] TTS 音频加速比 %.2f 超过 2x，改为加速+截断",
-                speed_factor,
-            )
-            accel_path = local.with_suffix(".accel.mp3")
-            subprocess.run(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-i",
-                    str(local),
-                    "-filter:a",
-                    "atempo=2.0",
-                    "-c:a",
-                    "libmp3lame",
-                    "-q:a",
-                    "2",
-                    str(accel_path),
-                ],
-                capture_output=True,
-                timeout=60,
-            )
-            if accel_path.exists() and accel_path.stat().st_size > 0:
-                accel_path.replace(local)
-                # 再截断到目标时长
-                trimmed = local.with_suffix(".trimmed.mp3")
-                subprocess.run(
-                    [
-                        "ffmpeg",
-                        "-y",
-                        "-i",
-                        str(local),
-                        "-t",
-                        str(target_sec),
-                        "-c:a",
-                        "libmp3lame",
-                        "-q:a",
-                        "2",
-                        str(trimmed),
-                    ],
-                    capture_output=True,
-                    timeout=30,
-                )
-                if trimmed.exists() and trimmed.stat().st_size > 0:
-                    trimmed.replace(local)
-                logger.info("[爆款视频] TTS 音频加速+截断完成: %.1fs → %.1fs", audio_dur, target_sec)
-        else:
-            # 用 atempo 加速
-            accelerated = local.with_suffix(".accel.mp3")
-            subprocess.run(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-i",
-                    str(local),
-                    "-filter:a",
-                    f"atempo={speed_factor:.4f}",
-                    "-c:a",
-                    "libmp3lame",
-                    "-q:a",
-                    "2",
-                    str(accelerated),
-                ],
-                capture_output=True,
-                timeout=60,
-            )
-            if accelerated.exists() and accelerated.stat().st_size > 0:
-                accelerated.replace(local)
-                logger.info(
-                    "[爆款视频] TTS 音频加速完成: %.1fs → %.1fs (speed=%.2fx)",
-                    audio_dur,
-                    target_sec,
-                    speed_factor,
-                )
-
-        return tts_path
-    except Exception as e:
-        logger.warning("[爆款视频] TTS 音频时长校验异常: %s", e, exc_info=True)
-        return tts_path  # 校验异常不阻塞
-
-
 def _get_expected_shot_count(duration: int) -> tuple[int, int] | None:
    """根据视频时长返回期望的镜头数量范围 (min, max)，从 constants 统一取值。"""
    try:
@@ -2029,8 +2260,14 @@ def _truncate_voiceover(text: str, max_chars: int) -> str:


 def _shot_visual(s: dict) -> str:
-    """合并镜头画面与动作描述。"""
-    visual = str(s.get("scene_and_dialogue", "") or s.get("visual", "")).strip()
+    """合并镜头画面与动作描述（#2308: scene_and_dialogue现在只放对白，画面应读visual字段）。"""
+    visual = str(s.get("visual", "") or "").strip()
+    # 兜底：旧数据/兜底脚本可能没填visual，则从scene_and_dialogue取
+    if not visual:
+        sd = str(s.get("scene_and_dialogue", "") or "").strip()
+        # scene_and_dialogue现在是纯对白（用｜分隔是旧格式），只有不含对白痕迹时才当画面
+        if "｜" not in sd and not any(c in sd for c in "，。？！") or len(sd) < 20:
+            visual = sd
    act = str(s.get("action_details", "") or "").strip()
    if act and act not in visual:
        visual = f"{visual}，{act}" if visual else act
@@ -2074,6 +2311,12 @@ _WAN_HEAD_SCENE_FALLBACK = "画面清晰有质感"
 _WAN_STYLE_LABEL = "风格说明："
 _WAN_SHOT_PREFIX = "第{n}个镜头[{label}]"

+# #2311 短音色样本固定文案：约8秒，覆盖中文四声/开合齐撮/常见声母韵母，男女声通用
+# 用固定文案合成音色样本（而非整段口播），reference_audio只传音色特征；
+# 口播内容全部放prompt里，由模型原生音画同生，避免撞reference_audio时长硬限。
+VOICE_SAMPLE_SCRIPT_ZH = "你好，很高兴认识你。今天天气真不错，我们一起来看看吧。"
+VOICE_SAMPLE_SCRIPT_EN = "Hi there, nice to meet you. The weather is great today, let's take a look together."
+

 def _build_style_section(style_guide: dict | None, scene_lighting: str) -> str:
    """从 style_guide / scene_lighting 生成整体风格描述文本。"""
@@ -2207,15 +2450,16 @@ def _generate_character_sheet(
        return None


-def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
+def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob, has_voice_ref: bool = False) -> str:
    """把编导脚本拼成符合官方推荐格式的 prompt（按 provider 分发）。

    - doubao (Seedance): [X-Y秒] 单行时间戳 + 镜头级 @图片N 绑定
    - dashscope (Wan 3.0): 第N个镜头[X-Y秒] 单行格式

-    PR3 P2(e): 字符串常量已抽取到模块级 _SEEDANCE_* / _WAN_*，
-    风格段与人物段通过 _build_style_section / _build_character_section 组装，
-    输出内容与旧版完全一致。
+    #2311 短音色样本模式：
+    - has_voice_ref=True 时，prompt 顶部加音色提取指令（"提取音频1的音色特征，
+      让角色用该音色说出以下台词"），台词用"台词"标签标注
+    - has_voice_ref=False 时，用模型原生音色（无reference_audio），台词用"配音/口播"标签
    """
    from packages.domain.points_rules import get_viral_video_model_config

@@ -2253,6 +2497,20 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
        # ========== Wan 3.0 格式 ==========
        head = f"{theme}，{style_text or _WAN_HEAD_SCENE_FALLBACK}，总时长{total_duration}秒。"
        body: list[str] = [head, ""]
+        # #2311 短音色样本：加音色提取指令
+        if has_voice_ref:
+            _all_voice = []
+            for s in shots:
+                v = str(s.get("voiceover", "") or "").strip()
+                if v:
+                    _all_voice.append(v)
+            _full_script = " ".join(_all_voice).strip()
+            if _full_script:
+                body.append(f'【音色与台词】提取音频1中的音色特征，让角色自然流畅说出："{_full_script}"')
+                body.append(
+                    "【口型与神态】语音自然驱动口型开合，保持音色全程一致，说话时伴随自然的眨眼、表情变化和轻微动作。"
+                )
+                body.append("")
        for i, s in enumerate(shots):
            tr = str(s.get("time_range", ""))
            pr = _parse_shot_seconds(tr)
@@ -2263,7 +2521,8 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
            if cam:
                pieces.append(f"运镜：{cam}")
            pieces.append(f"画面：{_shot_visual(s)}")
-            if voice:
+            if voice and not has_voice_ref:
+                # 有音色参考时台词已在顶部统一写，避免双指令源
                pieces.append(f'配音："{voice}"')
            body.append(" ".join(pieces))
        style_parts = [str(x) for x in hc if x] + [str(x) for x in np_list if x]
@@ -2279,6 +2538,10 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
    lines.append(f"- 总时长：{total_duration}秒{_SEEDANCE_DURATION_FOOT}")
    lines.append(f"- 画幅：{aspect_ratio}")
    lines.append(f"- 整体风格：{style_text}")
+    if has_voice_ref:
+        lines.append(
+            "- 音色：Use the voice from @Audio1 to say exactly the dialogue written below, maintain the same voice throughout, lip-sync naturally"
+        )
    lines.append("")

    if images:
@@ -2287,8 +2550,22 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
            ri = _ref(i)
            if ri is not None:
                lines.append(f"镜头{i + 1}参考@图片{ri + 1}")
+        if has_voice_ref:
+            lines.append("音色参考@Audio1")
        lines.append("")

+    # #2311 短音色样本：有音色参考时统一在顶部写完整台词（避免镜头级分散导致模型截断）
+    if has_voice_ref:
+        _all_voice = []
+        for s in shots:
+            v = str(s.get("voiceover", "") or "").strip()
+            if v:
+                _all_voice.append(v)
+        _full_script = " ".join(_all_voice).strip()
+        if _full_script:
+            lines.append(f'【台词】Use the voice from @Audio1 to say exactly: "{_full_script}"')
+            lines.append("")
+
    lines.append(_SEEDANCE_SCRIPT_HEADER)
    for i, s in enumerate(shots):
        tr = str(s.get("time_range", ""))
@@ -2301,7 +2578,8 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
        if cam:
            pieces.append(f"景别/运镜：{cam}")
        pieces.append(f"画面：{_shot_visual(s)}")
-        if voice:
+        if voice and not has_voice_ref:
+            # 有音色参考时台词已在顶部统一写
            pieces.append(f'口播："{voice}"')
        ri = _ref(i)
        if ri is not None:
@@ -2328,7 +2606,7 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
    from packages.domain.points_rules import get_viral_video_model_config
    from packages.shared.ai_service import call_video_generation

-    prompt = _assemble_seedance_prompt(copy_result, job)
+    prompt = _assemble_seedance_prompt(copy_result, job, has_voice_ref=bool(tts_audio_url))
    dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
    ratio = getattr(job, "video_ratio", None) or "9:16"
    model = getattr(job, "video_model", "") or None
@@ -2995,31 +3273,47 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
        logger.warning("[爆款视频][阶段3] 合规审核异常，继续出片: %s", e)
    _emit_progress(job_id, ViralVideoStage.REVIEW, 70.0, "合规审核完成")

-    voiceover = copy_result.get("voiceover_script", "") or job.effective_copy_text
+    copy_result.get("voiceover_script", "") or job.effective_copy_text

-    # provider 判断：Wan 3.0(dashscope) 在无自定义音色时跳过 TTS，用模型原生音频
+    # #2311 短音色样本模式：reference_audio 只传 ~8s 样本提取 speaker embedding，
+    # 口播文案全部放 prompt，由模型原生音画同生（generate_audio=true）。
+    # - Wan/Seedance + 无自定义音色 → 跳过TTS，模型纯原生音频
+    # - 有自定义音色 → 热点路径零TTS：预置音色用静态样本缓存、克隆音色用原始上传样本预处理缓存，仅异常兜底才TTS
    from packages.domain.points_rules import get_viral_video_model_config

    _vv_model = getattr(job, "video_model", "") or None
    _vv_provider = get_viral_video_model_config(_vv_model).get("provider", "doubao")
    _has_custom_voice = bool((getattr(job, "voice_id", "") or "").strip())
-    _skip_tts = _vv_provider == "dashscope" and not _has_custom_voice
+    _skip_tts = not _has_custom_voice  # 无自定义音色一律跳过TTS，原生音频

    tts_url = None
    if _skip_tts:
-        logger.info("[爆款视频][阶段3] Wan原生音频模式：跳过TTS，由模型生成配音/BGM/音效 job_id=%s", job_id)
+        logger.info(
+            "[爆款视频][阶段3] 原生音频模式：跳过TTS，由模型生成配音/BGM/音效 job_id=%s provider=%s",
+            job_id,
+            _vv_provider,
+        )
        _set_stage(job, repo, session, ViralVideoStage.TTS, "使用模型原生音频...")
        _emit_progress(job_id, ViralVideoStage.TTS, 78.0, "使用模型原生音频", {"has_tts": False, "native_audio": True})
    else:
-        # Step 5: TTS 整段合成
-        _set_stage(job, repo, session, ViralVideoStage.TTS, "正在合成AI配音...")
-        tts_path = _step_tts(job, voiceover)
-        # Bug3 fix: TTS 音频时长校验与修正
-        if tts_path is not None:
-            _dur_limit = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
-            tts_path = _check_and_fix_tts_duration(tts_path, _dur_limit)
-        tts_url = _upload_tts_to_oss(job, tts_path)
-        _emit_progress(job_id, ViralVideoStage.TTS, 78.0, "配音完成", {"has_tts": tts_url is not None})
+        _set_stage(job, repo, session, ViralVideoStage.TTS, "正在准备音色样本...")
+        tts_url = _resolve_voice_sample_audio(job)
+        logger.info(
+            "[爆款视频][阶段3] 音色样本准备完成 job_id=%s provider=%s model=%s voice_id=%s voice_source=%s has_url=%s",
+            job_id,
+            _vv_provider,
+            _vv_model or "",
+            (getattr(job, "voice_id", "") or "")[:24],
+            (getattr(job, "voice_source", "") or "") or "preset",
+            bool(tts_url),
+        )
+        _emit_progress(
+            job_id,
+            ViralVideoStage.TTS,
+            78.0,
+            "音色样本准备完成" if tts_url else "使用模型原生音频",
+            {"has_tts": tts_url is not None, "voice_sample_mode": True},
+        )

    # Step 6: 单次 Seedance（失败自动退款）
    _set_stage(job, repo, session, ViralVideoStage.RENDERING, "正在生成视频（约1-3分钟）...")
@@ -30,14 +30,20 @@ _warned: dict[str, float] = {}
 _warn_lock = threading.Lock()


-def _warn_once(key: str, template_type: str, e: BaseException | str | None = None) -> None:
+def _warn_once(key: str, context: str, detail: BaseException | str | None = None) -> None:
+    """冷却告警通用方法：5 分钟内同 key 只打一次，避免日志刷屏。
+
+    - key: 告警分组键（去重用）
+    - context: 本条告警的简短场景描述（直接拼进日志，不要再写死 'fallback to default' 字样）
+    - detail: 具体异常或描述
+    """
    now = time.monotonic()
    with _warn_lock:
        last = _warned.get(key, 0.0)
        if now - last < _WARN_COOLDOWN_SECONDS:
            return
        _warned[key] = now
-    logger.warning("viral_video prompt template %s fallback to default: %s", template_type, e)
+    logger.warning("viral_video prompt %s: %s", context, detail)


@dataclass
@@ -85,10 +91,10 @@ def _get_session():

        url = str(get_shared_settings().database_url)
    except Exception as e:  # noqa: BLE001
-        _warn_once("session_config", "db_config", e)
+        _warn_once("session_config", "db config failed (will fallback)", e)
        return None
    if not url:
-        _warn_once("session_url", "db_url", "database_url empty")
+        _warn_once("session_url", "db_url empty (will fallback)", "database_url empty")
        return None
    url = url.replace("postgresql+asyncpg://", "postgresql+psycopg://")
    if url.startswith("postgresql://") and "psycopg" not in url:
@@ -128,14 +134,14 @@ def _load_from_db(prompt_type: str) -> Optional[PromptTemplate]:
            is_active=bool(row[6]),
        )
    except Exception as e:  # noqa: BLE001 - 表不存在/DB 不可用时静默回落
-        _warn_once(f"load_db:{prompt_type}", prompt_type, e)
+        _warn_once(f"load_db:{prompt_type}", f"db load failed for {prompt_type} (will fallback to code default)", e)
        return None
    finally:
        if session is not None:
            try:
                session.close()
            except Exception as e:  # noqa: BLE001
-                _warn_once(f"session_close:{prompt_type}", prompt_type, e)
+                _warn_once(f"session_close:{prompt_type}", f"session close failed for {prompt_type}", e)


 def get_template(prompt_type: str, *, force_refresh: bool = False) -> Optional[PromptTemplate]:
@@ -151,7 +157,11 @@ def get_template(prompt_type: str, *, force_refresh: bool = False) -> Optional[P

    template = _load_from_db(prompt_type) or _fallback(prompt_type)
    if template is None:
-        _warn_once(f"no_template:{prompt_type}", prompt_type, "no template in db or code defaults")
+        _warn_once(
+            f"no_template:{prompt_type}",
+            f"NO TEMPLATE for {prompt_type} (neither db nor code defaults)",
+            "no template in db or code defaults",
+        )
    if template is not None:
        with _lock:
            _cache[prompt_type] = (now, template)
@@ -166,17 +176,75 @@ def invalidate() -> None:
        _warned.clear()


-class _SafeDict(dict):
-    def __missing__(self, key: str) -> str:
-        return "{" + key + "}"
-
-
 def _safe_format(text: str, kwargs: dict) -> str:
-    try:
-        return text.format_map(_SafeDict(kwargs))
-    except Exception as e:  # noqa: BLE001
-        _warn_once("safe_format", "format", e)
-        return text
+    """安全格式化模板：任何占位符异常（缺key/属性访问失败/格式符非法/JSON花括号样例）
+    都原样保留，不抛异常、不静默回落默认模板，只打一条warning告警。
+
+    不用 str.format_map(_SafeDict)：_SafeDict 只能防 KeyError，无法防 AttributeError
+    （如 {duration.1格式写全6项参数} 访问 str.1格式... 会抛 AttributeError），
+    也无法防非法 format_spec（如 JSON 样例 {"key":"val"} 里冒号被当格式符）。
+    """
+    import string as _string
+
+    out: list[str] = []
+    i, n = 0, len(text)
+    errors: list[str] = []
+    while i < n:
+        c = text[i]
+        if c == "{":
+            if i + 1 < n and text[i + 1] == "{":
+                out.append("{")
+                i += 2
+                continue
+            j = i + 1
+            depth = 1
+            while j < n and depth > 0:
+                if text[j] == "{":
+                    depth += 1
+                elif text[j] == "}":
+                    depth -= 1
+                j += 1
+            if depth == 0:
+                inner = text[i + 1 : j - 1]
+                try:
+                    val = _string.Formatter().vformat("{" + inner + "}", (), kwargs)
+                    out.append(val)
+                except Exception as e:  # noqa: BLE001
+                    # 非法占位符原样保留，避免模板里的样例JSON/说明文本炸format
+                    errors.append(f"{inner!r}: {type(e).__name__}")
+                    out.append(text[i:j])
+                i = j
+                continue
+        if c == "}":
+            if i + 1 < n and text[i + 1] == "}":
+                out.append("}")
+                i += 2
+                continue
+        out.append(c)
+        i += 1
+    if errors:
+        # 模板里花括号炸format分两种：
+        # 1. 配置bug（如v5事故{duration.1格式写全6项参数}非法属性访问）→ 需要排查
+        # 2. 样例文本花括号（如{大远景/远景/全景}给LLM看的景别选项、JSON示例、XML说明）→ 原样保留属正常
+        # 注意：本函数只原样保留，不会让template回落到DEFAULT_TEMPLATES；
+        # 真正的回落到代码默认模板只在DB查询失败/无激活模板时发生（见_load_from_db/get_template），
+        # 那里会打独立ERROR日志并带job_id（见viral_video.py中DEFAULT_TEMPLATES检测块）。
+        # #2310: 降级为WARNING，因为大部分是DB模板中的样例花括号正常保留，不是bug；
+        # 仅在错误数异常多(>10)或包含明显属性访问错误(.xxx/[xxx])时提级到WARNING加关注
+        _has_real_bug = any("." in e or "[" in e for e in errors[:5])
+        _log_fn = logger.warning if _has_real_bug else logger.debug
+        _log_fn(
+            "[viral][prompt] 模板格式化有%d个花括号原样保留（示例：%s）——"
+            "样例文本(景别选项/JSON示例)可忽略；若应为变量请检查占位符拼写",
+            len(errors),
+            errors[0],
+        )
+        _warn_once(
+            "safe_format",
+            f"format-keep({len(errors)}placeholders)",
+            f"{len(errors)}个占位符原样保留（首条{errors[0]}），未回落默认模板",
+        )
+    return "".join(out)


 def render_system_prompt(template: PromptTemplate, **kwargs) -> str:
@@ -113,7 +113,9 @@ _STORYBOARD_SYSTEM = (
    <clip index="1">
      <time_range>0-3秒</time_range>
      <voiceover>这一镜的口播台词</voiceover>
-      <visual>具体、有画面感的镜头描述（主体/动作/镜头运动/景别/光线）</visual>
+      <shot>景别、机位高度、焦段、光圈、运镜（6项摄影参数，简短，如『中近景平视，50mm f/1.8，缓慢推镜』）</shot>
+      <visual>画面主体、场景、光线、氛围（只写画面内容，不含景别/运镜参数，如『暖光木色门店，店员微笑递茶，背景虚化』）</visual>
+      <action_details>人物动作、表情、物品操作细节（真人出镜镜头必填，纯产品/空镜填空线『-』）</action_details>
      <reference_image_index>0</reference_image_index>
    </clip>
  </clips>
@@ -126,7 +128,7 @@ _STORYBOARD_SYSTEM = (

 ## 写作要求
 1. 口播台词：像真人面对镜头说话，短句、口语化、有停顿有情绪，开头 3 秒给出钩子；不要书面腔，不要机械报参数。
-2. 画面描述：写清"观众会看到什么"，有动作、有镜头运动、有景别和光线，具体可拍；不堆砌形容词，不写无法实现的画面。
+2. <shot> 只写6项摄影参数（景别/机位高度/焦段/光圈/运镜），简短不超过20字；<visual> 写观众会看到的画面主体/场景/光线/氛围；<action_details> 写人物动作/表情/物品操作细节（真人出镜必填）。三字段各司其职，不要混写。
 3. copy_display_markdown：直接展示给最终用户的文案，用 Markdown 写成自然、流畅、有感染力的成片文案，可用小标题与短句组织；不要做字段列表，不要出现"镜头一/台词："这类制作说明。
 4. 内容必须来自图片观察与用户给出的信息，不编造卖点、不夸大、不使用绝对化用语和虚假承诺。
 5. reference_image_index 填本镜参考图片序号（从 0 开始），没有合适参考图填 -1。
@@ -167,19 +169,25 @@ _STORYBOARD_EXAMPLE = """<script>
    <clip index="1">
      <time_range>0-4秒</time_range>
      <voiceover>生完娃，是不是连照镜子的勇气都没了？</voiceover>
-      <visual>中近景，暖光下一位妈妈略显疲惫地看向镜中，镜头缓缓推近</visual>
+      <shot>中近景平视，50mm f/2.0，缓慢推镜</shot>
+      <visual>暖光浴室镜前，妈妈略显疲惫地看向镜中自己，镜灯打在侧脸</visual>
+      <action_details>妈妈手扶洗手台，微微低头再抬眼看镜，眼眶微红</action_details>
      <reference_image_index>0</reference_image_index>
    </clip>
    <clip index="2">
      <time_range>4-10秒</time_range>
      <voiceover>推开门，暖光和一杯热茶先接住了你。</voiceover>
-      <visual>中景，门店前台暖光木色装修，店员微笑递上茶水，镜头平移扫过环境</visual>
+      <shot>中景平视，35mm f/2.8，镜头缓慢平移</shot>
+      <visual>门店前台暖光木色装修，茶水冒热气，背景柔和虚化</visual>
+      <action_details>店员微笑双手递上茶杯，妈妈伸手接过，指尖轻触杯沿</action_details>
      <reference_image_index>1</reference_image_index>
    </clip>
    <clip index="3">
      <time_range>10-15秒</time_range>
      <voiceover>来御众堂躺一会儿，整个人都松了。</voiceover>
-      <visual>特写，妈妈闭眼放松做护理，暖光打在侧脸，最后镜头拉远露出品牌LOGO</visual>
+      <shot>近景缓慢拉远至中景，85mm f/1.8，匀速后拉</shot>
+      <visual>护理床暖光打在妈妈侧脸，最后画面露出御众堂品牌LOGO</visual>
+      <action_details>妈妈闭眼放松，嘴角微微上扬，肩膀缓缓下沉</action_details>
      <reference_image_index>2</reference_image_index>
    </clip>
  </clips>
@@ -123,7 +123,7 @@ class Reviewer:
            messages,
            temperature=0.2,
            max_tokens=1024,
-            timeout=25,
+            timeout=15,  # #2307: 25s过长，超时降级放行
        )
        if not raw:
            return None
@@ -21,7 +21,8 @@ class PresetVoice:
        description: 音色描述
        gender: 性别（male/female）
        language: 语言代码（如 zh-CN）
-        preview_url: 预览音频 URL（可选）
+        preview_url: 预览音频 URL（可选，前端试听用，短文本）
+        sample_audio_url: 音色样本音频 URL（#2311，reference_audio用，8秒固定文案，部署预生成）
        tags: 标签列表
    """

@@ -31,6 +32,9 @@ class PresetVoice:
    gender: str
    language: str = "zh-CN"
    preview_url: str = ""
+    sample_audio_url: str = (
+        ""  # #2311: 用于reference_audio的8秒静态音色样本公网URL（部署时由seed_preset_voice_samples.py预生成）
+    )
    tags: list[str] | None = None

    def to_dict(self) -> dict:
@@ -42,6 +46,7 @@ class PresetVoice:
            "gender": self.gender,
            "language": self.language,
            "preview_url": self.preview_url,
+            "sample_audio_url": self.sample_audio_url,
            "tags": self.tags or [],
        }

@@ -56,6 +61,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="female",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/a8fb29a2-7d44-9f16-9dab-f17fa806cf3d.mp3?Expires=1785716123&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=c74kFX%2BcDpmuBny2dPBkvak01oY%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longxiaochun_v3_sample.mp3",
        tags=["知性", "女声", "助手"],
    ),
    PresetVoice(
@@ -65,6 +71,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="female",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/d065a6da-de0d-9ad2-b64a-00aaf9aa3dea.mp3?Expires=1785716203&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=9hRh7xbmS8eDdiBDDFUatvad1jg%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longxiaoxia_v3_sample.mp3",
        tags=["沉稳", "女声", "播报"],
    ),
    PresetVoice(
@@ -74,6 +81,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="male",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/b139d925-4669-979d-89db-2dc16945b577.mp3?Expires=1785730686&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=X2p%2FA7Dm%2BTNfeFilW%2FQO9VoYPLQ%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longsanshu_v3_sample.mp3",
        tags=["沉稳", "男声", "有声书"],
    ),
    PresetVoice(
@@ -83,6 +91,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="female",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/1ea86db3-4dc9-927e-9b76-c83d89f6788c.mp3?Expires=1785716207&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=Vg0ckgSgPzgNQ%2FNdeJzrE%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longyue_v3_sample.mp3",
        tags=["温暖", "女声", "广告"],
    ),
    PresetVoice(
@@ -92,6 +101,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="male",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/6b47a5ce-ae21-995a-ac8b-5e55ed04ec45.mp3?Expires=1785716209&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=zm6WGmthpwTjf8PzgNQ%2FNdeJzrE%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longshu_v3_sample.mp3",
        tags=["沉稳", "男声", "教育"],
    ),
    PresetVoice(
@@ -101,6 +111,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="female",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/13bf9f6a-72a9-949c-a896-ec3e6b46338b.mp3?Expires=1785730692&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=pxrP2rYmOumc4dw9TD8TZf%2B27gk%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longyingjing_v3_sample.mp3",
        tags=["冷静", "女声", "纪录片"],
    ),
    PresetVoice(
@@ -110,6 +121,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="male",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/5e69f0c1-45fa-92ca-ae5c-38272f231efb.mp3?Expires=1785730692&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=uREIiIUFrN7CpJ7avXKWCo04A6I%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longshuo_v3_sample.mp3",
        tags=["干练", "男声", "科技"],
    ),
    PresetVoice(
@@ -119,6 +131,7 @@ PRESET_VOICES: list[PresetVoice] = [
        gender="female",
        language="zh-CN",
        preview_url="http://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/prod/cosyvoice-v3-flash/20260802/1320180660510733/a38e4c05-25bf-9640-92d6-0b4a14555cdb.mp3?Expires=1785716215&OSSAccessKeyId=LTAI5tGzqbGcEmE58b221XQy&Signature=WtbsqqfR4A6OhpL%2B%2Fe0lh7YBMWM%3D",
+        sample_audio_url="https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/static/voice-samples/preset/longtian_v3_sample.mp3",
        tags=["活泼", "女声", "短视频"],
    ),
 ]
@@ -286,6 +286,11 @@ class DoubaoClient:
        if not self.is_available:
            return None

+        # #2304 追加修复：每次调用必须重置 finish_reason 标记，避免跨调用泄漏导致误判截断（Bug:
+        # attempt1 收到 finish=length 后 last_finish_reason='length'，attempt2 开头 chunk 处理后
+        # 立即读到这个陈旧值误判截断、丢弃好内容）。
+        self.last_finish_reason = ""
+
        url = f"{self.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
@@ -302,6 +307,11 @@ class DoubaoClient:
        if self.extra_params:
            payload.update(self.extra_params)
        if kwargs:
+            # #2303 追加修复：豆包 ark 端点不认 enable_thinking 布尔参数，
+            # 必须用 thinking={"type":"disabled"} 才会真正关闭 reasoning（A/B实测TTFT从100s+→2s）。
+            # 调用方传 enable_thinking=False 时，统一翻译为标准字段。
+            if kwargs.pop("enable_thinking", None) is False:
+                payload["thinking"] = {"type": "disabled"}
            payload.update(kwargs)

        last_error: Optional[Exception] = None
@@ -405,10 +415,21 @@ class DoubaoClient:
        if self.extra_params:
            payload.update(self.extra_params)
        if kwargs:
+            # 同同步路径：enable_thinking=False 翻译为 thinking={"type":"disabled"}
+            _thinking_disabled = kwargs.pop("enable_thinking", None) is False
            payload.update(kwargs)
+            if _thinking_disabled:
+                payload["thinking"] = {"type": "disabled"}
+
+        # 每次调用 reset finish_reason 标记（同 chat_completion，避免跨 attempt 泄漏误判截断）
+        self.last_finish_reason = ""

        _t0 = time.time()
-        for attempt in range(self.max_retries + 1):
+        # #2303 追加修复：流式路径 SDK 内重试对 read timeout 翻倍耗时（单 attempt 外层120s +
+        # SDK 重试120s*2 ≈ 360-420s），把两个 attempt 总预算吃光。外层 _step_script_generation
+        # 已实现 fast→fast→pro 三级attempt兜底，这里禁 SDK 内重试，超时/错误直接抛给外层。
+        _stream_max_retries = 0
+        for attempt in range(_stream_max_retries + 1):
            try:
                _req_timeout = httpx.Timeout(connect=10.0, read=120.0, write=30.0, pool=10.0)
                if timeout:
@@ -439,6 +460,8 @@ class DoubaoClient:
                            continue
                        delta = choices[0].get("delta") or {}
                        content_piece = delta.get("content") or ""
+                        # reasoning_content 由豆包 seed-2-1-pro 思考模式返回，
+                        # 对模板填充式生成无用（会占满 deadline），调用方可传 enable_thinking=False 关闭
                        if content_piece:
                            yield content_piece
                        finish_reason = choices[0].get("finish_reason")
@@ -453,19 +476,13 @@ class DoubaoClient:
                )
                return
            except Exception as e:
-                if attempt < self.max_retries:
-                    wait = 0.5 * (2**attempt)
-                    logger.warning(
-                        "豆包流式API失败，%.1fs后重试 (%d/%d, elapsed=%.1fs): %s",
-                        wait,
-                        attempt + 1,
-                        self.max_retries + 1,
-                        time.time() - _t0,
-                        e,
-                    )
-                    time.sleep(wait)
-                    continue
-                logger.error("豆包流式API最终失败: elapsed=%.1fs err=%s", time.time() - _t0, e)
+                # 流式路径禁 SDK 内重试（外层 attempt 机制兜底，避免单 attempt 被 SDK 重试拖到 300s+）
+                err_msg = str(e).lower()
+                is_timeout = any(k in err_msg for k in ("timeout", "timed out", "deadline", "超时", "read timeout"))
+                if is_timeout:
+                    logger.warning("[doubao] 流式 read timeout elapsed=%.1fs，交外层attempt处理", time.time() - _t0)
+                else:
+                    logger.error("豆包流式API失败: elapsed=%.1fs err=%s", time.time() - _t0, e)
                return

    def vision_completion(
@@ -536,8 +553,12 @@ class DoubaoClient:
        if self.extra_params:
            payload.update(self.extra_params)
        if kwargs:
+            if kwargs.pop("enable_thinking", None) is False:
+                payload["thinking"] = {"type": "disabled"}
            payload.update(kwargs)

+        self.last_finish_reason = ""  # reset，避免跨调用泄漏
+
        req_timeout = self._resolve_timeout(timeout or self.timeout)
        last_error: Optional[Exception] = None
        _t0 = time.time()
@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""#2311 部署脚本：为所有预置音色生成 8 秒静态音色样本并上传 OSS。
+
+用途：
+  部署 release 时执行，为 8 个 CosyVoice 预置音色各合成一段 8s 固定样本 mp3，
+  上传到 OSS 永久路径 static/voice-samples/preset/{voice_id}_sample.mp3（公共读），
+  并回填 packages/domain/preset_voices.py 中 PresetVoice.sample_audio_url 字段。
+
+  出片流程通过 PresetVoice.sample_audio_url 直接读取，零 TTS、零冷启动。
+
+用法（在 API/Worker 容器内执行）：
+  cd /path/to/project && python scripts/seed_preset_voice_samples.py
+
+可选参数：
+  --dry-run          只打印要做的操作，不实际上传/回填
+  --force            强制重新生成（即使 OSS 已有文件也覆盖）
+  --no-public        上传后不设公共读 ACL（默认公共读）
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import re
+import sys
+import tempfile
+import urllib.request
+from pathlib import Path
+
+import oss2  # type: ignore
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger(__name__)
+
+# 固定 8 秒样本文案（覆盖中文四声、开合齐撮、常见声母韵母）
+VOICE_SAMPLE_SCRIPT_ZH = "你好，很高兴认识你。今天天气真不错，我们一起来看看吧。"
+VOICE_SAMPLE_SCRIPT_EN = "Hi there, nice to meet you. The weather is great today, let's take a look together."
+
+# OSS 路径前缀
+OSS_PREFIX = "static/voice-samples/preset/"
+
+
+def _project_root() -> Path:
+    """定位项目根目录（兼容容器/软链/不同 cwd 场景）。"""
+    script_dir = Path(__file__).resolve().parent
+    candidate = script_dir.parent
+    if (candidate / "packages" / "domain" / "preset_voices.py").exists():
+        return candidate
+    cur = Path.cwd().resolve()
+    for _ in range(8):
+        if (cur / "packages" / "domain" / "preset_voices.py").exists():
+            return cur
+        if cur.parent == cur:
+            break
+        cur = cur.parent
+    raise RuntimeError("无法定位项目根目录，请在项目根目录或 scripts/ 下执行本脚本")
+
+
+PROJECT_ROOT = _project_root()
+PRESET_VOICES_PY = PROJECT_ROOT / "packages" / "domain" / "preset_voices.py"
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+def _get_cosyvoice_service():
+    from packages.application.cosyvoice_service import CosyVoiceService
+
+    return CosyVoiceService()
+
+
+def _get_oss_bucket():
+    endpoint = os.environ.get("OSS_ENDPOINT", "")
+    ak = os.environ.get("OSS_ACCESS_KEY_ID", "")
+    sk = os.environ.get("OSS_ACCESS_KEY_SECRET", "")
+    bucket_name = os.environ.get("OSS_BUCKET_NAME", "")
+
+    if not (endpoint and ak and sk and bucket_name):
+        try:
+            from packages.shared.config import get_shared_settings
+
+            s = get_shared_settings()
+            endpoint = endpoint or getattr(s, "oss_endpoint", "")
+            ak = ak or getattr(s, "oss_access_key_id", "")
+            sk = sk or getattr(s, "oss_access_key_secret", "")
+            bucket_name = bucket_name or getattr(s, "oss_bucket_name", "")
+        except Exception as e:
+            logger.warning("从settings读取OSS配置失败: %s", e)
+
+    if not (endpoint and ak and sk and bucket_name):
+        logger.error("缺少 OSS 凭证，请设置 OSS_ENDPOINT/OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET/OSS_BUCKET_NAME")
+        sys.exit(3)
+
+    ep_host = endpoint.replace("https://", "").replace("http://", "").rstrip("/")
+    auth = oss2.Auth(ak, sk)
+    bucket = oss2.Bucket(auth, f"https://{ep_host}", bucket_name)
+    public_base = f"https://{bucket_name}.{ep_host}"
+    return bucket, public_base
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="#2311 预生成预置音色静态样本")
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--force", action="store_true")
+    ap.add_argument("--no-public", action="store_true")
+    args = ap.parse_args()
+
+    from packages.domain.preset_voices import PRESET_VOICES
+
+    logger.info("项目根目录: %s", PROJECT_ROOT)
+    logger.info("预置音色总数: %d", len(PRESET_VOICES))
+    cosy = _get_cosyvoice_service()
+    bucket, public_base = _get_oss_bucket()
+
+    voice_to_url: dict[str, str] = {}
+    tmpdir = Path(tempfile.mkdtemp(prefix="preset_voice_sample_"))
+
+    for voice in PRESET_VOICES:
+        voice_id = voice.voice_id
+        lang = voice.language or "zh-CN"
+        script = VOICE_SAMPLE_SCRIPT_EN if lang.lower().startswith("en") else VOICE_SAMPLE_SCRIPT_ZH
+        oss_key = f"{OSS_PREFIX}{voice_id}_sample.mp3"
+        public_url = f"{public_base}/{oss_key}"
+
+        exists = False
+        try:
+            exists = bucket.object_exists(oss_key)
+        except Exception as e:
+            logger.warning("[%s] 检查OSS存在失败: %s", voice_id, e)
+
+        if exists and not args.force:
+            logger.info("[%s] OSS已有样本，跳过: %s", voice_id, public_url)
+            voice_to_url[voice_id] = public_url
+            continue
+
+        if args.dry_run:
+            logger.info("[DRY][%s] 将合成「%s...」→ oss://%s", voice_id, script[:20], oss_key)
+            voice_to_url[voice_id] = public_url
+            continue
+
+        logger.info("[%s] 合成中...", voice_id)
+        try:
+            result = cosy.synthesize_speech(text=script, voice_id=voice_id, format="mp3")
+        except Exception as e:
+            logger.error("[%s] TTS合成失败: %s", voice_id, e, exc_info=True)
+            continue
+
+        local_mp3 = tmpdir / f"{voice_id}_sample.mp3"
+        try:
+            urllib.request.urlretrieve(result.audio_url, str(local_mp3))
+            size = local_mp3.stat().st_size
+            if size < 4096:
+                logger.error("[%s] 文件太小(%d)，跳过", voice_id, size)
+                continue
+            logger.info("[%s] 下载完成: %d bytes", voice_id, size)
+        except Exception as e:
+            logger.error("[%s] 下载合成音频失败: %s", voice_id, e, exc_info=True)
+            continue
+
+        logger.info("[%s] 上传 oss://%s", voice_id, oss_key)
+        try:
+            headers = {"Content-Type": "audio/mpeg"}
+            bucket.put_object_from_file(oss_key, str(local_mp3), headers=headers)
+            if not args.no_public:
+                bucket.put_object_acl(oss_key, oss2.OBJECT_ACL_PUBLIC_READ)
+            logger.info("[%s] 上传完成 → %s", voice_id, public_url)
+            voice_to_url[voice_id] = public_url
+        except Exception as e:
+            logger.error("[%s] OSS上传失败: %s", voice_id, e, exc_info=True)
+            continue
+
+    if not voice_to_url:
+        logger.warning("没有成功生成任何样本")
+        return 1
+
+    if args.dry_run:
+        logger.info("[DRY] 将回填 %d 个 sample_audio_url 到 %s", len(voice_to_url), PRESET_VOICES_PY)
+        for vid, url in voice_to_url.items():
+            logger.info("  %s → %s", vid, url)
+        return 0
+
+    text = PRESET_VOICES_PY.read_text(encoding="utf-8")
+    orig = text
+    for voice_id, url in voice_to_url.items():
+        block_pat = re.compile(
+            r'(PresetVoice\([^)]*?voice_id="' + re.escape(voice_id) + r'"[^)]*?sample_audio_url=)"[^"]*"',
+            re.DOTALL,
+        )
+        new_text, n = block_pat.subn(rf'\1"{url}"', text, count=1)
+        if n == 0:
+            logger.warning("[%s] 回填sample_audio_url失败（正则未匹配），请手动添加", voice_id)
+            continue
+        text = new_text
+
+    if text != orig:
+        PRESET_VOICES_PY.write_text(text, encoding="utf-8")
+        logger.info("✅ 已回填 %d 条 sample_audio_url 到 %s", len(voice_to_url), PRESET_VOICES_PY)
+        logger.info("下一步：git diff packages/domain/preset_voices.py 确认后 commit & push")
+    else:
+        logger.info("preset_voices.py 未变化（可能全部已回填过）")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -486,6 +486,87 @@ class TestLipsyncServiceUnit:
        assert call_kwargs.kwargs["audio_url"] == "https://example.com/direct-audio.mp3"
        assert job.audio_url == "https://example.com/direct-audio.mp3"

+    def test_create_job_presynth_backfills_script_text(self, mock_mediakit, mock_cosyvoice):
+        """预合成模式 script_text 为空时，从 sentence_timings 拼接兜底.
+
+        背景：前端 #1875 起预合成 payload 不带 script_text，导致 job.script_text
+        落库为空，Ditto 任务 script_len=0、情绪分析被跳过。
+        """
+        from app.services.lipsync_service import LipsyncService
+
+        mock_db = MagicMock()
+        svc = LipsyncService(
+            mock_db,
+            client=mock_mediakit,
+            cosyvoice_service=mock_cosyvoice,
+            voice_clone_repo=MagicMock(),
+        )
+        svc.settings.use_gpu_lipsync = False
+        timings = [
+            {"index": 0, "text": "大家好，", "start_time": 0.0, "end_time": 1.0},
+            {"index": 1, "text": "欢迎来到直播间。", "start_time": 1.0, "end_time": 2.5},
+        ]
+        job = svc.create_job(
+            user_id="user-1",
+            video_url="https://example.com/video.mp4",
+            audio_url="https://example.com/audio.mp3",
+            audio_duration=2.5,
+            sentence_timings=timings,
+            script_text="",  # 前端预合成 payload 不带文案
+        )
+
+        assert job.script_text == "大家好，欢迎来到直播间。"
+
+    def test_create_job_presynth_keeps_supplied_script_text(self, mock_mediakit, mock_cosyvoice):
+        """预合成模式显式传了 script_text 时，保留原文不做拼接."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_db = MagicMock()
+        svc = LipsyncService(
+            mock_db,
+            client=mock_mediakit,
+            cosyvoice_service=mock_cosyvoice,
+            voice_clone_repo=MagicMock(),
+        )
+        svc.settings.use_gpu_lipsync = False
+        timings = [
+            {"index": 0, "text": "分句一", "start_time": 0.0, "end_time": 1.0},
+        ]
+        job = svc.create_job(
+            user_id="user-1",
+            video_url="https://example.com/video.mp4",
+            audio_url="https://example.com/audio.mp3",
+            audio_duration=1.0,
+            sentence_timings=timings,
+            script_text="原始文案，含标点。",
+        )
+
+        assert job.script_text == "原始文案，含标点。"
+
+    def test_create_job_presynth_empty_timings_falls_back_to_direct(self, mock_mediakit, mock_cosyvoice):
+        """sentence_timings 空列表时 is_pre_synth=False，退化为直接音频模式（语义固化）."""
+        from app.services.lipsync_service import LipsyncService
+
+        mock_db = MagicMock()
+        svc = LipsyncService(
+            mock_db,
+            client=mock_mediakit,
+            cosyvoice_service=mock_cosyvoice,
+            voice_clone_repo=MagicMock(),
+        )
+        svc.settings.use_gpu_lipsync = False
+        job = svc.create_job(
+            user_id="user-1",
+            video_url="https://example.com/video.mp4",
+            audio_url="https://example.com/audio.mp3",
+            sentence_timings=[],
+            script_text="",
+        )
+
+        # 直接音频模式：无 timings 可拼，script_text 保持空（不报错）
+        assert job.script_text == ""
+        mock_mediakit.submit_lipsync.assert_called_once()
+

 class TestErrorHandling:
    """v3: 音色解析与错误码在 service 层处理，路由层做 HTTP 状态码映射."""
@@ -0,0 +1,165 @@
+"""#2308 parser字段映射单测：
+1. 新schema XML（<shot>+<visual>+<action_details>）字段正确映射
+2. 旧schema XML（<visual>混写）能拆分出cam/visual/action
+3. markdown中文分镜字段正确提取
+"""
+
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+
+@pytest.fixture
+def mock_job():
+    job = MagicMock()
+    job.duration = 15
+    job.video_ratio = "9:16"
+    job.marketing_purpose = "品牌推广"
+    job.image_analysis = {"images": []}
+    job.images = []
+    job.user_copy_text = ""
+    job.fusion_level = "ai_polish"
+    return job
+
+
+def _import_wv(monkeypatch):
+    for mod in ["worker_app.celery_app", "worker_app.db"]:
+        monkeypatch.setitem(sys.modules, mod, MagicMock())
+    import celery
+
+    monkeypatch.setattr(celery, "shared_task", lambda **kw: (lambda f: f))
+    from worker_app.tasks import viral_video as wv
+
+    return wv
+
+
+# === 1. 新schema（v3.1+）：独立 <shot>/<visual>/<action_details> ===
+NEW_SCHEMA_XML = """<script>
+  <copy_display_markdown><![CDATA[来御众堂放松一下]]></copy_display_markdown>
+  <clips>
+    <clip index="1">
+      <time_range>0-4秒</time_range>
+      <voiceover>生完娃，是不是连照镜子都没勇气了？</voiceover>
+      <shot>中近景平视，50mm f/2.0，缓慢推镜</shot>
+      <visual>暖光浴室镜前，妈妈略显疲惫地看向镜中自己，镜灯打在侧脸</visual>
+      <action_details>妈妈手扶洗手台，微微低头再抬眼看镜，眼眶微红</action_details>
+      <reference_image_index>0</reference_image_index>
+    </clip>
+    <clip index="2">
+      <time_range>4-10秒</time_range>
+      <voiceover>推开门，暖光和一杯热茶先接住了你。</voiceover>
+      <shot>中景平视，35mm f/2.8，缓慢平移</shot>
+      <visual>门店前台暖光木色装修，茶水冒热气，背景柔和虚化</visual>
+      <action_details>店员微笑双手递上茶杯，妈妈伸手接过，指尖轻触杯沿</action_details>
+      <reference_image_index>1</reference_image_index>
+    </clip>
+  </clips>
+  <voiceover_script>生完娃，是不是连照镜子都没勇气了？推开门，暖光和一杯热茶先接住了你。</voiceover_script>
+  <theme>产后妈妈走进御众堂</theme>
+</script>"""
+
+
+def test_new_schema_xml_field_mapping(monkeypatch, mock_job):
+    wv = _import_wv(monkeypatch)
+    r = wv._parse_script(NEW_SCHEMA_XML, mock_job)
+    assert r is not None
+    assert len(r["shots"]) == 2
+    s0 = r["shots"][0]
+    # shot_type 应是纯摄影参数
+    assert "缓慢推镜" in s0["shot_type_angle_movement"]
+    assert "妈妈" not in s0["shot_type_angle_movement"]
+    # scene_and_dialogue 只放对白
+    assert s0["scene_and_dialogue"] == "生完娃，是不是连照镜子都没勇气了？"
+    # action_details 有内容
+    assert "手扶洗手台" in s0["action_details"]
+    # visual 字段有画面
+    assert "暖光" in s0["visual"]
+    assert "推镜" not in s0["visual"]
+
+
+# === 2. 旧schema（无<shot>/<action_details>标签，<visual>混写）：兼容拆分 ===
+OLD_SCHEMA_XML = """<script>
+  <clips>
+    <clip index="1">
+      <time_range>0-4秒</time_range>
+      <voiceover>生完娃，是不是连照镜子都没勇气了？</voiceover>
+      <visual>中近景平视，暖光下妈妈略显疲惫地看向镜中，镜头缓缓推近，妈妈手扶洗手台微微低头</visual>
+      <reference_image_index>0</reference_image_index>
+    </clip>
+  </clips>
+  <voiceover_script>生完娃，是不是连照镜子都没勇气了？</voiceover_script>
+  <theme>测试</theme>
+</script>"""
+
+
+def test_old_schema_xml_splits_visual(monkeypatch, mock_job):
+    wv = _import_wv(monkeypatch)
+    r = wv._parse_script(OLD_SCHEMA_XML, mock_job)
+    assert r is not None
+    s0 = r["shots"][0]
+    # 摄影参数段应含景别/运镜关键词
+    assert "中近景" in s0["shot_type_angle_movement"] or "推" in s0["shot_type_angle_movement"]
+    # action_details 应含动作动词
+    assert s0["action_details"], "action_details不应为空（旧schema兼容应从<visual>拆出动作）"
+    assert "手扶" in s0["action_details"] or "低头" in s0["action_details"]
+    # scene_and_dialogue 只放对白，不重复画面
+    assert s0["scene_and_dialogue"] == "生完娃，是不是连照镜子都没勇气了？"
+
+
+# === 3. markdown 中文分镜字段 ===
+MARKDOWN_INPUT = """## 视频主题：御众堂探店
+
+### 镜头1
+- 时间范围：0-4秒
+- 景别/角度与运镜：中近景平视，缓慢推镜
+- 画面：暖光下妈妈看向镜中，镜灯打在侧脸
+- 动作与真人细节：妈妈手扶洗手台，微微低头
+- 口播："生完娃，是不是连照镜子都没勇气了？"
+
+### 镜头2
+- 时间范围：4-10秒
+- 景别：中景平视
+- 运镜：缓慢平移
+- 画面：门店前台暖光木色装修
+- 动作：店员递茶，妈妈伸手接过
+- 口播："推开门，暖光和一杯热茶先接住了你"
+"""
+
+
+def test_markdown_field_extraction(monkeypatch, mock_job):
+    wv = _import_wv(monkeypatch)
+    r = wv._parse_script(MARKDOWN_INPUT, mock_job)
+    assert r is not None
+    assert len(r["shots"]) == 2
+    s0 = r["shots"][0]
+    assert "中近景" in s0["shot_type_angle_movement"]
+    assert "妈妈手扶洗手台" in s0["action_details"]
+    assert s0["scene_and_dialogue"] == "生完娃，是不是连照镜子都没勇气了？"
+
+
+# === 4. _split_visual_field 单元测试 ===
+def test_split_visual_field(monkeypatch):
+    wv = _import_wv(monkeypatch)
+    # 典型旧schema混写："中近景平视，暖光下一位妈妈略显疲惫地看向镜中，镜头缓缓推近，妈妈手扶洗手台微微低头"
+    cam, vis, act = wv._split_visual_field(
+        "中近景平视，暖光下妈妈略显疲惫地看向镜中，镜头缓缓推近，妈妈手扶洗手台微微低头"
+    )
+    assert cam and "中近景" in cam
+    # 动作/表情段应被拆到 act（含"妈妈"主语+动词）
+    assert act and ("手扶" in act or "疲惫" in act or "看向" in act)
+    # 画面描述允许为空（当动作段本身承载了主要画面信息时），或含"暖光"
+    assert (not vis) or ("暖光" in vis) or ("妈妈" in vis)
+
+
+def test_split_visual_short(monkeypatch):
+    wv = _import_wv(monkeypatch)
+    cam, vis, act = wv._split_visual_field("特写固定镜头")
+    # 切不开时整体归camera/visual，action为空
+    assert "特写" in cam
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])
@@ -1449,3 +1449,73 @@ class TestMarkdownTimeRangeSplit:
        assert result is not None
        assert len(result["shots"]) == 2
        assert "你好" in result["voiceover_script"]
+
+
+class TestSafeFormatHardening:
+    """_safe_format 必须能扛住模板里任意花括号样例，不抛异常、不返回半截内容"""
+
+    def test_missing_key_preserves_placeholder(self):
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format("时长{duration}秒，{missing}保留", {"duration": "20"})
+        assert out == "时长20秒，{missing}保留"
+
+    def test_attribute_access_on_string_preserves_placeholder(self):
+        """复现v5事故：{duration.1格式写全6项参数}访问str的非法属性不应抛异常"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format(
+            "总时长：{duration.1格式写全6项参数，第二个镜头起末尾加转场}秒",
+            {"duration": "20"},
+        )
+        assert "总时长：{duration.1格式写全6项参数，第二个镜头起末尾加转场}秒" == out
+
+    def test_json_braces_in_example_preserved(self):
+        """模板里给LLM示例JSON时，{"key":"val"}不应被当format_spec解析"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format(
+            '输出JSON示例：{"key": "value", "n": 1}，时长{duration}秒',
+            {"duration": "20"},
+        )
+        assert '{"key": "value", "n": 1}' in out
+        assert "时长20秒" in out
+
+    def test_chinese_example_braces_preserved(self):
+        """中文花括号示例如{示例}原样保留"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format("规则：不要输出{示例}这类占位符，时长{duration}", {"duration": "20"})
+        assert "{示例}" in out
+        assert "时长20" in out
+
+    def test_escaped_braces_unescaped(self):
+        """{{escaped}} 正常输出 {escaped}"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format("这是{{escaped}}，时长{duration}", {"duration": "20"})
+        assert out == "这是{escaped}，时长20"
+
+    def test_legitimate_attribute_access_works(self):
+        """合法对象属性访问仍正常工作"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        class Obj:
+            brand = "御众堂"
+
+        out = _safe_format("品牌：{obj.brand}", {"obj": Obj()})
+        assert out == "品牌：御众堂"
+
+    def test_invalid_format_spec_preserves_placeholder(self):
+        """{val:invalid} 这种坏格式符原样保留"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format("坏格式{val:invalid_chinese}结束", {"val": "x"})
+        assert "{val:invalid_chinese}" in out
+
+    def test_normal_format_spec_still_works(self):
+        """合法格式符 {n:>3} 仍正常工作"""
+        from packages.application.viral_video.prompt_loader import _safe_format
+
+        out = _safe_format("数字{n:>3}结束", {"n": 5})
+        assert "5" in out