测试视频复刻

core/backend/apps/ai/services.py

@@ -901,13 +901,17 @@ NO_EMBEDDED_CAPTIONS_REQUIREMENT = (
     "只保留商品包装上参考图中原有的真实物理文字与标识。旁白和环境音可以保留，但不要把声音转成画面文字。"
 )
 
-# 故事板图会作为 Seedance 的参考图。导演信息应由页面从结构化脚本渲染，
-# 不能烧进参考图，否则边框、时间标签和旁白文字会被视频模型误当成待保留的画面元素。
-STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT = (
-    "【视频关键帧硬性规则】这张图将直接作为视频生成参考图：只生成一张干净、全幅、写实的 9:16 关键画面，"
-    "画面必须完整呈现本段最关键的角色、商品、场景、动作和空间关系。"
-    "禁止拼图、多宫格、故事板边框、时间轴、机位标注、旁白文字、标题、价格贴片、字幕、UI 或任何叠加文字；"
-    "只保留商品包装参考图中原有的真实物理文字与标识。"
+# 故事板是创作者确认镜头节奏的导演稿，因此必须由 image-2 直接生成完整排版图；
+# 视频提示词会明确要求只提取其中的角色、商品、场景、动作和节奏，忽略版式文字。
+STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT = (
+    "【导演故事板成图硬性要求】输出必须是一张完整、正式、可交付的电商短视频导演故事板图片，而不是无说明的拼图或单张海报。"
+    "整体为竖版 9:16，采用清爽专业的影视分镜排版：顶部 1 行标题区，标出“导演故事板”、本段时长和画幅；"
+    "主体严格按照本段脚本的秒级分镜，优先整理为 3 个连续时间段（例如 0–5s、5–10s、10–15s；若脚本时长不同则按实际时间划分）。"
+    "每个时间段必须包含三部分：左侧约 22% 宽的浅色信息栏，清晰写时间、景别、机位/运镜和简短动作；右侧约 78% 的写实主画面；"
+    "画面底部横条使用扬声器图标与“口播/旁白：”呈现该时间段对应旁白。各段按时间自上而下排列，边框、间距、对齐统一。"
+    "右侧画面要真实展现该秒级分镜的不同关键动作，人物脸、服装、商品外观、配色和场景连续一致；商品用法必须正确，包装真实文字和 Logo 不得改造。"
+    "故事板中的中文标注必须工整、简洁、可读，只写脚本已有的时间、机位、动作与旁白，不虚构价格、功效、活动或品牌信息。"
+    "禁止只给一张静态人物图、禁止无文字的四格拼图、禁止把三段画面重复成同一姿势。"
 )
 
 
@@ -2720,7 +2724,7 @@ def build_storyboard_frame_prompt(project, segment, extra_prompt: str = "") -> s
     )
     cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
     return _apply_storyboard_output_ratio(
-        f"{cleaned}\n{STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT}", project
+        f"{cleaned}\n{STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT}", project
     )
 
 
@@ -2748,6 +2752,8 @@ def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt:
     default = (
         "{设定}{分镜}【脚本】{脚本}\n"
         "严格按照脚本里的秒级分镜切换景别和动作，商品用法必须真实，不要诡异姿势或错误容器。"
+        "如果参考图是带时间、机位、旁白标注的导演故事板，只提取其中的角色、商品、场景、动作和时间顺序；"
+        "不要把边框、标题、时间码、扬声器图标或任何故事板文字生成进视频。"
         "电商带货短视频，商品露出清晰，节奏有转化感。不要字幕，不要背景音乐，但是要有音效，逼真的音效。"
     )
     rendered = render_prompt(
@@ -2933,7 +2939,7 @@ def build_storyboard_frame_prompt_refs(project, segment, refs: list[dict], extra
     )
     cleaned = "\n".join(line for line in rendered.split("\n") if line.strip())
     return _apply_storyboard_output_ratio(
-        f"{cleaned}\n{STORYBOARD_VIDEO_KEYFRAME_REQUIREMENT}", project
+        f"{cleaned}\n{STORYBOARD_DIRECTOR_LAYOUT_REQUIREMENT}", project
     )
 
 

core/backend/apps/ai/test_video_replace.py

@@ -2,6 +2,8 @@
 
 运行:DB_ENGINE=sqlite python manage.py test apps.ai.test_video_replace --settings=airshelf.settings.test
 """
+import tempfile
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
@@ -170,6 +172,95 @@ class SubmitVideoReplaceTests(TestCase):
         self.assertNotIn("reference_video", roles)     # 参考视频不再发给火山
         self.assertNotIn("video_url", [item.get("type") for item in content])
 
+    def _fake_source_file(self):
+        """_FileFromPath 会 stat 真实文件,给个空的临时 mp4 顶替 TOS 下载结果。"""
+        tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
+        tmp.write(b"\x00")
+        tmp.close()
+        self.addCleanup(lambda: Path(tmp.name).unlink(missing_ok=True))
+        return tmp.name
+
+    def test_digest_asset_video_runs_end_to_end_on_replace_task(self):
+        """回归:复刻任务的 model_config 是 Seedance(视频),拆解若走
+        execute_routed_text_request 会撞「AITask.model_config 必须保持为主模型」直接炸。
+        这条用例不 mock digest_asset_video 本体,只挡住 TOS/ffmpeg/模型,确保真身跑得通。"""
+        from apps.ai import video_digest
+
+        task = self._submit()
+        text_model = ModelConfig.objects.filter(capability="text", status="active").first()
+        self.assertIsNotNone(text_model)
+        self.assertNotEqual(text_model.id, task.model_config_id)  # 就是要两者不同
+        with patch.object(video_digest, "_download_source_to_temp", return_value=self._fake_source_file()), \
+             patch.object(
+                 video_digest, "digest_input_from_upload",
+                 return_value=(None, [], 8.0, {"width": 720, "height": 1280, "file_name": "参考.mp4"}),
+             ), \
+             patch.object(video_digest, "resolve_digest_model_config", return_value=text_model), \
+             patch("apps.ai.services._collect_extract_text", return_value=(DIGEST_SAMPLE, {})):
+            run_replace_digest(task)
+        task.refresh_from_db()
+        self.assertEqual(task.status, AITask.Status.SUBMITTED)
+        self.assertIn("【镜头 01】", task.request_payload["prompt"])
+        self.assertIn("@目标商品", task.request_payload["prompt"])
+
+    def test_replace_digest_prompt_matches_standalone(self):
+        """复刻内的提炼必须和「提炼提示词」页一字不差:同一份 SKILL.md、同一段引导语、
+        同样的 temperature / max_tokens,且不许掺商品信息。两处不一致 → 用户在提炼页
+        看到的分镜稿和复刻实际用的对不上,排查会乱。"""
+        from apps.ai import video_digest
+
+        task = self._submit()
+        text_model = ModelConfig.objects.filter(capability="text", status="active").first()
+        seen = {}
+
+        def _capture(provider, model_config, messages, **kwargs):
+            seen["messages"] = messages
+            seen["kwargs"] = kwargs
+            return (DIGEST_SAMPLE, {})
+
+        with patch.object(video_digest, "_download_source_to_temp", return_value=self._fake_source_file()), \
+             patch.object(
+                 video_digest, "digest_input_from_upload",
+                 return_value=(None, [], 8.0, {"width": 720, "height": 1280, "file_name": "参考.mp4"}),
+             ), \
+             patch.object(video_digest, "resolve_digest_model_config", return_value=text_model), \
+             patch("apps.ai.services._collect_extract_text", side_effect=_capture):
+            run_replace_digest(task)
+
+        standalone = video_digest.build_digest_messages(
+            [], 8.0, product_hint="", video=None,
+            aspect_ratio="9:16 竖屏", file_title="参考",
+        )
+        self.assertEqual(seen["messages"], standalone)
+        self.assertEqual(seen["messages"][0]["content"], video_digest.load_digest_skill())
+        self.assertEqual(seen["kwargs"].get("temperature"), 0.4)
+        self.assertEqual(
+            seen["kwargs"].get("extra_body"), {"max_tokens": video_digest.DIGEST_MAX_TOKENS}
+        )
+        # 提炼阶段绝不能掺进商品名,否则拆出来的就不是「这条参考视频原本的样子」
+        self.assertNotIn("净颜精华", str(seen["messages"]))
+
+    def test_digest_retries_once_before_giving_up(self):
+        """Gemini 偶尔吐废稿。复刻失败要用户从头重选素材,所以同模型再试一次。"""
+        from apps.ai import video_digest
+
+        task = self._submit()
+        text_model = ModelConfig.objects.filter(capability="text", status="active").first()
+        with patch.object(video_digest, "_download_source_to_temp", return_value=self._fake_source_file()), \
+             patch.object(
+                 video_digest, "digest_input_from_upload",
+                 return_value=(None, [], 8.0, {"width": 720, "height": 1280, "file_name": "参考.mp4"}),
+             ), \
+             patch.object(video_digest, "resolve_digest_model_config", return_value=text_model), \
+             patch(
+                 "apps.ai.services._collect_extract_text",
+                 side_effect=[("", {}), (DIGEST_SAMPLE, {})],
+             ) as call:
+            run_replace_digest(task)
+        self.assertEqual(call.call_count, 2)          # 第一次空文废稿,第二次成功
+        task.refresh_from_db()
+        self.assertEqual(task.status, AITask.Status.SUBMITTED)
+
     def test_digest_failure_fails_task_without_charging(self):
         from apps.ai.video_digest import VideoDigestError
 
@@ -182,9 +273,86 @@ class SubmitVideoReplaceTests(TestCase):
         task.refresh_from_db()
         self.assertEqual(task.status, AITask.Status.FAILED)
         self.assertIn("不完整", task.error_message or "")
+        # 模型抖动不能报成「内容未通过生成审核」,那会让用户去白换素材
+        self.assertEqual(task.error_code, "processing_failed")
         self.assertFalse(self.provider.create_video_task.called)
         self.assertEqual(CreditReservation.objects.filter(task=task).count(), 0)
 
+    def test_missing_digest_model_is_rejected_at_submit(self):
+        """提炼模型没配好要当场 400,不能建了任务、等一轮提炼才告诉用户。"""
+
+
+        before = AITask.objects.count()
+        with patch("apps.ai.video_digest.resolve_digest_model_config", return_value=None):
+            with self.assertRaisesMessage(ValueError, "视频提炼模型未配置"):
+                self._submit()
+        self.assertEqual(AITask.objects.count(), before)
+        self.assertFalse(self.provider.create_video_task.called)
+
+
+    def _make_triview(self):
+        """商品库的三视图是 standalone Asset:metadata.view=three_view + product_id。"""
+        tri = _asset(self.team, self.user, name="净颜精华·商品三视图", preview="http://tos/triview.png")
+        tri.metadata = {"product_id": str(self.product.id), "view": "three_view"}
+        tri.save(update_fields=["metadata"])
+        return tri
+
+    def _run_digest(self, task):
+        from apps.ai import video_digest
+
+        text_model = ModelConfig.objects.filter(capability="text", status="active").first()
+        with patch.object(video_digest, "_download_source_to_temp", return_value=self._fake_source_file()), \
+             patch.object(
+                 video_digest, "digest_input_from_upload",
+                 return_value=(None, [], 8.0, {"width": 720, "height": 1280, "file_name": "参考.mp4"}),
+             ), \
+             patch.object(video_digest, "resolve_digest_model_config", return_value=text_model), \
+             patch("apps.ai.services._collect_extract_text", return_value=(DIGEST_SAMPLE, {})):
+            run_replace_digest(task)
+        task.refresh_from_db()
+        return task
+
+    def test_product_triview_is_sent_with_the_video(self):
+        """商品库选商品时,三视图要跟着一起发给火山,并在提示词里被点名。"""
+        tri = self._make_triview()
+        task = self._submit()
+        labels = [ref["label"] for ref in task.request_payload["references"]]
+        self.assertIn("目标商品", labels)
+        self.assertIn("目标商品三视图", labels)
+
+        task = self._run_digest(task)
+        self.assertEqual(task.status, AITask.Status.SUBMITTED)
+        prompt = task.request_payload["prompt"]
+        self.assertIn("@目标商品三视图", prompt)
+        self.assertIn("白底多角度图", prompt)
+
+        content = self.provider.create_video_task.call_args.kwargs.get("content_items") or []
+        image_items = [item for item in content if item.get("type") == "image_url"]
+        self.assertEqual(len(image_items), 2)   # 商品实拍图 + 三视图
+        api_prompt = task.request_payload.get("api_prompt") or ""
+        self.assertNotIn("@目标商品", api_prompt)  # @label 已换成火山认的「图片N」指代
+        del tri
+
+    def test_prompt_skips_triview_note_when_product_has_none(self):
+        """没有三视图就别在提示词里提它,免得模型去找一张不存在的图。"""
+        task = self._run_digest(self._submit())
+        prompt = task.request_payload["prompt"]
+        self.assertNotIn("三视图", prompt)
+        self.assertIn("@目标商品", prompt)
+
+    def test_triview_yields_budget_so_refs_never_exceed_nine(self):
+        """三视图占一张名额:商品图要让位,否则火山那边超 9 张直接拒。"""
+        from apps.ai.video_replace import MAX_IMAGES
+
+        self._make_triview()
+        for i in range(MAX_IMAGES + 3):
+            extra = _asset(self.team, self.user, name=f"图{i}.png", preview=f"http://tos/p{i}.png")
+            ProductImage.objects.create(product=self.product, asset=extra, sort_order=i + 1)
+        task = self._submit()
+        refs = task.request_payload["references"]
+        self.assertEqual(len(refs), MAX_IMAGES)
+        self.assertEqual(refs[-1]["label"], "目标商品三视图")
+
     def test_compact_digest_trims_human_only_sections(self):
         compact = compact_digest_for_video(DIGEST_SAMPLE)
         self.assertIn("【镜头 02】", compact)

core/backend/apps/ai/video_digest.py

@@ -48,6 +48,8 @@ MAX_FRAMES = 36
 FRAME_WIDTH = 768
 FRAME_QUALITY = 3
 DIGEST_MAX_TOKENS = 12288
+# 视频复刻内联拆解的同模型重试次数(Gemini 偶尔吐废稿;复刻失败要用户从头重选素材,值得多试一次)
+DIGEST_MAX_ATTEMPTS = 2
 _SHOT_MARK = re.compile(r"【(?:镜头\s*\d+|第\s*\d+\s*镜)】")
 
 _FFMPEG_TIMEOUT = 60
@@ -728,12 +730,12 @@ def _store_raw_source(*, team, path: str, suffix: str) -> tuple[str, str]:
     return stored.object_key, storage.public_url(object_key=stored.object_key)
 
 
-def _download_source_to_temp(object_key: str, suffix: str) -> str:
+def _download_source_to_temp(object_key: str, suffix: str, *, bucket: str = "") -> str:
     from apps.assets.storage import TosStorage
 
     ext = suffix if str(suffix).startswith(".") else f".{suffix or 'mp4'}"
     storage = TosStorage()
-    body = storage.client.get_object(Bucket=storage.bucket, Key=object_key)["Body"].read()
+    body = storage.client.get_object(Bucket=bucket or storage.bucket, Key=object_key)["Body"].read()
     tmp = tempfile.NamedTemporaryFile(suffix=ext.lower(), delete=False)
     try:
         tmp.write(body)
@@ -1357,7 +1359,7 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
     复刻本来就是一次收费,拆解是它的内部工序。模型调用的审计仍记在传入的复刻
     task 上(AIModelAttempt),出问题查得到。
     """
-    from apps.ai.services import execute_routed_text_request
+    from apps.ai.services import _collect_extract_text, get_text_provider
 
     primary = asset.files.filter(is_primary=True).first() or asset.files.first()
     if primary is None or not primary.object_key:
@@ -1372,7 +1374,7 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
 
     local_path = ""
     try:
-        local_path = _download_source_to_temp(primary.object_key, suffix)
+        local_path = _download_source_to_temp(primary.object_key, suffix, bucket=primary.bucket or "")
         base_name = (asset.name or "参考视频").rsplit(".", 1)[0]
         upload = _FileFromPath(local_path, f"{base_name}{suffix}")
         video, frames, duration, extras = digest_input_from_upload(upload)
@@ -1385,6 +1387,11 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
             duration,
             len(frames),
         )
+        # ⚠️ 这里必须和「提炼提示词」页(_digest_video)构造出完全相同的 messages:
+        # 同一份 SKILL.md 作 system、同一段 user 引导语、同样的 temperature / max_tokens。
+        # 绝对不要在这里追加商品信息(product_hint 保持默认空)——两处提炼稿一旦不同,
+        # 用户在提炼页看到的分镜和复刻实际用的分镜就对不上,排查会乱。
+        # test_video_replace.test_replace_digest_prompt_matches_standalone 会守住这一点。
         messages = build_digest_messages(
             frames,
             duration,
@@ -1392,28 +1399,34 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
             aspect_ratio=ratio_label(int(extras.get("width") or 0), int(extras.get("height") or 0)),
             file_title=title_from_filename(str(extras.get("file_name") or "")),
         )
-        routed = execute_routed_text_request(
-            task=task,
-            primary_model=model_config,
-            messages=messages,
-            streaming=True,
-            structured_output=False,
-            business_operation="video_digest",
-            temperature=0.4,
-            validate_text=lambda text: validate_digest_text(
-                text, duration=duration, frame_count=len(frames) or (24 if video else 0)
-            ),
-            extra_body={"max_tokens": DIGEST_MAX_TOKENS},
-            request_summary={
-                "duration_seconds": round(duration, 2),
-                "frame_count": len(frames),
-                "input": "native_video" if video is not None else "frames",
-                "for": "video_replace",
-            },
-            allow_retry=False,
-            allow_fallback=False,
-        )
-        _text, _response, digest = routed.value
+        # 这里不能走 execute_routed_text_request:它硬性要求 task.model_config 就是本次主模型,
+        # 而复刻任务的 model_config 是 Seedance(视频模型),传进去必抛
+        # 「AITask.model_config 必须保持为用户选择或系统默认的主模型」。
+        # 改为直连脚本/提炼同一条流式通道,并自己做「同模型重试」——绝不 fallback 到别的
+        # 文本模型,它们看不了视频,换过去必然废稿。
+        provider = get_text_provider(model_config)
+        expected_frames = len(frames) or (24 if video else 0)
+        digest = ""
+        last_error: Exception | None = None
+        for attempt in range(1, DIGEST_MAX_ATTEMPTS + 1):
+            try:
+                text, _payload = _collect_extract_text(
+                    provider,
+                    model_config,
+                    messages,
+                    temperature=0.4,
+                    extra_body={"max_tokens": DIGEST_MAX_TOKENS},
+                )
+                digest = validate_digest_text(text, duration=duration, frame_count=expected_frames)
+                break
+            except Exception as exc:  # noqa: BLE001 — 空文/废稿/网络抖动都值得再来一次
+                last_error = exc
+                logger.warning(
+                    "replace digest attempt %s/%s failed for task %s: %s",
+                    attempt, DIGEST_MAX_ATTEMPTS, task.id, exc,
+                )
+        if not digest:
+            raise VideoDigestError(f"参考视频拆解失败：{last_error}")
         meta = {
             "digest_model": model_config.name,
             "digest_input": "native_video" if video is not None else "frames",

core/backend/apps/ai/video_replace.py

@@ -40,6 +40,9 @@ FEATURE = "video_replace"
 REPLACE_MODES = {"product", "character"}
 MAX_IMAGES = 9
 LEGACY_PROMPT_PREFIX = "[视频复刻]"
+# 商品三视图在商品库是 standalone Asset(metadata.view=three_view + product_id),
+# 和项目内的 BaseAssetGroup 是两套存储,这里直接取商品库那份。
+TRIVIEW_LABEL = "目标商品三视图"
 REVIEW_UNAVAILABLE = "素材审核服务暂不可用，请稍后重试"
 REVIEW_FAILED = "参考素材未通过真人合规审核，请更换视频或图片后重试"
 REVIEW_SUBMIT_FAILED = "素材提交审核失败，请稍后重试"
@@ -57,10 +60,17 @@ PRODUCT_DIGEST_TAIL = (
     "1. 严格按上面分镜稿的镜头顺序、时间分配、景别、机位、运镜和剪辑节奏还原全片，不要自行增删镜头。\n"
     "2. 分镜稿里出现的原商品，全部替换成 @目标商品；商品的外形、颜色、材质、包装和标识必须与参考图完全一致，"
     "不要改造型、不要改配色、不要凭空补细节。\n"
+    "{triview}"
     "3. 人物、场景、光线、色调、动作和表情按分镜稿保持不变，只换商品。\n"
     "4. 台词/旁白按分镜稿逐字念出；原文提到旧商品名称的地方，改说「{subject}」。\n"
     "5. 分镜稿里的「字幕」栏只是对原片的记录，不要把这些文字画到画面上。"
 )
+# 商品库里有三视图时才追加这一条(没有就不提,免得模型去找一张不存在的图)。
+PRODUCT_TRIVIEW_NOTE = (
+    "   · @目标商品三视图 是这件商品的白底多角度图，用它确认商品的立体结构、比例和各个面的细节；"
+    "镜头转到任何角度，商品都要和三视图对得上。\n"
+)
+
 # 拆解期间的占位提示词。真正的提示词在 worker 拆完后回写。
 DIGEST_PENDING_PROMPT = "正在提炼参考视频的分镜稿…"
 
@@ -97,7 +107,7 @@ def compact_digest_for_video(digest_text: str) -> str:
     return "\n".join(line for line in lines if line.strip()).strip()
 
 
-def build_product_replace_prompt(digest_text: str, subject_name: str) -> str:
+def build_product_replace_prompt(digest_text: str, subject_name: str, *, has_triview: bool = False) -> str:
     """分镜稿 + 商品替换要求 → 交给 Seedance 的完整提示词。
 
     @目标商品 必须与 references 里第一张商品图的 label 对上,否则 build_content_items
@@ -109,7 +119,10 @@ def build_product_replace_prompt(digest_text: str, subject_name: str) -> str:
     if not body:
         raise ValueError("参考视频拆解结果为空，请重试")
     subject = (subject_name or "").strip() or "目标商品"
-    tail = PRODUCT_DIGEST_TAIL.format(subject=subject)
+    tail = PRODUCT_DIGEST_TAIL.format(
+        subject=subject,
+        triview=PRODUCT_TRIVIEW_NOTE if has_triview else "",
+    )
     return enforce_no_embedded_captions(f"{PRODUCT_DIGEST_HEAD}\n\n{body}\n\n{tail}")
 
 
@@ -210,7 +223,17 @@ def submit_video_replace(*, team, user, params: dict):
     }
 
     if replace_mode == "product":
-        # 商品复刻:参考视频只喂给提炼模型,不进火山 references(所以也不用送火山审核)。
+        # 商品复刻:参考视频只喂给提炼模型,不进火山 references(所以参考视频本身不再送火山审核)。
+        # 两个「一定会失败」的前提在这里就查掉,别让用户白等一轮提炼才看到报错:
+        from .video_digest import resolve_digest_model_config
+
+        if resolve_digest_model_config() is None:
+            raise ValueError("视频提炼模型未配置，请联系管理员")
+        # 商品图仍要过审才能当生成参考。提前送审 → 审核和提炼并行跑,明确不过审的直接 400。
+        review_state = _ensure_replace_refs_reviewed(team, image_refs)
+        if review_state == "failed":
+            raise ValueError(REVIEW_FAILED)
+        image_refs = _refresh_replace_refs(team, image_refs)
         # 先秒回一个 CREATED 任务,拆解这种慢活(Gemini 半分钟起)交给 worker。
         extra.update({
             "digest_source_asset_id": str(video.id),
@@ -313,18 +336,23 @@ def run_replace_digest(task) -> None:
             task.team, uuid.UUID(str(payload.get("digest_source_asset_id") or "")), "参考视频"
         )
     except (ValueError, TypeError):
-        _fail_reviewing_task(task, "参考视频已失效，请重新上传")
+        _fail_reviewing_task(task, "参考视频已失效，请重新上传", error_code="asset_unavailable")
         return
 
     try:
         digest, meta = digest_asset_video(asset=asset, task=task)
-        prompt = build_product_replace_prompt(digest, str(payload.get("subject_name") or ""))
+        has_triview = any(
+            (ref or {}).get("label") == TRIVIEW_LABEL for ref in (payload.get("references") or [])
+        )
+        prompt = build_product_replace_prompt(
+            digest, str(payload.get("subject_name") or ""), has_triview=has_triview
+        )
     except (VideoDigestError, ValueError) as exc:
-        _fail_reviewing_task(task, str(exc))
+        _fail_reviewing_task(task, str(exc), error_code="processing_failed")
         return
     except Exception as exc:  # noqa: BLE001 — 拆解任何异常都要把任务收尾,别留 CREATED 僵尸
         logger.exception("video replace digest failed for %s", task.id)
-        _fail_reviewing_task(task, f"参考视频拆解失败：{exc}")
+        _fail_reviewing_task(task, f"参考视频拆解失败：{exc}", error_code="processing_failed")
         return
 
     with transaction.atomic():
@@ -354,10 +382,20 @@ def _legacy_replace_mode(prompt: str) -> str:
     return "character" if prompt.startswith("[视频复刻·角色]") else "product"
 
 
-def _fail_reviewing_task(task, message: str):
+def _fail_reviewing_task(task, message: str, *, error_code: str = ""):
+    """收尾一个还没提交火山的任务。
+
+    ``_fail_pending_free_video`` 把错误码写死成 content_rejected —— 那是审核不通过的语义,
+    前端会渲染成「内容未通过生成审核」且不可重试。拆解失败/素材丢失不是那回事,
+    这里按实际原因改码,否则模型抖一下会被误报成合规问题,用户白白去换素材。
+    """
     from .free_video import _fail_pending_free_video
 
-    return _fail_pending_free_video(task, message)
+    task = _fail_pending_free_video(task, message)
+    if error_code and task.status == AITask.Status.FAILED and task.error_code != error_code:
+        task.error_code = error_code
+        task.save(update_fields=["error_code", "updated_at"])
+    return task
 
 
 def _enqueue_replace_review_poll(task):
@@ -701,6 +739,21 @@ def _library_image_ref(asset: Asset, *, team, label: str) -> dict:
     }
 
 
+def _product_triview_asset(team, product_id: uuid.UUID):
+    """商品库里这个商品的三视图(白底多角度单张 16:9)。没有就返回 None,不挡生成。"""
+    return (
+        Asset.objects.filter(
+            team=team,
+            is_deleted=False,
+            purged_at__isnull=True,
+            metadata__product_id=str(product_id),
+            metadata__view="three_view",
+        )
+        .order_by("-created_at")
+        .first()
+    )
+
+
 def _product_library_refs(team, product_id: uuid.UUID) -> tuple[str, list, str]:
     product = (
         Product.objects.filter(id=product_id, team=team, purged_at__isnull=True, status=Product.Status.ACTIVE)
@@ -710,21 +763,27 @@ def _product_library_refs(team, product_id: uuid.UUID) -> tuple[str, list, str]:
     )
     if product is None:
         raise ValueError("商品不存在或已被删除")
+    triview = _product_triview_asset(team, product_id)
+    # 三视图也占一张参考图的名额,商品图要给它让位,否则火山那边超 9 张直接拒。
+    image_budget = MAX_IMAGES - 1 if triview is not None else MAX_IMAGES
     assets = []
-    seen = set()
+    seen = {triview.id} if triview is not None else set()
     for image in product.images.all():
         asset = image.asset
         if asset is None or asset.id in seen or asset.is_deleted:
             continue
         seen.add(asset.id)
         assets.append(asset)
-        if len(assets) >= MAX_IMAGES:
+        if len(assets) >= image_budget:
             break
     if not assets and product.cover_asset_id and not product.cover_asset.is_deleted:
         assets.append(product.cover_asset)
-    if not assets:
+    if not assets and triview is None:
         raise ValueError("这个商品还没有可用图片")
     refs = [_library_image_ref(asset, team=team, label="目标商品" if index == 0 else f"目标商品{index + 1}") for index, asset in enumerate(assets)]
+    if triview is not None:
+        # 放最后:@目标商品 仍指向第一张实拍图,三视图作为「各面长什么样」的补充证据。
+        refs.append(_library_image_ref(triview, team=team, label=TRIVIEW_LABEL))
     return product.title, refs, "library"