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

- tone字段：domain entity/SQLAlchemy model/JSONB列/API schema(GenerateCopyRequest+ViralVideoJobResponse)/route赋值/_to_response/repo save/update/_to_domain + alembic migration #110 全链路打通
- ShotScript schema补voiceover字段（parser已输出，API层缺失）
- 删除_safe_json_loads死代码（worker viral_video.py 24行+2个废弃单测）
- 补充tests mock_job默认tone字段
- 267 viral单测全过

apps/worker/worker_app/tasks/viral_video.py

@@ -444,7 +444,6 @@ 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
@@ -458,15 +457,9 @@ 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)
@@ -481,54 +474,9 @@ 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)
@@ -538,7 +486,7 @@ def _call_llm_once(
         if raw:
             full_parts = [raw]
             try:
-                on_delta(raw, raw)
+                on_delta(raw, raw)  # 一次性推送
             except Exception:
                 pass
         else:
@@ -1124,15 +1072,13 @@ 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..."分段
-    格式D（纯中文 markdown 时间式）：无 <script> 标签，按"0-4秒""4-8秒"时间范围开头分段（v6 格式）
+    格式C（纯中文 markdown）：无 <script> 标签，按"镜头1/镜头2..."分段
 
     解析策略：
     1. 正则提取最后一个 <script>...</script> 块
     2. 有 XML 块 → 走 XML 解析
-    3. 无 XML 块但有 v3 结构标签 → 裸 XML 解析
-    4. 都没有 → 走中文 markdown 兜底解析（C/D 两种分块方式自动识别）
-    5. 都解析不出 → 返回 None
+    3. 无 XML 块 → 走中文 markdown 兜底解析
+    4. 都解析不出 → 返回 None
     """
     if not raw:
         return None
@@ -1221,11 +1167,6 @@ _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]+)"),
@@ -1248,14 +1189,14 @@ def _parse_script_markdown(
     ratio: str,
     default_theme: str,
 ) -> dict | None:
-    """中文 markdown 兜底解析（格式C/D）：按"镜头1/镜头一/第1镜"或"0-4秒/4-8秒"时间范围分割。"""
+    """中文 markdown 兜底解析（格式C）：按"镜头1/镜头一/第1镜"等分割。"""
     # 提取 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)
@@ -1270,11 +1211,6 @@ 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
 
@@ -1421,27 +1357,6 @@ def _step_script_generation(
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
 
     template = get_template("storyboard")
-    # P0 观测：模板来自代码默认（DB 加载失败）时打 ERROR，避免静默回落
-    # 判定方式：system_prompt 与 DEFAULT_TEMPLATES storyboard 完全一致
-    try:
-        from packages.application.viral_video.prompts import DEFAULT_TEMPLATES as _DEF_T
-
-        _def_story = next((t for t in _DEF_T if t["prompt_type"] == "storyboard"), None)
-        if (
-            template is not None
-            and _def_story is not None
-            and template.system_prompt == _def_story["system_prompt"]
-            and (template.example_output or "") == (_def_story.get("example_output") or "")
-        ):
-            logger.error(
-                "[爆款视频] storyboard 模板使用代码默认 v%s/%s（DB 加载失败或无激活模板），"
-                "检查 DB 连接与 is_active——本次生成质量无保障，job_id=%s",
-                template.version,
-                template.name,
-                job.id,
-            )
-    except Exception:  # noqa: BLE001
-        pass
     marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
     theme_hint = _determine_theme(image_analysis, marketing_purpose)
 
@@ -1505,6 +1420,8 @@ 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")
@@ -1545,26 +1462,10 @@ 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(
-                    "[爆款视频] 分镜脚本解析失败 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())

packages/application/viral_video/prompt_loader.py

@@ -166,63 +166,17 @@ 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是配置bug（如v5事故里{duration.1格式写全6项参数}），
-        # 提升到 ERROR 级别，避免被warning海淹没
-        logger.error(
-            "[viral][prompt] 模板格式化有%d个占位符无法渲染（示例：%s），"
-            "已原样保留——请检查system/user_prompt_template中花括号是否为样例文本而非变量",
-            len(errors),
-            errors[0],
-        )
-        _warn_once("safe_format", "format", f"{len(errors)}个占位符异常（首条{errors[0]}）")
-    return "".join(out)
+    try:
+        return text.format_map(_SafeDict(kwargs))
+    except Exception as e:  # noqa: BLE001
+        _warn_once("safe_format", "format", e)
+        return text
 
 
 def render_system_prompt(template: PromptTemplate, **kwargs) -> str:

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"),
     )
 

tests/unit/test_viral_video_v8_bugs.py

@@ -1254,268 +1254,3 @@ 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"]
-
-
-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