fix(viral-video) P0/P1: Seedance模型ID映射+很近全局替换+失败状态强落库 (#2157)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

apps/worker/worker_app/tasks/viral_video.py

@@ -27,6 +27,7 @@ import threading
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
+from typing import Any
 
 from celery import Task, shared_task
 from celery.exceptions import Retry
@@ -728,6 +729,22 @@ def _safe_json_loads(raw: str | dict | list | None):
     return None
 
 
+def _replace_henjin_everywhere(obj: Any) -> Any:
+    """递归遍历 copy_result 里所有字符串值，把'很近'替换成'最近'。
+    覆盖 overview.theme、scene_and_lighting、voiceover_script、
+    shots[].scene_and_dialogue/action_details/audio_bgm 等所有字段。
+    """
+    if isinstance(obj, str):
+        if "很近" in obj:
+            return obj.replace("很近", "最近")
+        return obj
+    if isinstance(obj, list):
+        return [_replace_henjin_everywhere(x) for x in obj]
+    if isinstance(obj, dict):
+        return {k: _replace_henjin_everywhere(v) for k, v in obj.items()}
+    return obj
+
+
 def _fallback_script(job: ViralVideoJob) -> dict:
     """脚本生成失败时的兜底脚本（极简但可用）。"""
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
@@ -920,19 +937,14 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
         voiceover_len = len(voiceover)
         shots_cnt = len((normalized or {}).get("shots") or [])
         # 判定是否"退化到兜底质量"：口播过短（<20字）或镜头数<1；正常的短口播（如15s视频~40字）不视为兜底
-        # v1.6.1 双保险：先做 hard fix 字符串替换（"很近" → "最近"），再做不合格判定
-        if "很近" in voiceover:
-            logger.warning("[爆款视频] 编导脚本含错别字'很近'，hard fix 替换为'最近' label=%s", label)
-            voiceover = voiceover.replace("很近", "最近")
-            normalized["voiceover_script"] = voiceover
-            # 同时在 shots 对白里替换
-            for sh in normalized.get("shots") or []:
-                if isinstance(sh, dict):
-                    sd = sh.get("scene_and_dialogue") or ""
-                    if "很近" in sd:
-                        sh["scene_and_dialogue"] = sd.replace("很近", "最近")
+        # v1.6.1 P1修复：递归替换 copy_result 里所有字符串字段的"很近"→"最近"（覆盖 overview/scene_and_lighting/voiceover/shots.* 全部字段）
+        _before_dump = json.dumps(normalized, ensure_ascii=False)
+        if "很近" in _before_dump:
+            logger.warning("[爆款视频] 编导脚本含错别字'很近'，递归替换为'最近' label=%s", label)
+            normalized = _replace_henjin_everywhere(normalized)
+            voiceover = (normalized or {}).get("voiceover_script") or ""
         fallback_marker = "我最近在用的好物" in voiceover  # _fallback_script 的特征串
-        has_typo_henjin = "很近" in voiceover  # v1.6.1: 错别字"很近"视为不合格，触发重试
+        has_typo_henjin = "很近" in json.dumps(normalized, ensure_ascii=False)  # 递归检查仍有"很近"视为不合格
         is_fallback = fallback_marker or shots_cnt < 1 or voiceover_len < 20 or has_typo_henjin
         logger.info(
             "[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s raw_type=%s",
@@ -1328,16 +1340,36 @@ def run_video_style_analysis(self: Task, job_id: str) -> dict:
 
 
 def _mark_failed_and_notify(job_id: str, session, repo, job, err_msg: str, stage: str = "") -> None:
+    """标记任务失败并通知。若传入的 session 已失效（因前面异常导致 rollback 状态），
+    会自动 fallback 到新建 SessionLocal 重新标记，确保状态一定落库。"""
     try:
-        if session is None:
-            session = SessionLocal()
-            repo = SQLAlchemyViralVideoJobRepository(session)
-            job = repo.get(job_id)
-        if job is not None and not job.is_terminal:
-            job.mark_failed(err_msg)
-            _save_job(repo, job, session)
+        # 尝试用传入的 session 标记
+        marked = False
+        if job is not None and not job.is_terminal and session is not None:
+            try:
+                job.mark_failed(err_msg)
+                _save_job(repo, job, session)
+                marked = True
+            except Exception as se:
+                logger.warning("[爆款视频] 用原 session 标记失败失败，fallback 新session: %s", se)
+                try:
+                    session.rollback()
+                except Exception:
+                    pass
+        if not marked:
+            # fallback：新建独立 session 重新标记（保证状态一定落库）
+            ssn = SessionLocal()
+            try:
+                r = SQLAlchemyViralVideoJobRepository(ssn)
+                j = r.get(job_id)
+                if j is not None and not j.is_terminal:
+                    j.mark_failed(err_msg)
+                    r.update(j)
+                    ssn.commit()
+            finally:
+                ssn.close()
     except Exception as inner:
-        logger.warning("[爆款视频] 标记失败状态时出错: %s", inner)
+        logger.warning("[爆款视频] 标记失败状态时出错（最终fallback也失败）: %s", inner, exc_info=True)
     _emit_progress(
         job_id,
         stage,

packages/shared/ai_client.py

@@ -25,6 +25,41 @@ from packages.shared.config import get_shared_settings
 logger = logging.getLogger(__name__)
 
 
+# 内部 pricing key → 火山方舟实际模型 ID
+# 前端/后端内部用简短别名 seedance-2.5/seedance-2.0 做 VIRAL_VIDEO_MODEL_PRICES 配置 key，
+# 但传给方舟 API 时必须用真实模型 ID（如 doubao-seedance-2-5-260628），否则会 404 InvalidEndpointOrModel.NotFound。
+VIRAL_VIDEO_MODEL_ID_MAP: dict[str, str] = {
+    "seedance-2.5": "doubao-seedance-2-5-260628",
+    "seedance-2.0": "doubao-seedance-2-0-250628",
+}
+
+
+def _resolve_video_model_id(model: str | None) -> str:
+    """把内部 pricing key（seedance-2.5/seedance-2.0）解析成方舟实际模型 ID。
+    未识别的别名或空值回落到 settings.doubao_video_model 默认值。
+    """
+    settings = get_shared_settings()
+    default_model = getattr(settings, "doubao_video_model", None) or "doubao-seedance-2-5-260628"
+    m = (model or "").strip()
+    if not m:
+        return default_model
+    # 已经是 doubao- 前缀的完整 ID，直接用
+    if m.startswith("doubao-") or m.startswith("ep-"):
+        return m
+    # 别名 → 完整 ID
+    resolved = VIRAL_VIDEO_MODEL_ID_MAP.get(m.lower())
+    if resolved:
+        return resolved
+    # 未识别：带 seedance 字样且不是 doubao-/ep- 开头的，拼前缀兜底
+    if "seedance" in m.lower():
+        # seedance-2.5 → doubao-seedance-2-5-260628 等兜底逻辑
+        normalized = m.lower().replace(".", "-")
+        if normalized in VIRAL_VIDEO_MODEL_ID_MAP:
+            return VIRAL_VIDEO_MODEL_ID_MAP[normalized]
+    logger.warning("[ai_client] 未识别的 video_model=%r，使用默认模型 %s", m, default_model)
+    return default_model
+
+
 class DoubaoClient:
     """豆包大模型 API 客户端.
 
@@ -282,8 +317,8 @@ class DoubaoClient:
         poll_interval = getattr(settings, "doubao_video_poll_interval", 10) or 10
         # 收紧总超时：轮询 8min + 下载 2min = 最长 ~10min，防止出现 20min 卡死
         total_timeout = getattr(settings, "doubao_video_timeout", 480) or 480
-        default_video_model = getattr(settings, "doubao_video_model", None) or "doubao-seedance-2-5-260628"
-        video_model = model or default_video_model
+        # 内部 pricing key (seedance-2.5/seedance-2.0) → 方舟实际模型 ID
+        video_model = _resolve_video_model_id(model)
 
         ref_audios = [u for u in (reference_audios or [])[:10] if u and isinstance(u, str)]
         ref_videos = [u for u in (reference_videos or [])[:3] if u and isinstance(u, str)]
@@ -408,12 +443,13 @@ class DoubaoClient:
         if not task_id:
             logger.error(
                 "Seedance 创建任务最终失败: model=%s base_url=%s err=%s body=%s  【排查】"
-                "1) 方舟控制台已开通 doubao-seedance-2-5-260628；2) API Key 有该模型权限；"
+                "1) 方舟控制台已开通 %s；2) API Key 有该模型权限；"
                 "3) DOUBAO_BASE_URL=https://ark.cn-beijing.volces.com/api/v3；4) 参考素材 URL 公网可访问。",
                 video_model,
                 self.base_url,
                 last_err,
                 (body or "")[:500],
+                video_model,
             )
             return None
 

tests/unit/test_ai_client_video.py

@@ -483,3 +483,70 @@ class TestVideoGenerationCancelled:
                 doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
             )
             assert client.video_generation("p", output_dir=str(tmp_path)) is None
+
+
+# ============ #2157 _resolve_video_model_id 模型ID映射单测 ============
+
+
+class TestResolveVideoModelId:
+    """覆盖 _resolve_video_model_id 各分支（#2157 P0 修复）。"""
+
+    def _import_target(self):
+        from packages.shared.ai_client import _resolve_video_model_id
+
+        return _resolve_video_model_id
+
+    def test_none_uses_default(self):
+        fn = self._import_target()
+        with patch("packages.shared.ai_client.get_shared_settings") as ms:
+            ms.return_value = MagicMock(doubao_video_model="doubao-seedance-2-5-260628")
+            assert fn(None) == "doubao-seedance-2-5-260628"
+
+    def test_empty_uses_default(self):
+        fn = self._import_target()
+        with patch("packages.shared.ai_client.get_shared_settings") as ms:
+            ms.return_value = MagicMock(doubao_video_model="doubao-seedance-2-5-260628")
+            assert fn("   ") == "doubao-seedance-2-5-260628"
+
+    def test_doubao_prefix_passthrough(self):
+        fn = self._import_target()
+        assert fn("doubao-seedance-2-5-260628") == "doubao-seedance-2-5-260628"
+
+    def test_ep_prefix_passthrough(self):
+        fn = self._import_target()
+        assert fn("ep-20260721114705-b568m") == "ep-20260721114705-b568m"
+
+    def test_seedance_2_5_alias(self):
+        fn = self._import_target()
+        assert fn("seedance-2.5") == "doubao-seedance-2-5-260628"
+
+    def test_seedance_2_0_alias(self):
+        fn = self._import_target()
+        assert fn("seedance-2.0") == "doubao-seedance-2-0-250628"
+
+    def test_seedance_2_5_uppercase(self):
+        fn = self._import_target()
+        assert fn("Seedance-2.5") == "doubao-seedance-2-5-260628"
+
+    def test_seedance_dot_normalize(self):
+        fn = self._import_target()
+        # 传 "seedance-2.5" 带 dot 走 MAP.get 已命中；
+        # 构造带 seedance 但 key 变体的兜底场景
+        with patch(
+            "packages.shared.ai_client.VIRAL_VIDEO_MODEL_ID_MAP",
+            {
+                "seedance-2-5": "doubao-seedance-2-5-260628",
+            },
+        ):
+            assert fn("seedance-2.5") == "doubao-seedance-2-5-260628"
+
+    def test_unknown_seedance_falls_back_to_default_and_warns(self, caplog):
+        fn = self._import_target()
+        import logging
+
+        with patch("packages.shared.ai_client.get_shared_settings") as ms:
+            ms.return_value = MagicMock(doubao_video_model="doubao-seedance-2-5-260628")
+            with caplog.at_level(logging.WARNING, logger="shared.ai_client"):
+                # 任何不认识的别名
+                assert fn("seedance-9.9") == "doubao-seedance-2-5-260628"
+        assert any("未识别" in r.message for r in caplog.records if "ai_client" in r.name)

tests/unit/test_viral_video_p0.py

@@ -313,3 +313,133 @@ class TestResumeReadsImageAnalysis:
         # resume 本身应该调用 _run_render_pipeline
         resume_src = inspect.getsource(vv.resume_viral_video_pipeline)
         assert "_run_render_pipeline" in resume_src
+
+
+# ============ #2157 _replace_henjin_everywhere 递归替换单测 ============
+
+
+class TestReplaceHenjinEverywhere:
+    """覆盖 #2157 P1：递归替换 copy_result 中所有层级的'很近'→'最近'。"""
+
+    def _import(self):
+        from apps.worker.worker_app.tasks.viral_video import _replace_henjin_everywhere
+
+        return _replace_henjin_everywhere
+
+    def test_plain_string_no_henjin(self):
+        fn = self._import()
+        assert fn("最近好物推荐") == "最近好物推荐"
+        assert fn("") == ""
+        assert fn(None) is None
+        assert fn(123) == 123
+
+    def test_string_with_henjin(self):
+        fn = self._import()
+        assert fn("很近是不是总觉得颈肩发僵") == "最近是不是总觉得颈肩发僵"
+        # 多次出现
+        assert fn("很近很近都很近") == "最近最近都最近"
+
+    def test_list_recursive(self):
+        fn = self._import()
+        out = fn(["很近a", "b", ["很近c", "d"]])
+        assert out == ["最近a", "b", ["最近c", "d"]]
+
+    def test_dict_recursive_nested(self):
+        fn = self._import()
+        obj = {
+            "overview": {"theme": "很近颈肩", "title": "x"},
+            "scene_and_lighting": "很近才好用",
+            "voiceover_script": "很近是不是",
+            "final_copy": "很近好物",
+            "shots": [
+                {"scene_and_dialogue": "很近第一镜", "action_details": "很近动作", "audio_bgm": "很近音乐"},
+                {"nested": {"deep": "很近深层"}},
+            ],
+            "int_field": 42,
+        }
+        import json
+
+        out = fn(obj)
+        assert "很近" not in json.dumps(out, ensure_ascii=False)
+        assert out["overview"]["theme"] == "最近颈肩"
+        assert out["shots"][0]["scene_and_dialogue"] == "最近第一镜"
+        assert out["shots"][1]["nested"]["deep"] == "最近深层"
+        assert out["int_field"] == 42
+
+
+class TestMarkFailedAndNotifySessionFallback:
+    """#2157 P1：_mark_failed_and_notify 在原session失效时fallback到新SessionLocal。"""
+
+    def test_fallback_to_new_session_when_original_save_raises(self, tmp_path):
+        from unittest.mock import MagicMock, patch
+
+        from apps.worker.worker_app.tasks import viral_video as vv
+
+        job = MagicMock()
+        job.is_terminal = False
+        job.mark_failed = MagicMock()
+
+        # 原 session 保存抛异常
+        orig_session = MagicMock()
+        orig_repo = MagicMock()
+
+        def _raise(*a, **kw):
+            raise RuntimeError("session in rollback")
+
+        # 第一次调用_save_job抛异常，触发fallback
+        with patch.object(vv, "_save_job", side_effect=_raise):
+            fake_ssn = MagicMock()
+            fake_repo = MagicMock()
+            fake_job_in_db = MagicMock()
+            fake_job_in_db.is_terminal = False
+            fake_repo.get.return_value = fake_job_in_db
+            with patch.object(vv, "SessionLocal", return_value=fake_ssn):
+                with patch.object(vv, "SQLAlchemyViralVideoJobRepository", return_value=fake_repo):
+                    with patch.object(vv, "_emit_progress") as mock_emit:
+                        vv._mark_failed_and_notify("job-1", orig_session, orig_repo, job, "boom", stage="render")
+        # 原session上mark_failed被调用过
+        job.mark_failed.assert_called()
+        # fallback路径：新session上repo.get(job-1)被调用，且新job被mark_failed并commit
+        fake_repo.get.assert_called_with("job-1")
+        fake_job_in_db.mark_failed.assert_called_with("boom")
+        fake_repo.update.assert_called_with(fake_job_in_db)
+        fake_ssn.commit.assert_called()
+        fake_ssn.close.assert_called()
+        mock_emit.assert_called_once()
+
+    def test_original_session_happy_path_no_fallback(self):
+        from unittest.mock import MagicMock, patch
+
+        from apps.worker.worker_app.tasks import viral_video as vv
+
+        job = MagicMock()
+        job.is_terminal = False
+        session = MagicMock()
+        repo = MagicMock()
+        with patch.object(vv, "_save_job") as mock_save:
+            with patch.object(vv, "_emit_progress") as mock_emit:
+                # 不mock SessionLocal，如果fallback被错误触发会抛AttributeError
+                vv._mark_failed_and_notify("job-2", session, repo, job, "err", stage="copy")
+        job.mark_failed.assert_called_with("err")
+        mock_save.assert_called()
+        mock_emit.assert_called_once()
+
+    def test_terminal_job_not_marked(self):
+        from unittest.mock import MagicMock, patch
+
+        from apps.worker.worker_app.tasks import viral_video as vv
+
+        job = MagicMock()
+        job.is_terminal = True  # 已终态
+        session = MagicMock()
+        repo = MagicMock()
+        with patch.object(vv, "_save_job") as mock_save:
+            with patch.object(vv, "_emit_progress"):
+                fake_ssn = MagicMock()
+                with patch.object(vv, "SessionLocal", return_value=fake_ssn):
+                    with patch.object(vv, "SQLAlchemyViralVideoJobRepository") as mock_repo_cls:
+                        vv._mark_failed_and_notify("job-3", session, repo, job, "x")
+        # 终态job不调用mark_failed
+        job.mark_failed.assert_not_called()
+        # 且因 job 已终态，_save_job 也不应被调用（marked=False 才fallback；但此处 job 非 None 且 is_terminal=True，marked 保持 False 进入 fallback）
+        # fallback路径会重新打开session，get到的job也是终态，不会update