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

- 新增 _MD_TIME_SPLIT_RE 按行首时间范围标记切分镜头（v6中文分镜格式兜底）
- 镜头编号标记优先，无编号标记时fallback到时间范围标记
- 支持「-」「~」「到」「至」多种连接符
- 流式early-break标记列表增加「0-」「0~」「1-」等时间范围开头模式
- 更新_parse_script docstring 增加格式D说明
- 新增5个单测覆盖时间范围分块、多种连接符、编号优先、无标记返回None、_parse_script调度
- 275 viral passed (270+5)

apps/worker/worker_app/tasks/viral_video.py

@@ -444,6 +444,7 @@ def _call_llm_once(
     # 这里用外层 deadline + 迭代超时双重保护。
     _stream_deadline = time.time() + max(timeout or 120, 30)
     stream_ok = False
+    _length_truncated = False  # finish_reason=length 时标记为被截断（raw 不完整，parser 必败）
     try:
         stream_iter = client.chat_completion_stream(
             messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout or 120
@@ -457,9 +458,15 @@ def _call_llm_once(
                 chunk = next(stream_iter)
             except StopIteration:
                 stream_ok = True
+                # 流结束时再检查一次 finish_reason（最后一个 chunk 可能带 finish_reason 但无 content）
+                if getattr(client, "last_finish_reason", "") == "length":
+                    _length_truncated = True
+                    logger.warning(
+                        "[爆款视频] 流式结束时检测到 finish_reason=length（received=%d chars），丢弃走下一 attempt",
+                        sum(len(p) for p in full_parts),
+                    )
                 break
             except Exception as inner_e:
-                # 超时类异常直接 break（不降级同步，因为已耗尽超时配额）
                 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)
@@ -474,9 +481,54 @@ 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
     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:
             raw = client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout)
@@ -486,7 +538,7 @@ def _call_llm_once(
         if raw:
             full_parts = [raw]
             try:
-                on_delta(raw, raw)  # 一次性推送
+                on_delta(raw, raw)
             except Exception:
                 pass
         else:
@@ -1072,13 +1124,15 @@ def _parse_script(raw: str, job: ViralVideoJob) -> dict | None:
 
     格式A（纯 XML）：整段都是 <script>...</script> XML（v3/v4 格式）
     格式B（双段式）：中文可读说明 + 末尾 <script>...</script> XML 块
-    格式C（纯中文 markdown）：无 <script> 标签，按"镜头1/镜头2..."分段
+    格式C（纯中文 markdown 编号式）：无 <script> 标签，按"镜头1/镜头2..."分段
+    格式D（纯中文 markdown 时间式）：无 <script> 标签，按"0-4秒""4-8秒"时间范围开头分段（v6 格式）
 
     解析策略：
     1. 正则提取最后一个 <script>...</script> 块
     2. 有 XML 块 → 走 XML 解析
-    3. 无 XML 块 → 走中文 markdown 兜底解析
-    4. 都解析不出 → 返回 None
+    3. 无 XML 块但有 v3 结构标签 → 裸 XML 解析
+    4. 都没有 → 走中文 markdown 兜底解析（C/D 两种分块方式自动识别）
+    5. 都解析不出 → 返回 None
     """
     if not raw:
         return None
@@ -1167,6 +1221,11 @@ _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]+)"),
@@ -1196,7 +1255,7 @@ def _parse_script_markdown(
     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)
@@ -1211,6 +1270,11 @@ def _parse_script_markdown(
                 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
 
@@ -1420,8 +1484,6 @@ def _step_script_generation(
         video_style_section=style_section,
         duration_constraint_text=_duration_constraint_text,
     )
-    llm_messages[0]["content"]
-    llm_messages[-1]["content"]
 
     # primary 模型（DB 配置的首选，豆包 Seed 2.1 Pro，稳但稍慢）
     client_primary = ai_router.get_llm_client("storyboard", variant="primary")
@@ -1462,10 +1524,26 @@ def _step_script_generation(
                 on_delta=on_delta,
             )
             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)
             if copy_result is None:
-                logger.warning("[viral][script] attempt=%s XML解析失败 raw_len=%d", label, len(raw))
+                logger.warning(
+                    "[爆款视频] 分镜脚本解析失败 label=%s raw_len=%d head=%r tail=%r",
+                    label,
+                    len(raw),
+                    raw[:80].replace("\n", "\\n"),
+                    raw[-80:].replace("\n", "\\n"),
+                )
                 continue
             reason = _validate_copy_result(copy_result, dur)
             voiceover_len = len((copy_result.get("voiceover_script") or "").strip())

tests/unit/test_viral_video_v8_bugs.py

@@ -1254,3 +1254,198 @@ class TestV4ValidationDeadlineFixes:
         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"]