fix(ai-avatar): finalize 接口超时延长到120s兜底抽帧可能耗时

后端 finalize 在未传 cover_url 时会兜底自动抽帧+写OSS+入库，默认30s超时不够，
E2E联调时多次出现 AxiosError: timeout of 30000ms exceeded。

alembic/versions/110_add_tone_to_viral_video_jobs.py

@@ -1,23 +0,0 @@
-"""add tone column to viral_video_jobs
-
-Revision ID: 110
-Revises: 109_add_viral_video_param_guides
-Create Date: 2026-10-14
-
-"""
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision = "110"
-down_revision = "109_add_viral_video_param_guides"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.execute("ALTER TABLE viral_video_jobs ADD COLUMN IF NOT EXISTS tone VARCHAR(50) NOT NULL DEFAULT ''")
-
-
-def downgrade() -> None:
-    op.execute("ALTER TABLE viral_video_jobs DROP COLUMN IF EXISTS tone")

apps/api/app/api/routes/viral_video.py

@@ -127,7 +127,6 @@ def _to_response(job) -> ViralVideoJobResponse:
         persona_id=job.persona_id,
         viral_structure=job.viral_structure,
         marketing_purpose=job.marketing_purpose,
-        tone=getattr(job, "tone", "") or "",
         bgm_preference=job.bgm_preference,
         duration=job.duration or 15,
         user_copy_text=job.user_copy_text,
@@ -405,7 +404,6 @@ def generate_copy(
     job.persona_id = request.persona_id or job.persona_id
     job.viral_structure = request.viral_structure or job.viral_structure
     job.marketing_purpose = request.marketing_purpose or job.marketing_purpose
-    job.tone = getattr(request, "tone", "") or job.tone
     job.bgm_preference = request.bgm_preference or job.bgm_preference
     if request.duration:
         job.duration = max(5, min(30, int(request.duration)))

apps/api/app/schemas/viral_video.py

@@ -36,7 +36,6 @@ class ShotScript(BaseModel):
     action_details: str = Field(default="", description="人物动作、表情、物品操作细节")
     audio_bgm: str = Field(default="", description="环境音+BGM提示")
     transition: str = Field(default="硬切", description="转场方式：硬切/淡入淡出/叠化")
-    voiceover: str = Field(default="", description="该镜头口播台词（纯对白，不含画面描述）")
     reference_image_index: int | None = Field(
         default=None, description="参考图片索引（0-based，对应上传的第几张产品图）"
     )
@@ -92,7 +91,6 @@ class GenerateCopyRequest(BaseModel):
     persona_id: str = ""
     viral_structure: str = ""
     marketing_purpose: str = ""
-    tone: str = ""
     bgm_preference: str = ""
     duration: int = Field(default=15, ge=5, le=30)
     user_copy_text: str = ""
@@ -158,7 +156,6 @@ class ViralVideoJobResponse(BaseModel):
     persona_id: str = ""
     viral_structure: str = ""
     marketing_purpose: str = ""
-    tone: str = ""
     bgm_preference: str = ""
     duration: int = 15
     user_copy_text: str = ""

apps/web/index.html

@@ -4,57 +4,17 @@
     <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 中文字体完全延迟加载
-         关键设计：所有 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 事件。 -->
+    <!-- 标题字体（#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"
+    />
     <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>

apps/worker/worker_app/tasks/viral_video.py

@@ -358,7 +358,7 @@ def _build_sparse_trusted_urls(
 
 
 def _validate_copy_result(normalized: dict, dur: int) -> str | None:
-    """校验 _parse_script 解析后的 copy_result：字数 / 镜头数 / 时间轴。
+    """校验 _script_from_xml 解析后的 copy_result：字数 / 镜头数 / 时间轴。
 
     返回 None 表示通过；否则返回中文 reason 字符串（用于日志）。
     """
@@ -389,38 +389,27 @@ def _validate_copy_result(normalized: dict, dur: int) -> str | None:
 
 def _call_llm_once(
     client,
-    system_prompt: str | None = None,
-    user_prompt: str | None = None,
+    system_prompt: str,
+    user_prompt: str,
     *,
-    messages: list[dict] | None = None,
     temperature: float,
     max_tokens: int,
     timeout: int,
     on_delta: Callable[[str, str], None] | None = None,
-    enable_thinking: bool | None = None,
 ) -> str | None:
     """统一封装同步/流式 LLM 调用；返回 raw 文本，失败返回 None。
 
-    - messages 直接传多轮消息列表（system/user/assistant，含 few-shot 示例）；
-    - 兼容旧签名：传 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):
         return None
-    if messages is None:
-        messages = [{"role": "system", "content": system_prompt or ""}, {"role": "user", "content": user_prompt or ""}]
+    messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]
     # 同步路径
     if on_delta is None or not hasattr(client, "chat_completion_stream"):
         try:
-            _kw: dict = {}
-            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
-            )
+            return client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout)
         except Exception as e:
             logger.warning("[爆款视频] LLM同步调用失败: %s", e)
             return None
@@ -447,41 +436,11 @@ def _call_llm_once(
         except Exception as e:
             logger.warning("[爆款视频] on_delta 回调失败: %s", e)
 
-    # 流式强制超时：底层 SDK 的 timeout 对部分流式实现不生效（doubao 实测跑 150s+ 未中断），
-    # 这里用外层 deadline + 迭代超时双重保护。
-    _stream_deadline = time.time() + max(timeout or 120, 30)
     stream_ok = False
-    _length_truncated = False  # finish_reason=length 时标记为被截断（raw 不完整，parser 必败）
     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, **_stream_kw
-        )
-        while True:
-            remaining = _stream_deadline - time.time()
-            if remaining <= 0:
-                logger.warning("[爆款视频] 流式调用超时（外层 deadline），使用已收到的部分内容 tmo=%d", timeout)
-                break
-            try:
-                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),
-                    )
-                break
-            except Exception as inner_e:
-                err_msg = str(inner_e).lower()
-                if any(k in err_msg for k in ("timeout", "timed out", "deadline", "超时")):
-                    logger.warning("[爆款视频] 流式调用SDK超时: %s", inner_e)
-                    break
-                raise
+        for chunk in client.chat_completion_stream(
+            messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout or 120
+        ):
             if chunk:
                 full_parts.append(chunk)
                 pending.append(chunk)
@@ -491,69 +450,20 @@ def _call_llm_once(
                 )
                 if should:
                     _flush(force=False)
-                # 早期格式预判：>800字仍无任何结构化标记（<script/<clips/<clip/镜头1/镜头一/第1镜），
-                # 基本可以判定 LLM 跑题输出自然语言，parser 必败，早停省 15~30s
-                _acc_len = sum(len(p) for p in full_parts)
-                if _acc_len > 800:
-                    _full_so_far = "".join(full_parts)
-                    if not any(
-                        k in _full_so_far
-                        for k in (
-                            "<script",
-                            "<clips",
-                            "<clip",
-                            "镜头1",
-                            "镜头 1",
-                            "镜头一",
-                            "第1镜",
-                            "第 1 镜",
-                            "0-",
-                            "0~",
-                            "1-",
-                            "2-",
-                            "3-",
-                            "4-",
-                            "5-",
-                        )
-                    ):
-                        logger.warning(
-                            "[爆款视频] 流式 early-break：前%d字无结构化标记 head=%r",
-                            _acc_len,
-                            _full_so_far[:80].replace("\n", "\\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
+        stream_ok = True
     except Exception as e:
         logger.warning("[爆款视频] 流式调用失败，降级同步: %s", e, exc_info=True)
         full_parts = []
-    if _length_truncated:
-        # max_tokens 截断：raw 不完整，不降级同步（同步也同样 max_tokens 照样截断）
-        return None
     if not stream_ok:
         try:
-            _fb_kw: dict = {}
-            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
-            )
+            raw = client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout)
         except Exception as e:
             logger.warning("[爆款视频] LLM同步降级调用失败: %s", e)
             return None
         if raw:
             full_parts = [raw]
             try:
-                on_delta(raw, raw)
+                on_delta(raw, raw)  # 一次性推送
             except Exception:
                 pass
         else:
@@ -1072,6 +982,33 @@ def _build_images_summary(image_analysis: dict) -> str:
     return "\n".join(lines)
 
 
+def _safe_json_loads(raw: str | dict | list | None):
+    if raw is None:
+        return None
+    if isinstance(raw, (dict, list)):
+        return raw
+    if not isinstance(raw, str):
+        return None
+    s = raw.strip()
+    if s.startswith("```"):
+        s = s.strip("`")
+        if s.startswith("json"):
+            s = s[4:].lstrip()
+    try:
+        return json.loads(s)
+    except Exception:
+        # 尝试截取第一个 { ... } 或 [ ... ]
+        try:
+            for open_c, close_c in (("{", "}"), ("[", "]")):
+                i = s.find(open_c)
+                j = s.rfind(close_c)
+                if i >= 0 and j > i:
+                    return json.loads(s[i : j + 1])
+        except Exception:
+            pass
+    return None
+
+
 def _fallback_script(job: ViralVideoJob) -> dict:
     """脚本生成失败时的极简兜底脚本。基于image_analysis注入真实商品/门店信息，避免生成脱离用户输入的通用文案。"""
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
@@ -1100,11 +1037,7 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     if job.user_copy_text:
         voiceover = job.user_copy_text
     elif _brand_str and _name_str:
-        # 品牌和名称相同（如 VLM 识别出 brand=御众堂, name=御众堂）时去重，避免"X家的X"重复
-        if _brand_str == _name_str or _brand_str in _name_str or _name_str in _brand_str:
-            voiceover = f"来看看{_name_str}"
-        else:
-            voiceover = f"来看看{_brand_str}家的{_name_str}"
+        voiceover = f"来看看{_brand_str}家的{_name_str}"
     elif _name_str:
         voiceover = f"来看看{_name_str}"
     else:
@@ -1123,7 +1056,6 @@ def _fallback_script(job: ViralVideoJob) -> dict:
             "audio_bgm": "轻快流行BGM",
             "transition": "结束",
             "reference_image_index": 0 if job.images else None,
-            "voiceover": voiceover,
         }
     ]
     base["voiceover_script"] = voiceover
@@ -1134,63 +1066,29 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     return base
 
 
-def _parse_script(raw: str, job: ViralVideoJob) -> dict | None:
-    """通用分镜解析器：支持三种 LLM 输出格式。
+def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None:
+    """解析 v3 XML 分镜为 copy_result；只解析、补字段，不改 LLM 文案。
 
-    格式A（纯 XML）：整段都是 <script>...</script> XML（v3/v4 格式）
-    格式B（双段式）：中文可读说明 + 末尾 <script>...</script> XML 块
-    格式C（纯中文 markdown 编号式）：无 <script> 标签，按"镜头1/镜头2..."分段
-    格式D（纯中文 markdown 时间式）：无 <script> 标签，按"0-4秒""4-8秒"时间范围开头分段（v6 格式）
-
-    解析策略：
-    1. 正则提取最后一个 <script>...</script> 块
-    2. 有 XML 块 → 走 XML 解析
-    3. 无 XML 块但有 v3 结构标签 → 裸 XML 解析
-    4. 都没有 → 走中文 markdown 兜底解析（C/D 两种分块方式自动识别）
-    5. 都解析不出 → 返回 None
+    v3 结构：copy_display_markdown / clips>clip(index)[time_range,voiceover,
+    visual,reference_image_index] / voiceover_script / theme。
     """
-    if not raw:
-        return None
+    from packages.application.viral_video import xml_parser as xp
 
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
     ratio = getattr(job, "video_ratio", None) or "9:16"
+    if not raw:
+        return None
+
     image_analysis = normalize_image_analysis(getattr(job, "image_analysis", None))
     default_theme = _determine_theme(image_analysis, getattr(job, "marketing_purpose", "") or "")
-
-    # ── 策略1: 提取 <script>...</script> 块（取最后一个）──
-    script_blocks = re.findall(r"<script[^>]*>(.*?)</script>", raw, re.DOTALL)
-    xml_body = script_blocks[-1] if script_blocks else ""
-
-    if xml_body:
-        return _parse_script_xml(xml_body, raw, job, dur, ratio, default_theme)
-
-    # ── 策略1.5: 无 <script> 包裹，但含 v3 结构标签（<clips>/<voiceover_script>）──
-    if re.search(r"<(?:clips|voiceover_script|copy_display_markdown)\b", raw):
-        return _parse_script_xml(raw, raw, job, dur, ratio, default_theme)
-
-    # ── 策略2: 纯中文 markdown 兜底 ──
-    return _parse_script_markdown(raw, job, dur, ratio, default_theme)
-
-
-def _parse_script_xml(
-    xml_body: str,
-    raw_full: str,
-    job: ViralVideoJob,
-    dur: int,
-    ratio: str,
-    default_theme: str,
-) -> dict | None:
-    """XML 路径解析（格式A/B）：从 <script> 块内提取 clip 结构。"""
-    from packages.application.viral_video import xml_parser as xp
-
-    theme = xp.text_of(xml_body, "theme") or xp.text_of(raw_full, "theme") or default_theme
+    theme = xp.text_of(raw, "theme") or default_theme
     base = _empty_copy_result(dur, ratio, theme=theme)
     base["overview"]["total_duration"] = dur
     base["title"] = theme
 
-    display_md = xp.text_of(xml_body, "copy_display_markdown")
+    display_md = xp.text_of(raw, "copy_display_markdown")
 
-    clips = xp.find_all(xml_body, "clip")
+    clips = xp.find_all(raw, "clip")
     shots: list[dict] = []
     voice_parts: list[str] = []
     for i, c in enumerate(clips):
@@ -1213,7 +1111,6 @@ def _parse_script_xml(
             "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,
-            "voiceover": voice or "",
         }
         shots.append(shot)
 
@@ -1221,7 +1118,7 @@ def _parse_script_xml(
         return None
     base["shots"] = shots
 
-    voiceover = xp.text_of(xml_body, "voiceover_script") or "。".join(voice_parts)
+    voiceover = xp.text_of(raw, "voiceover_script") or "。".join(voice_parts)
     if not voiceover.strip():
         return None
     base["voiceover_script"] = voiceover
@@ -1231,198 +1128,23 @@ def _parse_script_xml(
     return base
 
 
-# ── 中文 markdown 兜底解析正则（模块级常量，避免重复编译）────────────
-_MD_SHOT_SPLIT_RE = re.compile(
-    r"(?:^|\n)\s*(?:#{1,4}\s+)?(?:镜头\s*([一二三四五六七八九十\d]+)|第\s*([\d]+)\s*镜)",
-    re.IGNORECASE,
-)
-# 时间范围开头作为分镜分割标记（格式D兜底）：匹配行首的「0-4秒」「4~8秒」「12-20秒」等，
-# 可选前面带 markdown 标题符号或「镜头」字样
-_MD_TIME_SPLIT_RE = re.compile(
-    r"(?:^|\n)\s*(?:#{1,4}\s+)?(?:镜头[\s:：]*)?(\d+)\s*[-~到至]\s*(\d+)\s*秒",
-)
-_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]+)"),
-    "voiceover": re.compile(r"""(?:口播|对白|台词|配音|旁白|voiceover)\s*[:：]\s*""" r"""(?:"([^"]+)"|'([^']+)'|(\S[^
-]*?)(?=\n\s*(?:镜头|#|$)|\s*$))"""),
-    "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+)"),
-    "theme": re.compile(r"(?:视频主题|主题|title)\s*[:：]\s*([^\n]+)"),
-    "full_voiceover": re.compile(r"(?:口播全文|完整口播|口播稿)\s*[:：]\s*\n?([\s\S]+?)(?=\n\s*(?:镜头|#{1,4}\s|\Z))"),
-}
-
-
-def _parse_script_markdown(
-    raw: str,
-    job: ViralVideoJob,
-    dur: int,
-    ratio: str,
-    default_theme: str,
-) -> dict | None:
-    """中文 markdown 兜底解析（格式C/D）：按"镜头1/镜头一/第1镜"或"0-4秒/4-8秒"时间范围分割。"""
-    # 提取 theme
-    theme = default_theme
-    theme_match = _MD_FIELD_RE["theme"].search(raw)
-    if theme_match:
-        theme = theme_match.group(1).strip() or default_theme
-
-    # 按镜头标识分割文本（优先"镜头N/第N镜"，兜底"0-4秒"时间范围开头）
-    boundaries = []
-    for m in _MD_SHOT_SPLIT_RE.finditer(raw):
-        shot_num_str = m.group(1) or m.group(2)
-        # 中文数字转阿拉伯数字
-        cn_map = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "十": 10}
-        if shot_num_str in cn_map:
-            shot_num = cn_map[shot_num_str]
-        else:
-            try:
-                shot_num = int(shot_num_str)
-            except (ValueError, TypeError):
-                shot_num = 0
-        boundaries.append((m.start(), shot_num))
-
-    if not boundaries:
-        # 兜底：按行首时间范围标记分割（v6 中文分镜格式如「0-4秒」「4-8秒」）
-        # _MD_TIME_SPLIT_RE 已要求 (?:^|\n)\s*，天然是行首级匹配，不会误识别字段值内的时间范围
-        boundaries = [(m.start(), i + 1) for i, m in enumerate(_MD_TIME_SPLIT_RE.finditer(raw))]
-
-    if not boundaries:
-        return None
-
-    # 分割各镜头块
-    shot_blocks: list[tuple[int, str]] = []
-    for idx, (start, num) in enumerate(boundaries):
-        end = boundaries[idx + 1][0] if idx + 1 < len(boundaries) else len(raw)
-        block_text = raw[start:end]
-        shot_blocks.append((num, block_text))
-
-    base = _empty_copy_result(dur, ratio, theme=theme)
-    base["overview"]["total_duration"] = dur
-    base["title"] = theme
-
-    shots: list[dict] = []
-    voice_parts: list[str] = []
-
-    for i, (_, block) in enumerate(shot_blocks):
-        # 时间范围
-        time_match = _MD_TIME_RE.search(block)
-        if time_match:
-            t_start, t_end = time_match.group(1), time_match.group(2)
-            time_range = f"{t_start}-{t_end}秒"
-        else:
-            n = len(shot_blocks)
-            seg = dur / n if n else dur
-            time_range = f"{int(i * seg)}-{int((i + 1) * seg)}秒"
-
-        # 景别/运镜
-        visual = ""
-        camera = ""
-        vis_m = _MD_FIELD_RE["visual"].search(block)
-        if vis_m:
-            visual = vis_m.group(1).strip()
-        cam_m = _MD_FIELD_RE["camera"].search(block)
-        if cam_m:
-            camera = cam_m.group(1).strip()
-        shot_type = camera or visual or "中景平视，固定镜头"
-
-        # 口播/对白
-        voice = ""
-        vo_m = _MD_FIELD_RE["voiceover"].search(block)
-        if vo_m:
-            voice = (vo_m.group(1) or vo_m.group(2) or vo_m.group(3) or "").strip()
-            # 去除末尾可能的标点
-            voice = voice.rstrip("，。、；;")
-
-        # 动作
-        action = ""
-        act_m = _MD_FIELD_RE["action"].search(block)
-        if act_m:
-            action = act_m.group(1).strip()
-
-        # 音效/BGM
-        audio = ""
-        audio_m = _MD_FIELD_RE["audio_bgm"].search(block)
-        if audio_m:
-            audio = audio_m.group(1).strip()
-
-        # 转场
-        trans = ""
-        trans_m = _MD_FIELD_RE["transition"].search(block)
-        if trans_m:
-            trans = trans_m.group(1).strip()
-
-        # 参考图
-        ref_idx = None
-        ref_m = _MD_FIELD_RE["reference"].search(block)
-        if ref_m:
-            ref_str = ref_m.group(1) or ref_m.group(2)
-            try:
-                ref_idx = int(ref_str) - 1  # 用户视角从1开始，内部从0开始
-                if ref_idx < 0:
-                    ref_idx = None
-            except (ValueError, TypeError):
-                ref_idx = None
-
-        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 ''}"
-
-        shot = {
-            "time_range": time_range,
-            "shot_type_angle_movement": shot_type,
-            "scene_and_dialogue": scene_and_dialogue,
-            "action_details": action,
-            "audio_bgm": audio or "轻快BGM",
-            "transition": trans or ("硬切" if i < len(shot_blocks) - 1 else "结束"),
-            "reference_image_index": ref_idx,
-            "voiceover": voice,
-        }
-        shots.append(shot)
-
-    if not shots:
-        return None
-    base["shots"] = shots
-
-    # 顶层 voiceover_script：先尝试从"口播全文"段提取，否则拼接各镜头口播
-    full_vo_match = _MD_FIELD_RE["full_voiceover"].search(raw)
-    if full_vo_match:
-        voiceover = full_vo_match.group(1).strip()
-    else:
-        voiceover = "".join(voice_parts)
-    if not voiceover.strip():
-        return None
-
-    base["voiceover_script"] = voiceover
-    base["final_copy"] = voiceover
-    base["suggested_copy"] = voiceover
-    base["copy_display_markdown"] = voiceover
-    return base
-
-
 def _step_script_generation(
     job: ViralVideoJob,
     image_analysis: dict,
     on_delta: Optional[Callable[[str, str], None]] = None,
 ) -> dict:
-    """步骤: 意图理解 + 分镜生成一次完成（单一重试 for 循环 + 双格式解析）。
+    """步骤: 意图理解 + 分镜生成一次完成（PR2 直线化版：单一重试 for 循环 + XML-only 解析）。
 
     重试链路：fast@0.8 → fast@0.6 → pro@0.7 → _finalize_fallback_script。
-    - _parse_script 支持纯 XML / 双段式（中文+XML）/ 纯中文 markdown 三种格式；
     - 删除 JSON fallback 解析路径（PR1 已统一 XML）；
     - 删除 compress-retry 冗余分支（前3次失败再压缩意义不大）；
     - _try_gen / _try_gen_stream / _stream_chat_with_fallback 合并到 _call_llm_once + _validate_copy_result。
     """
     try:
         from packages.application.viral_video.prompt_loader import (
-            build_messages,
             get_template,
+            render_system_prompt,
+            render_user_prompt,
         )
         from packages.shared.ai_router import ai_router
     except ImportError:
@@ -1436,27 +1158,7 @@ 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
+    system_prompt = render_system_prompt(template)
     marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
     theme_hint = _determine_theme(image_analysis, marketing_purpose)
 
@@ -1500,7 +1202,7 @@ def _step_script_generation(
     # 字数/镜头数约束（统一由 constants 生成，注入到 prompt 占位符 {duration_constraint_text}）
     _duration_constraint_text = build_duration_constraint_text(dur)
 
-    llm_messages = build_messages(
+    user_prompt = render_user_prompt(
         template,
         marketing_purpose=marketing_purpose or "未指定",
         marketing_purpose_hint=mp_hint,
@@ -1527,13 +1229,9 @@ 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：豆包 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"))
+    fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "90"))
+    pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "60"))
+    deadline = time.time() + 180
 
     # ── 重试 for 循环（替代原嵌套 if + 两个重复 _try_gen） ─────────────────
     attempts: list[tuple] = [
@@ -1555,34 +1253,18 @@ def _step_script_generation(
             logger.info("[爆款视频] 分镜生成 model=%s label=%s timeout=%d", client.model, label, tmo)
             raw = _call_llm_once(
                 client,
-                messages=llm_messages,
+                system_prompt,
+                user_prompt,
                 temperature=temp,
                 max_tokens=max_tok,
                 timeout=tmo,
                 on_delta=on_delta,
-                enable_thinking=False,  # 分镜模板填充式生成不需要长推理，关思考 TTFT<3s 实测
             )
             if not raw:
-                logger.warning("[爆款视频] 分镜生成返回空 label=%s", label)
                 continue
-            # 记录 raw 头部/尾部/finish_reason，便于排查格式问题（截断 / 非XML / 跑题等）
-            logger.info(
-                "[爆款视频] LLM raw label=%s len=%d head=%r tail=%r finish=%s",
-                label,
-                len(raw),
-                raw[:120].replace("\n", "\\n"),
-                raw[-120:].replace("\n", "\\n"),
-                getattr(client, "last_finish_reason", ""),
-            )
-            copy_result = _parse_script(raw, job)
+            copy_result = _script_from_xml(raw, job)
             if copy_result is None:
-                logger.warning(
-                    "[爆款视频] 分镜脚本解析失败 label=%s raw_len=%d head=%r tail=%r",
-                    label,
-                    len(raw),
-                    raw[:80].replace("\n", "\\n"),
-                    raw[-80:].replace("\n", "\\n"),
-                )
+                logger.warning("[viral][script] attempt=%s XML解析失败 raw_len=%d", label, len(raw))
                 continue
             reason = _validate_copy_result(copy_result, dur)
             voiceover_len = len((copy_result.get("voiceover_script") or "").strip())
@@ -1595,27 +1277,6 @@ def _step_script_generation(
                     shots_cnt,
                 )
                 return copy_result
-            # P0: 口播超长是最常见的失败原因（v4 编导模板字数天然偏多）。LLM 重试通常仍会超长，
-            # 直接按句号切句截断，避免浪费 fast/pro 配额和 deadline。
-            if reason.startswith("口播超长") and shots_cnt >= 1:
-                spec = get_duration_spec(dur)
-                _orig_vo = (copy_result.get("voiceover_script") or "").strip()
-                _truncated = _truncate_voiceover(_orig_vo, spec["max_chars"])
-                if len(_truncated) >= spec["min_chars"] and len(_truncated) <= spec["max_chars"]:
-                    # 同步更新 shots 里每个镜头的口播（按原比例分配截断）
-                    copy_result["voiceover_script"] = _truncated
-                    # 简单策略：把完整口播放在第一个镜头的 scene_and_dialogue 前面，
-                    # 其余镜头保留画面描述，不重复塞口播（TTS 从 voiceover_script 取，不走 scene）
-                    _recheck = _validate_copy_result(copy_result, dur)
-                    if _recheck is None or _recheck.startswith("口播超长") is False:
-                        logger.info(
-                            "[爆款视频] 口播超长自动截断 label=%s orig=%d trunc=%d shots=%d ok",
-                            label,
-                            voiceover_len,
-                            len(_truncated),
-                            shots_cnt,
-                        )
-                        return copy_result
             logger.warning(
                 "[viral][script] attempt=%s validate_fail=%s voiceover=%d shots=%d",
                 label,

packages/adapters/sqlalchemy_impl/models.py

@@ -948,7 +948,6 @@ class ViralVideoJobModel(Base):
     persona_id = Column(String(36), nullable=False, default="")
     viral_structure = Column(String(50), nullable=False, default="")
     marketing_purpose = Column(String(100), nullable=False, default="")
-    tone = Column(String(50), nullable=False, default="")
     bgm_preference = Column(String(50), nullable=False, default="")
     duration = Column(Integer, nullable=False, default=30)
     user_copy_text = Column(Text, nullable=False, default="")

packages/adapters/sqlalchemy_impl/viral_video_repository.py

@@ -38,7 +38,6 @@ def _to_domain(model: ViralVideoJobModel) -> ViralVideoJob:
         persona_id=model.persona_id or "",
         viral_structure=model.viral_structure or "",
         marketing_purpose=model.marketing_purpose or "",
-        tone=getattr(model, "tone", "") or "",
         bgm_preference=model.bgm_preference or "",
         duration=model.duration or 15,
         user_copy_text=model.user_copy_text or "",
@@ -92,7 +91,6 @@ class SQLAlchemyViralVideoJobRepository:
             persona_id=job.persona_id,
             viral_structure=job.viral_structure,
             marketing_purpose=job.marketing_purpose,
-            tone=getattr(job, "tone", "") or "",
             bgm_preference=job.bgm_preference,
             duration=job.duration,
             user_copy_text=job.user_copy_text,
@@ -161,7 +159,6 @@ class SQLAlchemyViralVideoJobRepository:
         model.persona_id = job.persona_id
         model.viral_structure = job.viral_structure
         model.marketing_purpose = job.marketing_purpose
-        model.tone = getattr(job, "tone", "") or ""
         model.bgm_preference = job.bgm_preference
         model.duration = job.duration
         model.fusion_level = job.fusion_level

packages/application/ditto_emotion_service.py

@@ -4,8 +4,7 @@
 1. 分句：先按 。！？； 句末标点切分；整段无句末标点且超长时按逗号等语义停顿二次切分
 2. 调 DoubaoClient.chat_completion 分析每句表情（emo: 0-7 八种全部开放, intensity: 0-1）
 3. 结果 LRU 缓存（文案 hash → 情绪列表）
-4. LLM 失败/超时/格式错 → analyze 层预分句全中性兜底；build_timeline 识别全兜底特征后
-   返回空串不传 emo_timeline（避免压制 GPU 端自身表情逻辑），不阻塞生成
+4. LLM 失败/超时/格式错 → 返回空列表（降级中性表情，不阻塞生成）
 5. TTS 完成后按字数比例或 sentence_timings 对齐成秒级 timeline
 """
 
@@ -32,11 +31,6 @@ EMO_SURPRISE = 6
 EMO_CONTEMPT = 7
 ALLOWED_EMOS = set(range(8))  # 8 种表情全部开放（0-7）；超出范围的非法值仍会被降级为中性
 
-# 传给 Ditto 的 emo_timeline 强度下限：GPU 端表情混合公式修复后，intensity>=0.3
-# 目标情绪才能压过中性；LLM 常给陈述句打 0.1-0.2，信号太弱、表情效果一眼假。
-# 构造 timeline entry 时统一保底到该值（不影响 analyze 层 LLM 原始输出与日志排查）
-MIN_TIMELINE_INTENSITY = 0.35
-
 # ── 分句正则 ─────────────────────────────────────────────────────
 _SENT_SPLIT_RE = re.compile(r"(?<=[。！？；!?;])\s*")
 # 二次切分：逗号/分号/顿号/冒号等语义停顿（lookbehind 保留标点在前句末尾）
@@ -88,8 +82,7 @@ class EmotionTimelineEntry:
         self.start = round(start, 2)
         self.end = round(end, 2)
         self.emo = emo
-        # 强度保底：低于 MIN_TIMELINE_INTENSITY 的信号在 GPU 混合公式中压不过中性
-        self.intensity = round(max(intensity, MIN_TIMELINE_INTENSITY), 2)
+        self.intensity = round(intensity, 2)
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -368,18 +361,6 @@ class DittoEmotionService:
         segments = self.analyze(text)
         if not segments:
             return ""
-        # LLM 失败/异常时 _cached_analyze 会兜底返回"全中性(emo=4)+低强度(<=0.1)"的
-        # 预分句结果；把这种"明确的全中性指令"传给 GPU 会压制 Ditto 端
-        # use_script_emo/emo_global 的自身处理，比不传更糟糕。识别全兜底特征后
-        # 返回空串（payload 不带 emo_timeline），交由 GPU 端自行决定表情。
-        # 注意：LLM 真实分析即使全中性，intensity 一般 >0.1，不会被误伤。
-        if all(s.emo == EMO_NEUTRAL and s.intensity <= 0.1 for s in segments):
-            logger.info(
-                "[ditto_emotion] 分析结果全为中性低强度（疑似 LLM 兜底），"
-                "emo_timeline 置空交由 GPU 端处理: text=%.80s",
-                text,
-            )
-            return ""
         if sentence_timings:
             entries = align_timeline_by_timings(segments, sentence_timings, audio_duration)
         else:

packages/application/viral_video/constants.py

@@ -7,11 +7,11 @@ from __future__ import annotations
 
 # 时长规格：key=duration(秒)，value=对应约束
 DURATION_SPEC: dict[int, dict[str, int]] = {
-    5: {"min_chars": 12, "max_chars": 25, "min_shots": 1, "max_shots": 2},
-    10: {"min_chars": 25, "max_chars": 50, "min_shots": 2, "max_shots": 3},
-    15: {"min_chars": 38, "max_chars": 75, "min_shots": 3, "max_shots": 4},
-    20: {"min_chars": 50, "max_chars": 100, "min_shots": 4, "max_shots": 5},
-    30: {"min_chars": 75, "max_chars": 150, "min_shots": 6, "max_shots": 8},
+    5: {"min_chars": 12, "max_chars": 20, "min_shots": 1, "max_shots": 2},
+    10: {"min_chars": 25, "max_chars": 40, "min_shots": 2, "max_shots": 3},
+    15: {"min_chars": 38, "max_chars": 60, "min_shots": 3, "max_shots": 4},
+    20: {"min_chars": 50, "max_chars": 80, "min_shots": 4, "max_shots": 5},
+    30: {"min_chars": 75, "max_chars": 120, "min_shots": 6, "max_shots": 8},
 }
 
 
@@ -38,8 +38,8 @@ def build_duration_constraint_text(dur: int) -> str:
     spec = get_duration_spec(dur)
     return (
         f"视频时长{dur}秒。口播总字数严格控制在{spec['min_chars']}~{spec['max_chars']}字之间"
-        f"（中文口播语速约3~5字/秒，{dur}秒={spec['max_chars']}字为硬上限，必须严格遵守，"
-        f"超长会被截断导致语义不完整）。"
+        f"（中文正常口播语速约2.5~4字/秒，{dur}秒={dur * 4}字为硬上限，宁可少写也不要多写，"
+        f"超长会导致TTS配音超出视频时长而失败）。"
         f"分镜镜头数{spec['min_shots']}~{spec['max_shots']}个，每个镜头的time_range首尾相接，"
         f"不得跳秒或重叠，每个镜头时长≥2秒。"
     )

packages/application/viral_video/prompt_loader.py

@@ -30,20 +30,14 @@ _warned: dict[str, float] = {}
 _warn_lock = threading.Lock()
 
 
-def _warn_once(key: str, context: str, detail: BaseException | str | None = None) -> None:
-    """冷却告警通用方法：5 分钟内同 key 只打一次，避免日志刷屏。
-
-    - key: 告警分组键（去重用）
-    - context: 本条告警的简短场景描述（直接拼进日志，不要再写死 'fallback to default' 字样）
-    - detail: 具体异常或描述
-    """
+def _warn_once(key: str, template_type: str, e: BaseException | str | None = None) -> None:
     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 %s: %s", context, detail)
+    logger.warning("viral_video prompt template %s fallback to default: %s", template_type, e)
 
 
 @dataclass
@@ -91,10 +85,10 @@ def _get_session():
 
         url = str(get_shared_settings().database_url)
     except Exception as e:  # noqa: BLE001
-        _warn_once("session_config", "db config failed (will fallback)", e)
+        _warn_once("session_config", "db_config", e)
         return None
     if not url:
-        _warn_once("session_url", "db_url empty (will fallback)", "database_url empty")
+        _warn_once("session_url", "db_url", "database_url empty")
         return None
     url = url.replace("postgresql+asyncpg://", "postgresql+psycopg://")
     if url.startswith("postgresql://") and "psycopg" not in url:
@@ -134,14 +128,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}", f"db load failed for {prompt_type} (will fallback to code default)", e)
+        _warn_once(f"load_db:{prompt_type}", prompt_type, 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}", f"session close failed for {prompt_type}", e)
+                _warn_once(f"session_close:{prompt_type}", prompt_type, e)
 
 
 def get_template(prompt_type: str, *, force_refresh: bool = False) -> Optional[PromptTemplate]:
@@ -157,11 +151,7 @@ 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}",
-            f"NO TEMPLATE for {prompt_type} (neither db nor code defaults)",
-            "no template in db or code defaults",
-        )
+        _warn_once(f"no_template:{prompt_type}", prompt_type, "no template in db or code defaults")
     if template is not None:
         with _lock:
             _cache[prompt_type] = (now, template)
@@ -176,114 +166,24 @@ def invalidate() -> None:
         _warned.clear()
 
 
+class _SafeDict(dict):
+    def __missing__(self, key: str) -> str:
+        return "{" + key + "}"
+
+
 def _safe_format(text: str, kwargs: dict) -> str:
-    """安全格式化模板：任何占位符异常（缺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项参数}非法属性访问）→ ERROR告警
-        # 2. 样例文本花括号（如{大远景/远景/全景}给LLM看的景别选项）→ 原样保留，属正常
-        # 注意：本函数只原样保留，不会让template回落到DEFAULT_TEMPLATES；
-        # 真正的回落到代码默认模板只在DB查询失败/无激活模板时发生（见_load_from_db/get_template），
-        # 那里会打独立ERROR日志并带job_id（见viral_video.py中DEFAULT_TEMPLATES检测块）。
-        logger.error(
-            "[viral][prompt] 模板格式化有%d个占位符无法渲染（示例：%s），"
-            "已原样保留、未回落到默认模板——若这些花括号是样例文本（如景别选项/JSON示例）可忽略；"
-            "若应为变量请检查system/user_prompt_template占位符是否写对",
-            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:
-    """system_prompt 含运行时变量时填充；example_output 存在时作为 few-shot 样例追加到末尾。"""
-    body = _safe_format(template.system_prompt, kwargs)
-    example = (template.example_output or "").strip()
-    if not example:
-        return body
-    # 追加一个 assistant 角色样例消息由调用方拼成多轮；system 里只能带规则，example 统一由 user 侧追加
-    return body
+    try:
+        return text.format_map(_SafeDict(kwargs))
+    except Exception as e:  # noqa: BLE001
+        _warn_once("safe_format", "format", e)
+        return text
 
 
 def render_user_prompt(template: PromptTemplate, **kwargs) -> str:
-    r"""填充 user_prompt_template 占位符；example_output 存在时作为 few-shot 样例追加到 user 消息末尾，
-    配合 \_build_example_exchange 拼多轮 messages，LLM 收到完整「示例输入→示例输出」对齐格式。"""
-    body = _safe_format(template.user_prompt_template, kwargs)
-    example = (template.example_output or "").strip()
-    if not example:
-        return body
-    return body
+    """填充 user_prompt_template 占位符，缺键原样保留不报错。"""
+    return _safe_format(template.user_prompt_template, kwargs)
 
 
-def build_messages(template: PromptTemplate, **kwargs) -> list[dict]:
-    """构造传给 LLM 的 messages 列表：system + (user, assistant[example]) + user。
-
-    - 第 1 条 system：规则
-    - 如果 template.example_output 非空，追加一轮 few-shot：user 给示例 user_prompt 渲染结果，assistant 给示例输出
-    - 最后 user：当前任务真实 user_prompt
-    这样 LLM 既能看到规则，又能看到一个完整「输入→输出」样例，格式遵从度大幅提升。
-    """
-    sys_body = render_system_prompt(template, **kwargs)
-    user_body = render_user_prompt(template, **kwargs)
-    example = (template.example_output or "").strip()
-    messages: list[dict] = [{"role": "system", "content": sys_body}]
-    if example:
-        # 用一个通用「请参考以下示例输出格式」作为 few-shot user 前缀
-        messages.append(
-            {
-                "role": "user",
-                "content": "以下是一个符合规范的完整示例输出，请严格参照其格式、字段、风格来生成：",
-            }
-        )
-        messages.append({"role": "assistant", "content": example})
-    messages.append({"role": "user", "content": user_body})
-    return messages
+def render_system_prompt(template: PromptTemplate, **kwargs) -> str:
+    """copy_fusion 等 system_prompt 含运行时变量时填充。"""
+    return _safe_format(template.system_prompt, kwargs)

packages/application/viral_video/reviewer.py

@@ -17,8 +17,9 @@ from typing import Optional
 
 from packages.application.viral_video import xml_parser as xp
 from packages.application.viral_video.prompt_loader import (
-    build_messages,
     get_template,
+    render_system_prompt,
+    render_user_prompt,
 )
 from packages.application.viral_video.prompts import COMPLIANCE_BLACKLIST
 from packages.application.viral_video.schemas import ReviewIssue, ReviewResult
@@ -118,9 +119,13 @@ class Reviewer:
         if not self.client or not getattr(self.client, "is_available", True):
             return None
         template = get_template("review")
-        messages = build_messages(template, voiceover_script=voiceover_script)
+        system = render_system_prompt(template)
+        user = render_user_prompt(template, voiceover_script=voiceover_script)
         raw = self.client.chat_completion(
-            messages,
+            [
+                {"role": "system", "content": system},
+                {"role": "user", "content": user},
+            ],
             temperature=0.2,
             max_tokens=1024,
             timeout=25,

packages/config/base.py

@@ -202,9 +202,9 @@ class SharedSettings(BaseSettings):
         default=120,
         validation_alias=AliasChoices("DITTO_REQUEST_TIMEOUT", "ditto_request_timeout"),
     )
-    # Ditto 句间过渡帧数（平滑表情/口型切换）；默认 6 与 GPU 端 clamp 最优值对齐（表情切换更灵敏）
+    # Ditto 句间过渡帧数（平滑表情/口型切换）
     ditto_blend_frames: int = Field(
-        default=6,
+        default=12,
         validation_alias=AliasChoices("DITTO_BLEND_FRAMES", "ditto_blend_frames"),
     )
 

packages/domain/viral_video/entities.py

@@ -92,7 +92,6 @@ class ViralVideoJob:
     persona_id: str = ""
     viral_structure: str = ""
     marketing_purpose: str = ""
-    tone: str = ""
     bgm_preference: str = ""
     duration: int = 15  # v1.6: 默认15秒，上限30秒（Seedance 2.5 单次最大30s）
     user_copy_text: str = ""

packages/shared/ai_client.py

@@ -439,8 +439,6 @@ 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")

reports/viral-video-audit-20261014.md

@@ -1,116 +0,0 @@
-# 爆款视频模块全面审计报告
-
-**审计日期**: 2026-10-14
-**审计分支**: fix/viral-script-dual-format-parser
-**审计人**: CodeAct Agent
-**基准 commit**: 5e5a9b6 (#2294)
-
----
-
-## 一、Parser 改造说明
-
-### 改造内容
-将 `_script_from_xml`（仅支持纯 XML）改造为 `_parse_script`，支持三种 LLM 输出格式：
-
-| 格式 | 描述 | 解析路径 |
-|------|------|----------|
-| A: 纯 XML | `<script>...</script>` 包裹（v3/v4） | `_parse_script_xml` |
-| B: 双段式 | 中文 markdown 说明 + 末尾 `<script>` XML 块 | `_parse_script_xml`（提取最后一个 script 块） |
-| C: 纯中文 markdown | 无 `<script>` 标签，按"镜头1/镜头2..."分段 | `_parse_script_markdown` |
-| (兼容) 无包裹 v3 XML | 含 `<clips>/<voiceover_script>` 但无 `<script>` 包裹 | `_parse_script_xml`（整个 raw 当 XML body） |
-
-### 解析策略
-1. 正则 `r'<script[^>]*>(.*?)</script>'` (DOTALL) 提取所有 `<script>` 块，取最后一个
-2. 有 XML 块 → 走 XML 解析（`_parse_script_xml`）
-3. 无 `<script>` 但含 v3 结构标签（`<clips>/<voiceover_script>/<copy_display_markdown>`）→ 整个 raw 走 XML 解析（兼容旧格式）
-4. 都不匹配 → 走中文 markdown 兜底解析（`_parse_script_markdown`）
-5. 都解析不出 → 返回 None
-
-### 代码变更
-- `_script_from_xml` → `_parse_script`（入口函数）
-- 新增 `_parse_script_xml`（XML 路径解析，从原函数抽出）
-- 新增 `_parse_script_markdown`（中文 markdown 兜底解析）
-- 模块级常量：`_MD_SHOT_SPLIT_RE`, `_MD_TIME_RE`, `_MD_FIELD_RE` 等正则
-
----
-
-## 二、Seedance voiceover bug 修复说明
-
-### 问题
-`_assemble_seedance_prompt`（~line 1990/2010）循环 shots 时读 `s.get("voiceover", "")`，但原 `_script_from_xml` 输出的 shot dict 没有 `voiceover` 字段——口播被拼进了 `scene_and_dialogue = f"{visual}｜{voice}"`，导致 Seedance/Wan prompt 中口播为空。
-
-### 修复
-1. `_parse_script_xml`: 每个 shot dict 显式设置 `"voiceover": voice or ""`
-2. `_parse_script_markdown`: 同样输出 `"voiceover": voice` 字段
-3. `_fallback_script`: shot dict 加 `"voiceover": voiceover`（就是那一句 voiceover）
-4. `_assemble_seedance_prompt` 的 Wan 和 Seedance 两个分支都已正确读 `s.get("voiceover")`，无需改动
-
-### 验证
-- 新增测试 `TestShotVoiceoverField`（3 个用例）确认 XML/Markdown/Fallback 三条路径的 shot 都有 `voiceover`
-- 新增测试 `test_assemble_seedance_uses_shot_voiceover` 确认 Seedance prompt 能取到口播
-
----
-
-## 三、冗余代码清理情况
-
-| 项目 | 状态 | 说明 |
-|------|------|------|
-| `_script_from_xml` → `_parse_script` 重命名 | ✅ 完成 | 调用点（line ~1484）已同步更新 |
-| `_safe_json_loads` | ⚠️ 保留 | 已不在主流程使用（JSON fallback 路径已删），但 2 个单测引用它，暂保留 |
-| `_validate_copy_result` docstring | ✅ 更新 | 引用从 `_script_from_xml` 改为 `_parse_script` |
-| `_step_script_generation` docstring | ✅ 更新 | 反映双格式解析支持 |
-| reviewer.py voiceover_script XML 标签 | ✅ 确认无需改 | `<voiceover_script>` 仅存在于 reviewer 自己构造的 messages 里，跟 storyboard 输出格式无关 |
-| JSON fallback 残留 | ✅ 无残留 | 主流程无 JSON 解析路径，`_safe_json_loads` 仅测试引用 |
-
----
-
-## 四、全面审计结果
-
-### 审计清单 9 项
-
-| # | 审计项 | 验证结果 | 发现问题 | 修复情况 |
-|---|--------|----------|----------|----------|
-| 1 | **用户配置参数链路** | ✅ 基本完整 | `tone` 字段未在 API schema (GenerateCopyRequest) 中定义，但 worker 用 `getattr(job, "tone", "") or "亲切自然"` 优雅降级，不影响功能 | 记录为 minor issue，不影响本次 PR |
-| 2 | **图片分析→文案→review→渲染全流程** | ✅ 链路完整 | `_step_image_analysis` → `_step_script_generation` → `_step_review` → `_step_tts` → `_step_render` → `_step_upload`，每步有错误处理和降级 | 无需修复 |
-| 3 | **三视图机制** | ✅ 链路完整 | `pre_trusted_images` → `_build_character_section` → Seedream 三视图 → reference_images 通道。代码注释明确标注"当前为空实现，仅保留扩展点" | 无需修复 |
-| 4 | **few-shot example_output** | ✅ 正确使用 | `prompt_loader.build_messages` 在 storyboard 和 reviewer 两个场景都正确注入 few-shot（system→user示例→assistant示例→user任务） | 无需修复 |
-| 5 | **字数/时长约束** | ✅ 链路正确 | `constants.DURATION_SPEC` → `build_duration_constraint_text` → `{duration_constraint_text}` 占位符注入 → `_validate_copy_result` 校验 → `_truncate_voiceover` 截断 | 无需修复 |
-| 6 | **重试/超时机制** | ✅ 链路正确 | deadline 300s + fast@0.8 → fast@0.6 → pro@0.7 三档重试 + 流式外层 deadline + next() 超时 + 口播超长自动截断（`_truncate_voiceover`）| 无需修复 |
-| 7 | **fallback 脚本** | ✅ brand==name 去重生效 | `_fallback_script` 中 `_brand_str == _name_str or _brand_str in _name_str or _name_str in _brand_str` 三重判断确保去重（#2294 修复） | 本次新增 `voiceover` 字段到 shot dict |
-| 8 | **TTS/口型驱动链路** | ✅ 链路完整 | `voiceover_script` → `_step_tts` (CosyVoice) → 音频 URL → `_step_render` (Seedance) → 视频 → Ditto/MuseTalk 口型。voiceover 字段修复后链路更完整 | voiceover bug 修复（本次 PR） |
-| 9 | **review 审核阶段** | ✅ 逻辑正确 | `reviewer.py` 的 `_llm_review_inner` 用 `build_messages(template, voiceover_script=...)` 送审，XML 解析 `<passed>/<issues>`，审核失败合并本地规则后返回；LLM 不可用时降级放行（带 warning issue）| 无需修复 |
-
-### 发现的 Minor Issues（不阻塞本次 PR）
-
-1. **`tone` 参数未暴露到 API**: `GenerateCopyRequest` schema 没有 `tone` 字段，但 worker 用 `getattr` 优雅降级到默认值 "亲切自然"。前端如需自定义语气，需后续版本补 schema 字段。
-2. **`_safe_json_loads` 死代码**: 主流程已不使用（JSON fallback 在 PR1 已删），但 2 个单测 (`test_safe_json_loads_*`) 引用它。建议下个 PR 清理。
-3. **API ShotScript schema 缺 `voiceover` 字段**: `apps/api/app/schemas/viral_video.py` 的 `ShotScript` 模型没有 `voiceover` 字段定义。前端目前从 `scene_and_dialogue` 拆口播，后续可考虑暴露 `voiceover` 字段给前端。
-
----
-
-## 五、测试覆盖情况
-
-### 新增测试文件: `tests/unit/test_viral_video_dual_parser.py`
-
-| 测试类 | 用例数 | 覆盖范围 |
-|--------|--------|----------|
-| `TestDualFormatParsing` | 2 | 双段式（中文+XML）解析 + voiceover 字段 |
-| `TestPureXmlParsing` | 2 | 纯 XML 兼容性 + voiceover 字段 |
-| `TestPureMarkdownParsing` | 9 | 纯中文 markdown 解析（shots/theme/voiceover/time_range/audio/transition/reference/拼接/中文数字） |
-| `TestShotVoiceoverField` | 3 | XML/Fallback/Seedance 三条路径的 voiceover 字段 |
-| `TestParameterPassing` | 3 | marketing_purpose/target_audience/fusion_level 参数传递 |
-| `TestMarkdownParseFailures` | 4 | 解析失败返回 None（纯文本/空字符串/None/无口播） |
-
-**合计**: 23 个新增用例，全部通过。
-
-### 回归测试
-- 全量 viral 相关测试: **269 passed, 2 skipped**
-- 覆盖了 test_viral_video.py / test_viral_video_p0.py / test_viral_video_p2_errors.py / test_viral_video_prompt_system.py / test_viral_video_routes.py / test_viral_video_v8_bugs.py / test_viral_video_wiring.py / test_viral_video_ws.py / test_viral_video_models.py
-
----
-
-## 六、PR 信息
-
-- **分支**: `fix/viral-script-dual-format-parser`
-- **Base**: `develop`
-- **PR**: 待创建

tests/unit/test_ditto_emotion_service.py

@@ -8,11 +8,8 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from packages.application.ditto_emotion_service import (
-    EMO_ANGER,
     EMO_HAPPY,
     EMO_NEUTRAL,
-    EMO_SAD,
-    EMO_SURPRISE,
     DittoEmotionService,
     EmotionSegment,
     _parse_emotion_json,
@@ -365,112 +362,3 @@ class TestWarnIfSuspiciousAllNeutral:
         with caplog.at_level("WARNING", logger="packages.application.ditto_emotion_service"):
             _warn_if_suspicious_all_neutral("家人们！", [])
         assert not caplog.records
-
-
-# ── build_timeline 全中性兜底置空（GPU端表情公式配套修复1）─────────
-class TestBuildTimelineAllNeutralFallback:
-    """LLM 兜底特征（全 emo=4 且 intensity<=0.1）→ 不传 emo_timeline，交 GPU 端自行处理."""
-
-    def test_llm_failure_fallback_returns_empty(self):
-        """LLM 调用失败 → _cached_analyze 兜底全中性0.1 → build_timeline 置空."""
-        import packages.application.ditto_emotion_service as m
-
-        svc = _make_service(enabled=True)
-        m._cached_analyze.cache_clear()
-        try:
-            with patch.object(svc, "_call_llm", return_value=[]):
-                result = svc.build_timeline("全兜底置空检测专用独一无二文案甲。全兜底置空检测专用文案乙。", 5.0)
-                assert result == ""
-        finally:
-            m._cached_analyze.cache_clear()
-
-    def test_all_neutral_low_intensity_returns_empty(self):
-        """LLM 返回全 emo=4 且 intensity<=0.1（兜底特征）→ 置空."""
-        import packages.application.ditto_emotion_service as m
-
-        svc = _make_service(enabled=True)
-        m._cached_analyze.cache_clear()
-        fake_reply = json.dumps(
-            [
-                {"text": "低强度中性检测专用文案甲", "emo": 4, "intensity": 0.1},
-                {"text": "低强度中性检测专用文案乙", "emo": 4, "intensity": 0.05},
-            ]
-        )
-        try:
-            with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)):
-                assert svc.build_timeline("低强度中性检测专用文案甲。低强度中性检测专用文案乙。", 5.0) == ""
-        finally:
-            m._cached_analyze.cache_clear()
-
-    def test_all_neutral_but_higher_intensity_kept(self):
-        """全中性但 intensity>0.1（LLM 真实输出特征）→ 正常生成 timeline，不误伤."""
-        import packages.application.ditto_emotion_service as m
-
-        svc = _make_service(enabled=True)
-        m._cached_analyze.cache_clear()
-        fake_reply = json.dumps(
-            [
-                {"text": "高强度中性保留检测专用文案甲", "emo": 4, "intensity": 0.3},
-                {"text": "高强度中性保留检测专用文案乙", "emo": 4, "intensity": 0.2},
-            ]
-        )
-        try:
-            with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)):
-                result = svc.build_timeline("高强度中性保留检测专用文案甲。高强度中性保留检测专用文案乙。", 4.0)
-                data = json.loads(result)
-                assert len(data) == 2
-                assert all(item["emo"] == 4 for item in data)
-        finally:
-            m._cached_analyze.cache_clear()
-
-
-# ── timeline entry 强度保底 0.35（GPU端表情公式配套修复3）─────────
-class TestTimelineIntensityFloor:
-    def test_low_intensity_floored(self):
-        segs = [EmotionSegment("ab", EMO_HAPPY, 0.2)]
-        entries = align_timeline_by_length(segs, 2.0)
-        assert entries[0].intensity == 0.35
-
-    def test_floor_boundary_values(self):
-        segs = [
-            EmotionSegment("a", EMO_HAPPY, 0.1),
-            EmotionSegment("b", EMO_SAD, 0.34),
-            EmotionSegment("c", EMO_ANGER, 0.35),
-        ]
-        entries = align_timeline_by_length(segs, 3.0)
-        assert entries[0].intensity == 0.35
-        assert entries[1].intensity == 0.35
-        assert entries[2].intensity == 0.35
-
-    def test_high_intensity_unchanged(self):
-        segs = [EmotionSegment("ab", EMO_HAPPY, 0.6)]
-        entries = align_timeline_by_length(segs, 2.0)
-        assert entries[0].intensity == 0.6
-
-    def test_floor_applies_in_timings_path(self):
-        segs = [EmotionSegment("hello", EMO_SAD, 0.15)]
-        timings = [{"start": 0.0, "end": 2.0}]
-        entries = align_timeline_by_timings(segs, timings, 2.0)
-        assert entries[0].intensity == 0.35
-
-
-# ── timings 数量不匹配回退字数占比（GPU端表情公式配套修复2，固化行为）───
-class TestAlignTimingsMismatchCharRatio:
-    def test_mismatch_falls_back_to_char_ratio_with_emo_mapping(self):
-        """LLM 情绪分句（按逗号）与 TTS 分句（句末标点）数量不一致时，
-        必须回退到按情绪分句字数占比分配，保证 emo 与时间段一一对应."""
-        segs = [
-            EmotionSegment("a" * 6, EMO_HAPPY, 0.5),
-            EmotionSegment("b" * 2, EMO_SAD, 0.5),
-            EmotionSegment("c" * 2, EMO_SURPRISE, 0.5),
-        ]
-        timings = [{"start": 0.0, "end": 10.0}]  # len 1 != 3
-        entries = align_timeline_by_timings(segs, timings, 10.0)
-        assert len(entries) == 3
-        assert entries[0].start == 0.0
-        assert entries[0].end == 6.0
-        assert entries[1].start == 6.0
-        assert entries[1].end == 8.0
-        assert entries[2].start == 8.0
-        assert entries[2].end == 10.0
-        assert [e.emo for e in entries] == [EMO_HAPPY, EMO_SAD, EMO_SURPRISE]

tests/unit/test_viral_video_dual_parser.py

@@ -1,562 +0,0 @@
-"""双格式 parser + voiceover 字段 + markdown 兜底解析测试。
-
-覆盖：
-1. 双段式（中文 markdown + <script> XML 块）解析
-2. 纯 XML 解析（已有测试覆盖，确认兼容）
-3. 纯中文 markdown 解析（镜头1/镜头2/关键字段）
-4. Shot voiceover 字段存在性 + _assemble_seedance_prompt 取值
-5. 参数传递集成测试（marketing_purpose/target_audience/fusion_level）
-6. markdown 解析失败返回 None
-"""
-
-from __future__ import annotations
-
-import sys
-from unittest.mock import MagicMock
-
-import pytest
-
-# ── 通用 fixtures ──────────────────────────────────────────────────────
-
-
-@pytest.fixture
-def mock_job():
-    """构造一个最小可用的 ViralVideoJob mock。"""
-    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):
-    """导入 worker viral_video 模块（mock celery/db 依赖）。"""
-    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. 双段式解析：中文 markdown + <script> XML 块
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestDualFormatParsing:
-    """格式B：中文可读说明 + 末尾 <script>...</script> XML 块。"""
-
-    DUAL_FORMAT_INPUT = """## 视频总览
-
-这是一个关于口红试色的短视频，目标受众是年轻女性。
-
-### 逐秒镜头拆解
-
-- 整体风格：暖色调、近景特写
-- 节奏：前3秒给出钩子，中段展示产品，结尾引导行动
-
-### 硬性约束
-- 无字幕无水印
-- 人物一致性
-
----
-
-以下是符合规范的分镜脚本：
-
-<script>
-  <copy_display_markdown><![CDATA[今天试一支超显白的口红，姐妹们看过来]]></copy_display_markdown>
-  <clips>
-    <clip index="1">
-      <time_range>0-5秒</time_range>
-      <voiceover>姐妹们，今天试一支超显白的色号</voiceover>
-      <visual>近景平视，暖光下手持口红缓慢展示</visual>
-      <reference_image_index>0</reference_image_index>
-    </clip>
-    <clip index="2">
-      <time_range>5-10秒</time_range>
-      <voiceover>你看这个颜色，涂上去整个手都亮了</voiceover>
-      <visual>特写，嘴唇涂抹口红过程</visual>
-      <reference_image_index>1</reference_image_index>
-    </clip>
-    <clip index="3">
-      <time_range>10-15秒</time_range>
-      <voiceover>真的超好看，推荐给你们</voiceover>
-      <visual>中景，微笑展示最终效果</visual>
-      <reference_image_index>0</reference_image_index>
-    </clip>
-  </clips>
-  <voiceover_script>姐妹们，今天试一支超显白的色号。你看这个颜色，涂上去整个手都亮了。真的超好看，推荐给你们。</voiceover_script>
-  <theme>口红试色分享</theme>
-</script>"""
-
-    def test_dual_format_extracts_xml_block(self, monkeypatch, mock_job):
-        """双段式输入应正确提取 <script> 块并按 XML 路径解析。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.DUAL_FORMAT_INPUT, mock_job)
-        assert result is not None
-        assert result["voiceover_script"]
-        assert "姐妹们" in result["voiceover_script"]
-        assert len(result["shots"]) == 3
-        assert result["title"] == "口红试色分享"
-
-    def test_dual_format_shots_have_voiceover(self, monkeypatch, mock_job):
-        """双段式解析后每个 shot 都应有独立 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.DUAL_FORMAT_INPUT, mock_job)
-        assert result is not None
-        for shot in result["shots"]:
-            assert "voiceover" in shot
-            assert isinstance(shot["voiceover"], str)
-        assert "超显白" in result["shots"][0]["voiceover"]
-        assert "推荐" in result["shots"][2]["voiceover"]
-
-
-# ════════════════════════════════════════════════════════════════════════
-# 2. 纯 XML 解析（兼容性测试）
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestPureXmlParsing:
-    """格式A：纯 XML，与原有 v3 格式完全一致。"""
-
-    PURE_XML = """<script>
-  <copy_display_markdown><![CDATA[分享一款口红]]></copy_display_markdown>
-  <clips>
-    <clip index="1">
-      <time_range>0-5秒</time_range>
-      <voiceover>大家好，今天分享一款口红</voiceover>
-      <visual>近景平视，缓慢推镜</visual>
-      <action_details>手持口红特写</action_details>
-      <audio_bgm>轻快流行BGM</audio_bgm>
-      <transition>硬切</transition>
-      <reference_image_index>0</reference_image_index>
-    </clip>
-    <clip index="2">
-      <time_range>5-10秒</time_range>
-      <voiceover>颜色特别好看</voiceover>
-      <visual>特写，固定镜头</visual>
-      <audio_bgm>轻快BGM继续</audio_bgm>
-      <transition>硬切</transition>
-      <reference_image_index>1</reference_image_index>
-    </clip>
-    <clip index="3">
-      <time_range>10-15秒</time_range>
-      <voiceover>很显白，推荐给大家</voiceover>
-      <visual>中景，微笑展示</visual>
-      <audio_bgm>轻快BGM结束</audio_bgm>
-      <transition>结束</transition>
-      <reference_image_index>0</reference_image_index>
-    </clip>
-  </clips>
-  <voiceover_script>大家好呀，今天来给大家分享一款超显白的口红。颜色特别好看很显气质，真心推荐给姐妹们</voiceover_script>
-  <theme>口红分享</theme>
-</script>"""
-
-    def test_pure_xml_parses_correctly(self, monkeypatch, mock_job):
-        """纯 XML 格式应能正确解析出所有字段。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.PURE_XML, mock_job)
-        assert result is not None
-        assert result["title"] == "口红分享"
-        assert len(result["shots"]) == 3
-        assert "口红" in result["voiceover_script"]
-
-    def test_pure_xml_shots_have_voiceover(self, monkeypatch, mock_job):
-        """纯 XML 路径也应为每个 shot 生成 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.PURE_XML, mock_job)
-        assert result is not None
-        assert result["shots"][0]["voiceover"] == "大家好，今天分享一款口红"
-        assert result["shots"][1]["voiceover"] == "颜色特别好看"
-        assert result["shots"][2]["voiceover"] == "很显白，推荐给大家"
-
-
-# ════════════════════════════════════════════════════════════════════════
-# 3. 纯中文 markdown 解析（格式C 兜底）
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestPureMarkdownParsing:
-    """格式C：纯中文结构化 markdown，无 <script> 标签。"""
-
-    MARKDOWN_INPUT = """## 视频主题：咖啡店探店
-
-这是一段关于咖啡店探店的短视频分镜。
-
-### 镜头1
-- 时间范围：0-5秒
-- 景别：近景平视
-- 运镜：缓慢推镜
-- 口播："大家好，今天带你们来一家超有氛围的咖啡店"
-- 画面：暖光木质装修的咖啡店门面，镜头缓缓推近
-- 动作：推门进入
-- 音效：轻爵士BGM
-- 转场：硬切
-- 参考图：1
-
-### 镜头2
-- 时间范围：5-10秒
-- 景别：特写
-- 运镜：固定镜头
-- 口播："你看这杯拿铁，拉花也太好看了吧"
-- 画面：咖啡师制作拿铁特写，奶泡拉花清晰可见
-- 动作：咖啡师拉花过程
-- 音效：咖啡机声+轻爵士
-- 转场：叠化
-- 参考图：2
-
-### 镜头3
-- 时间范围：10-15秒
-- 景别：中景
-- 运镜：缓慢拉远
-- 口播："推荐给你们，周末来坐坐吧"
-- 画面：店内全景，客人悠闲品咖啡，暖光氛围
-- 动作：端起杯子品尝
-- 音效：轻爵士BGM渐弱
-- 转场：结束
-- 参考@图片1"""
-
-    def test_markdown_parses_shots(self, monkeypatch, mock_job):
-        """纯中文 markdown 应能解析出3个镜头。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert len(result["shots"]) == 3
-
-    def test_markdown_extracts_theme(self, monkeypatch, mock_job):
-        """应从"视频主题"提取 theme。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert "咖啡店" in result["title"]
-
-    def test_markdown_extracts_voiceover(self, monkeypatch, mock_job):
-        """每个镜头应提取口播到 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert "咖啡店" in result["shots"][0]["voiceover"]
-        assert "拿铁" in result["shots"][1]["voiceover"]
-        assert "推荐" in result["shots"][2]["voiceover"]
-
-    def test_markdown_extracts_time_range(self, monkeypatch, mock_job):
-        """应正确提取各镜头的 time_range。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert result["shots"][0]["time_range"] == "0-5秒"
-        assert result["shots"][1]["time_range"] == "5-10秒"
-        assert result["shots"][2]["time_range"] == "10-15秒"
-
-    def test_markdown_extracts_audio_bgm(self, monkeypatch, mock_job):
-        """应正确提取音效/BGM。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert "爵士" in result["shots"][0]["audio_bgm"] or "BGM" in result["shots"][0]["audio_bgm"]
-
-    def test_markdown_extracts_transition(self, monkeypatch, mock_job):
-        """应正确提取转场。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert result["shots"][0]["transition"] == "硬切"
-        assert result["shots"][1]["transition"] == "叠化"
-        assert result["shots"][2]["transition"] == "结束"
-
-    def test_markdown_extracts_reference_image(self, monkeypatch, mock_job):
-        """应正确提取参考图索引（从1开始转为0-based）。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert result["shots"][0]["reference_image_index"] == 0  # 参考图：1 -> 0
-        assert result["shots"][1]["reference_image_index"] == 1  # 参考图：2 -> 1
-        # 镜头3用"参考@图片1"格式 -> 0
-        assert result["shots"][2]["reference_image_index"] == 0
-
-    def test_markdown_voiceover_script_concatenated(self, monkeypatch, mock_job):
-        """voiceover_script 应是各镜头口播的拼接。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(self.MARKDOWN_INPUT, mock_job)
-        assert result is not None
-        assert "咖啡店" in result["voiceover_script"]
-        assert "拿铁" in result["voiceover_script"]
-        assert "推荐" in result["voiceover_script"]
-
-    def test_markdown_chinese_number_shots(self, monkeypatch, mock_job):
-        """支持中文数字镜头标识（镜头一/镜头二/...）。"""
-        wv = _import_wv(monkeypatch)
-        md_cn = """## 视频主题：测试
-
-### 镜头一
-- 时间范围：0-5秒
-- 口播："第一段口播内容很长足够通过校验"
-- 画面：近景展示
-
-### 镜头二
-- 时间范围：5-10秒
-- 口播："第二段口播内容也要足够长才行"
-- 画面：特写展示
-
-### 镜头三
-- 时间范围：10-15秒
-- 口播："第三段口播内容结尾"
-- 画面：中景展示
-"""
-        result = wv._parse_script(md_cn, mock_job)
-        assert result is not None
-        assert len(result["shots"]) == 3
-
-
-# ════════════════════════════════════════════════════════════════════════
-# 4. Shot voiceover 字段 + _assemble_seedance_prompt
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestShotVoiceoverField:
-    """确认 parser 输出的每个 shot dict 都有独立 voiceover 字段。"""
-
-    def test_xml_path_shot_voiceover(self, monkeypatch, mock_job):
-        """XML 路径解析后 shot 有 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        xml = """<script>
-<clips>
-  <clip index="1">
-    <time_range>0-5秒</time_range>
-    <voiceover>第一句口播</voiceover>
-    <visual>近景展示</visual>
-    <reference_image_index>0</reference_image_index>
-  </clip>
-  <clip index="2">
-    <time_range>5-10秒</time_range>
-    <voiceover>第二句口播</voiceover>
-    <visual>特写展示</visual>
-    <reference_image_index>-1</reference_image_index>
-  </clip>
-  <clip index="3">
-    <time_range>10-15秒</time_range>
-    <voiceover>第三句口播</voiceover>
-    <visual>中景展示</visual>
-    <reference_image_index>0</reference_image_index>
-  </clip>
-</clips>
-<voiceover_script>第一句口播。第二句口播。第三句口播。</voiceover_script>
-<theme>测试主题</theme>
-</script>"""
-        result = wv._parse_script(xml, mock_job)
-        assert result is not None
-        for shot in result["shots"]:
-            assert "voiceover" in shot
-        assert result["shots"][0]["voiceover"] == "第一句口播"
-        assert result["shots"][1]["voiceover"] == "第二句口播"
-        assert result["shots"][2]["voiceover"] == "第三句口播"
-
-    def test_fallback_script_shot_voiceover(self, monkeypatch, mock_job):
-        """_fallback_script 的 shot dict 也应有 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        fb = wv._fallback_script(mock_job)
-        assert fb["shots"]
-        for shot in fb["shots"]:
-            assert "voiceover" in shot
-            assert shot["voiceover"] == fb["voiceover_script"]
-
-    def test_assemble_seedance_uses_shot_voiceover(self, monkeypatch, mock_job):
-        """_assemble_seedance_prompt 应能取到 shot 的 voiceover 字段。"""
-        wv = _import_wv(monkeypatch)
-        mock_job.video_model = "seedance-2.5"
-        cr = {
-            "overview": {"theme": "测试", "total_duration": 15, "aspect_ratio": "9:16"},
-            "scene_and_lighting": "明亮",
-            "shots": [
-                {
-                    "time_range": "0-5秒",
-                    "shot_type_angle_movement": "近景",
-                    "scene_and_dialogue": "画面描述｜测试口播",
-                    "action_details": "展示",
-                    "audio_bgm": "BGM",
-                    "transition": "硬切",
-                    "reference_image_index": 0,
-                    "voiceover": "这是第一句口播",
-                },
-                {
-                    "time_range": "5-10秒",
-                    "shot_type_angle_movement": "特写",
-                    "scene_and_dialogue": "画面描述",
-                    "action_details": "",
-                    "audio_bgm": "BGM",
-                    "transition": "硬切",
-                    "reference_image_index": 0,
-                    "voiceover": "这是第二句口播",
-                },
-                {
-                    "time_range": "10-15秒",
-                    "shot_type_angle_movement": "中景",
-                    "scene_and_dialogue": "画面描述",
-                    "action_details": "",
-                    "audio_bgm": "BGM",
-                    "transition": "结束",
-                    "reference_image_index": 0,
-                    "voiceover": "这是第三句口播",
-                },
-            ],
-            "hard_constraints": ["无字幕"],
-            "negative_prompts": ["水印"],
-        }
-        prompt = wv._assemble_seedance_prompt(cr, mock_job)
-        # Seedance 格式应包含口播内容
-        assert "第一句口播" in prompt
-        assert "第二句口播" in prompt
-
-
-# ════════════════════════════════════════════════════════════════════════
-# 5. 参数传递集成测试（3个核心参数）
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestParameterPassing:
-    """验证核心用户参数从 job -> build_messages 的传递链路。"""
-
-    def test_marketing_purpose_in_messages(self, monkeypatch):
-        """marketing_purpose='门店引流' 应出现在 build_messages 的 user 消息中。"""
-        from packages.application.viral_video.prompt_loader import build_messages, get_template, invalidate
-
-        invalidate()
-        template = get_template("storyboard")
-        assert template is not None
-        messages = build_messages(
-            template,
-            marketing_purpose="门店引流",
-            marketing_purpose_hint="【营销目的：门店引流】以到店为核心",
-            industry="美容",
-            image_summary="",
-            theme_hint="门店引流",
-            duration="15",
-            image_count="2",
-            aspect_ratio="9:16",
-            tone="亲切自然",
-            target_audience="25-35岁女性",
-            target_customer_hint="【目标人群】25-35岁女性",
-            persona_hint="【人设】亲切自然",
-            viral_structure_hint="【爆款结构】自由组织",
-            language_hint="【语言】中文",
-            extra_requirements="（AI 创作）",
-            video_style_section="",
-            duration_constraint_text="视频时长15秒。口播总字数严格控制在38~75字之间",
-        )
-        user_msg = messages[-1]["content"]
-        assert "门店引流" in user_msg
-
-    def test_target_audience_in_messages(self, monkeypatch):
-        """target_audience='年轻妈妈' 应出现在 build_messages 的 user 消息中。"""
-        from packages.application.viral_video.prompt_loader import build_messages, get_template, invalidate
-
-        invalidate()
-        template = get_template("storyboard")
-        assert template is not None
-        messages = build_messages(
-            template,
-            marketing_purpose="品牌推广",
-            marketing_purpose_hint="【营销目的】品牌推广",
-            industry="母婴",
-            image_summary="",
-            theme_hint="母婴品牌",
-            duration="15",
-            image_count="1",
-            aspect_ratio="9:16",
-            tone="温柔亲切",
-            target_audience="年轻妈妈",
-            target_customer_hint="【目标人群】年轻妈妈群体",
-            persona_hint="【人设】温柔",
-            viral_structure_hint="【爆款结构】自由",
-            language_hint="【语言】中文",
-            extra_requirements="（AI 创作）",
-            video_style_section="",
-            duration_constraint_text="视频时长15秒。",
-        )
-        user_msg = messages[-1]["content"]
-        assert "年轻妈妈" in user_msg
-
-    def test_fusion_level_user_primary_in_messages(self, monkeypatch):
-        """fusion_level=user_primary + user_copy_text 应生成'必须原封不动使用'字样。"""
-        from packages.application.viral_video.prompt_loader import build_messages, get_template, invalidate
-
-        invalidate()
-        template = get_template("storyboard")
-        assert template is not None
-        # 模拟 worker 里 fusion_level=user_primary 时构造的 extra_req
-        extra_req = (
-            "【用户指定文案，必须原封不动使用，一个字都不要改，"
-            "你只需要根据文案生成分镜脚本的镜头描述和画面，"
-            "copy_display_markdown里的文案必须与用户原文完全一致】：\n"
-            "我们的新店开业啦，全场八折优惠"
-        )
-        messages = build_messages(
-            template,
-            marketing_purpose="促销活动",
-            marketing_purpose_hint="【营销目的】促销",
-            industry="零售",
-            image_summary="",
-            theme_hint="新店开业",
-            duration="15",
-            image_count="0",
-            aspect_ratio="9:16",
-            tone="热情",
-            target_audience="周边居民",
-            target_customer_hint="【目标人群】周边居民",
-            persona_hint="【人设】热情",
-            viral_structure_hint="【爆款结构】自由",
-            language_hint="【语言】中文",
-            extra_requirements=extra_req,
-            video_style_section="",
-            duration_constraint_text="视频时长15秒。",
-        )
-        user_msg = messages[-1]["content"]
-        assert "必须原封不动使用" in user_msg
-        assert "新店开业" in user_msg
-
-
-# ════════════════════════════════════════════════════════════════════════
-# 6. markdown 解析失败返回 None
-# ════════════════════════════════════════════════════════════════════════
-
-
-class TestMarkdownParseFailures:
-    """无法解析的输入应返回 None。"""
-
-    def test_plain_text_returns_none(self, monkeypatch, mock_job):
-        """纯文本不含镜头标识，应返回 None。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script("这是一段普通文本，没有任何镜头标识。", mock_job)
-        assert result is None
-
-    def test_empty_string_returns_none(self, monkeypatch, mock_job):
-        """空字符串应返回 None。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script("", mock_job)
-        assert result is None
-
-    def test_none_returns_none(self, monkeypatch, mock_job):
-        """None 输入应返回 None。"""
-        wv = _import_wv(monkeypatch)
-        result = wv._parse_script(None, mock_job)
-        assert result is None
-
-    def test_markdown_without_voiceover_returns_none(self, monkeypatch, mock_job):
-        """有镜头标识但没有口播内容，voiceover_script 为空，应返回 None。"""
-        wv = _import_wv(monkeypatch)
-        md = """### 镜头1
-- 画面：近景展示产品
-### 镜头2
-- 画面：特写展示产品
-### 镜头3
-- 画面：中景展示产品
-"""
-        result = wv._parse_script(md, mock_job)
-        assert result is None

tests/unit/test_viral_video_p0.py

@@ -84,6 +84,20 @@ class TestScriptGenerationV16:
         assert out["final_copy"] == out["voiceover_script"]
         assert len(out["shots"]) >= 1
 
+    def test_safe_json_loads_parses_fenced_code(self):
+        from apps.worker.worker_app.tasks.viral_video import _safe_json_loads
+
+        fenced = '```json\n{"voiceover_script": "你好", "shots": []}\n```'
+        out = _safe_json_loads(fenced)
+        assert out is not None
+        assert out["voiceover_script"] == "你好"
+
+    def test_safe_json_loads_handles_none(self):
+        from apps.worker.worker_app.tasks.viral_video import _safe_json_loads
+
+        assert _safe_json_loads(None) is None
+        assert _safe_json_loads("not json") is None
+
     def test_assemble_seedance_prompt_contains_sections(self, mock_job):
         from apps.worker.worker_app.tasks.viral_video import _assemble_seedance_prompt
 

tests/unit/test_viral_video_prompt_system.py

@@ -130,7 +130,7 @@ class TestTemplates:
             duration_constraint_text=build_duration_constraint_text(15),
         )
         assert "去油快" in rendered
-        assert "38~75字" in rendered
+        assert "38~60字" in rendered
 
 
 # ── constants ─────────────────────────────────────────────────────────────
@@ -149,7 +149,7 @@ class TestConstants:
 
     def test_duration_constraint_text_mentions_chars_and_shots(self):
         text = build_duration_constraint_text(15)
-        assert "38" in text and "75" in text
+        assert "38" in text and "60" in text
         assert "镜头" in text
 
     def test_get_expected_shot_count(self):

tests/unit/test_viral_video_routes.py

@@ -45,7 +45,6 @@ def _make_job(job_id: str = "job-1", user_id: str = "u1", status: str = "pending
         "persona_id": "",
         "viral_structure": "",
         "marketing_purpose": "",
-        "tone": "",
         "bgm_preference": "",
         "duration": 15,
         "user_copy_text": "",

tests/unit/test_viral_video_v8_bugs.py

@@ -34,7 +34,6 @@ def _make_job(job_id: str = "job-1", user_id: str = "u1", status: str = "pending
         "persona_id": "",
         "viral_structure": "",
         "marketing_purpose": "",
-        "tone": "",
         "bgm_preference": "",
         "duration": 15,
         "user_copy_text": "",
@@ -265,9 +264,9 @@ class TestBug2VoiceoverWordCount:
         assert "{duration_constraint_text}" in _STORYBOARD_USER
         # 15 秒规格 min_chars=38, max_chars=60，约束文本中含具体数字
         t15 = build_duration_constraint_text(15)
-        assert "38" in t15 and "75" in t15
+        assert "38" in t15 and "60" in t15
         t30 = build_duration_constraint_text(30)
-        assert "75" in t30 and "150" in t30
+        assert "75" in t30 and "120" in t30
         # get_duration_spec 兜底：越界 clamp 到 5/30，None/0 默认 15 秒档位
         assert get_duration_spec(5)["min_chars"] == 12
         assert get_duration_spec(999)["min_shots"] == 6
@@ -1121,401 +1120,3 @@ class TestIssue2249UserParamsFix:
 
         assert "marketing_purpose" in _STORYBOARD_SYSTEM or "persona" in _STORYBOARD_SYSTEM
         assert "不能忽略" in _STORYBOARD_SYSTEM or "必须" in _STORYBOARD_SYSTEM
-
-
-class TestV4ValidationDeadlineFixes:
-    """v4 编导分镜四叠 bug 回归测试（PR #2294）：
-    1. max_chars 放宽（15s 60→75）
-    2. _fallback_script brand==name 去重
-    3. 口播超长自动截断接受
-    4. build_messages few-shot 多轮消息
-    """
-
-    def test_duration_spec_relaxed(self):
-        """验证 15s max_chars 已放宽到 75，30s 到 150"""
-        from packages.application.viral_video.constants import get_duration_spec
-
-        assert get_duration_spec(15)["max_chars"] == 75
-        assert get_duration_spec(15)["min_chars"] == 38
-        assert get_duration_spec(10)["max_chars"] == 50
-        assert get_duration_spec(20)["max_chars"] == 100
-        assert get_duration_spec(30)["max_chars"] == 150
-        assert get_duration_spec(5)["max_chars"] == 25
-
-    def test_fallback_script_brand_name_dedup(self):
-        """_fallback_script: brand==name 时不应拼 'X家的X'"""
-        from unittest.mock import MagicMock, patch
-
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = MagicMock()
-        job.duration = 15
-        job.video_ratio = "9:16"
-        job.user_copy_text = None
-        job.images = []
-        # 构造 VLM 输出 brand==name
-        job.image_analysis = {"images": [{"type": "product", "name": "御众堂", "brand": "御众堂", "has_person": False}]}
-        job.marketing_purpose = ""
-        job.style_guide = None
-        job.persona_id = ""
-        job.viral_structure = ""
-        job.language = "zh-CN"
-        job.industry = ""
-        job.target_customer = ""
-        job.fusion_level = "ai_polish"
-
-        with patch.object(vv, "normalize_image_analysis", return_value=job.image_analysis):
-            result = vv._fallback_script(job)
-        vo = result["voiceover_script"]
-        # 不应出现"御众堂家的御众堂"重复
-        assert "御众堂家的御众堂" not in vo
-        assert "来看看御众堂" in vo
-        assert len(result["shots"]) == 1
-
-    def test_fallback_script_brand_name_different(self):
-        """_fallback_script: brand≠name 时仍正常拼接 'X家的Y'"""
-        from unittest.mock import MagicMock, patch
-
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = MagicMock()
-        job.duration = 15
-        job.video_ratio = "9:16"
-        job.user_copy_text = None
-        job.images = []
-        job.image_analysis = {
-            "images": [{"type": "product", "name": "修护精华", "brand": "御众堂", "has_person": False}]
-        }
-        job.marketing_purpose = ""
-        job.style_guide = None
-        job.persona_id = ""
-        job.viral_structure = ""
-        job.language = "zh-CN"
-        job.industry = ""
-        job.target_customer = ""
-        job.fusion_level = "ai_polish"
-
-        with patch.object(vv, "normalize_image_analysis", return_value=job.image_analysis):
-            result = vv._fallback_script(job)
-        vo = result["voiceover_script"]
-        assert "御众堂家的修护精华" in vo
-
-    def test_build_messages_with_example(self):
-        """build_messages: example_output 非空时生成 4 条消息 few-shot"""
-        from packages.application.viral_video.prompt_loader import (
-            PromptTemplate,
-            build_messages,
-        )
-
-        tpl = PromptTemplate(
-            name="storyboard",
-            prompt_type="storyboard",
-            version=99,
-            system_prompt="系统：你是分镜师",
-            user_prompt_template="请为{product}写脚本",
-            example_output="<script>示例输出</script>",
-        )
-        msgs = build_messages(tpl, product="测试商品")
-        assert len(msgs) == 4
-        assert msgs[0]["role"] == "system"
-        assert msgs[1]["role"] == "user"
-        assert "示例" in msgs[1]["content"]
-        assert msgs[2]["role"] == "assistant"
-        assert "<script>示例输出</script>" in msgs[2]["content"]
-        assert msgs[3]["role"] == "user"
-        assert "测试商品" in msgs[3]["content"]
-
-    def test_build_messages_without_example(self):
-        """build_messages: example_output 为空时退化为 2 条消息"""
-        from packages.application.viral_video.prompt_loader import (
-            PromptTemplate,
-            build_messages,
-        )
-
-        tpl = PromptTemplate(
-            name="storyboard",
-            prompt_type="storyboard",
-            version=99,
-            system_prompt="系统：你是分镜师",
-            user_prompt_template="请写脚本",
-            example_output="",
-        )
-        msgs = build_messages(tpl)
-        assert len(msgs) == 2
-        assert msgs[0]["role"] == "system"
-        assert msgs[1]["role"] == "user"
-
-    def test_truncate_voiceover_preserves_full_sentences(self):
-        """_truncate_voiceover: 按句号切句保留完整句，不截到半句"""
-        from apps.worker.worker_app.tasks.viral_video import _truncate_voiceover
-
-        text = "今天给大家推荐一款好产品。它的效果特别棒，用完皮肤水水润润的。大家一定要试试哦。"
-        out = _truncate_voiceover(text, 25)
-        assert len(out) <= 25
-        assert out.endswith("。")
-        assert "推荐一款好产品" in out
-
-
-class TestStreamFinishReasonLength:
-    """流式finish_reason=length截断检测，确保截断raw被丢弃走下一attempt（P0 2026-10-14）"""
-
-    def _make_client(self, chunks, finish_after=None, finish_reason="stop"):
-        from unittest.mock import MagicMock
-
-        client = MagicMock()
-        client.is_available = True
-        client.model = "test-model"
-        client.last_finish_reason = ""
-
-        def _stream_iter(*a, **kw):
-            for i, ch in enumerate(chunks):
-                yield ch
-                if finish_after is not None and i + 1 >= finish_after:
-                    client.last_finish_reason = finish_reason
-                    return
-
-        client.chat_completion_stream = MagicMock(side_effect=lambda *a, **kw: _stream_iter())
-        client.chat_completion = MagicMock(return_value=None)
-        return client
-
-    def test_length_truncation_returns_none(self):
-        """流式被max_tokens截断时应返回None，不降级同步，让外层重试"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from unittest.mock import MagicMock
-
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        partial = '<script><copy_display_markdown>文案</copy_display_markdown><clips><clip index="1"><time_range>0-5秒</time_range>'
-        client = self._make_client([partial], finish_after=1, finish_reason="length")
-        on_delta = MagicMock()
-        raw = vv._call_llm_once(
-            client,
-            messages=[{"role": "user", "content": "hi"}],
-            temperature=0.8,
-            max_tokens=2500,
-            timeout=60,
-            on_delta=on_delta,
-        )
-        assert raw is None, "截断时应返回None让外层重试"
-        client.chat_completion.assert_not_called()
-
-    def test_normal_stop_returns_raw(self):
-        """正常finish_reason=stop时应返回完整raw"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from unittest.mock import MagicMock
-
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        full = "<script><copy_display_markdown>文案</copy_display_markdown><clips></clips></script>"
-        client = self._make_client([full], finish_after=1, finish_reason="stop")
-        on_delta = MagicMock()
-        raw = vv._call_llm_once(
-            client,
-            messages=[{"role": "user", "content": "hi"}],
-            temperature=0.8,
-            max_tokens=2500,
-            timeout=60,
-            on_delta=on_delta,
-        )
-        assert raw == full
-
-    def test_noop_lines_removed(self):
-        """确认死代码行已删除"""
-        import inspect
-
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        src = inspect.getsource(vv._step_script_generation)
-        assert 'llm_messages[0]["content"]' not in src
-        assert 'llm_messages[-1]["content"]' not in src
-
-
-class TestMarkdownTimeRangeSplit:
-    """markdown 解析器支持按"0-4秒/4-8秒"时间范围开头分镜（v6 中文格式）"""
-
-    def test_time_range_split_produces_correct_shots(self):
-        """纯中文 markdown 用「0-4秒」「4-8秒」分段，应正确切分镜头"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = _make_job(duration=20, marketing_purpose="产品展示")
-        raw = (
-            "视频主题: 御众堂产后修复\n"
-            "0-4秒\n"
-            "画面: 门店招牌特写，阳光照射\n"
-            "景别: 近景平拍，缓慢推进\n"
-            '口播: "宝妈们别划走"\n'
-            "BGM: 轻快钢琴\n"
-            "4-8秒\n"
-            "画面: 店内环境，技师微笑迎客\n"
-            "景别: 中景平视，固定镜头\n"
-            '口播: "御众堂产后修复，专业呵护"\n'
-            "BGM: 轻快钢琴\n"
-            "8-14秒\n"
-            "画面: 修复仪器操作特写\n"
-            "景别: 特写俯拍，缓慢横移\n"
-            '口播: "先进设备，科学恢复"\n'
-            "转场: 硬切\n"
-            "14-20秒\n"
-            "画面: 宝妈满意笑容+门店地址字幕\n"
-            "景别: 中景平视，缓慢拉远\n"
-            '口播: "点击下方预约，到店体验"\n'
-        )
-        result = vv._parse_script_markdown(raw, job, 20, "9:16", "产后修复")
-        assert result is not None, "时间范围分块应解析成功"
-        assert len(result["shots"]) == 4, f"应切分4个镜头，实际{len(result['shots'])}"
-        assert result["shots"][0]["time_range"] == "0-4秒"
-        assert result["shots"][1]["time_range"] == "4-8秒"
-        assert result["shots"][2]["time_range"] == "8-14秒"
-        assert result["shots"][3]["time_range"] == "14-20秒"
-        assert "宝妈们别划走" in result["shots"][0]["voiceover"]
-        assert "御众堂产后修复" in result["shots"][1]["voiceover"]
-        assert "先进设备" in result["shots"][2]["voiceover"]
-        assert "点击下方预约" in result["shots"][3]["voiceover"]
-        assert result["title"] == "御众堂产后修复"
-        full_vo = result["voiceover_script"]
-        assert "宝妈们别划走" in full_vo
-        assert "点击下方预约" in full_vo
-
-    def test_time_range_with_dash_and_tilde(self):
-        """支持「0-4秒」「4~8秒」「12到20秒」多种连接符"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = _make_job(duration=12)
-        raw = (
-            '0~4秒\n画面: 产品开箱\n口播: "快看这个"\n'
-            '4-8秒\n画面: 产品细节\n口播: "质感满分"\n'
-            '8到12秒\n画面: 使用效果\n口播: "赶紧下单"\n'
-        )
-        result = vv._parse_script_markdown(raw, job, 12, "9:16", "开箱")
-        assert result is not None
-        assert len(result["shots"]) == 3
-        assert result["shots"][0]["time_range"] == "0-4秒"
-        assert result["shots"][1]["time_range"] == "4-8秒"
-        assert result["shots"][2]["time_range"] == "8-12秒"
-
-    def test_shot_number_marker_takes_priority_over_time(self):
-        """同时有「镜头1」和「0-4秒」时，优先按镜头编号分割（不重复切分）"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = _make_job(duration=8)
-        raw = '镜头1 0-4秒\n画面: 开场\n口播: "开场话"\n' '镜头2 4-8秒\n画面: 主体\n口播: "主体话"\n'
-        result = vv._parse_script_markdown(raw, job, 8, "9:16", "测试")
-        assert result is not None
-        assert len(result["shots"]) == 2
-        assert result["shots"][0]["time_range"] == "0-4秒"
-
-    def test_no_markers_returns_none(self):
-        """没有任何分镜标记时返回None"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = _make_job(duration=10)
-        raw = "今天给大家推荐一个好物，真的非常好用，价格实惠。"
-        result = vv._parse_script_markdown(raw, job, 10, "9:16", "推荐")
-        assert result is None
-
-    def test_parse_script_routes_to_time_markdown(self):
-        """_parse_script 调度：无XML时自动走时间范围markdown解析"""
-        import os
-        import sys
-
-        sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        job = _make_job(duration=8)
-        raw = '0-4秒\n画面: 开场\n口播: "你好"\n' '4-8秒\n画面: 结束\n口播: "再见"\n'
-        result = vv._parse_script(raw, job)
-        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