优化视频复刻和商品内容大小

2026-08-28 18:42:28 +08:00
parent a618f02315
commit 6b6146e595
20 changed files with 1557 additions and 382 deletions
@@ -811,15 +811,17 @@ def _notify_failure(task: AITask, *, raw: str, hint: str) -> None:
    )


-def _store_free_video_media(*, task: AITask, media: str) -> Asset:
-    """下载火山结果 → 转存 TOS(火山原始 URL 仅 7 天有效)→ 建 Asset(FREE_CREATE,自动入库)
-    + ffmpeg 抽首帧封面挂同 Asset 非主文件。"""
-    fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
-    if "video" not in content_type:
-        content_type = "video/mp4"
-    # 先取字节再上传:boto3 upload_fileobj 完成后会 close 掉 BytesIO,之后 getvalue() 直接抛
-    # "I/O operation on closed file",封面抽帧就永远做不了(实测踩坑)。
-    video_bytes = fileobj.getvalue() if isinstance(fileobj, BytesIO) else b""
+def _store_free_video_media(*, task: AITask, media: str = "", video_bytes: bytes | None = None) -> Asset:
+    """下载火山结果(或已拼好的字节) → 转存 TOS → 建 Asset + ffmpeg 抽首帧封面。"""
+    content_type = "video/mp4"
+    if video_bytes is None:
+        fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
+        if "video" not in content_type:
+            content_type = "video/mp4"
+        # 先取字节再上传:boto3 upload_fileobj 完成后会 close 掉 BytesIO,之后 getvalue() 直接抛
+        # "I/O operation on closed file",封面抽帧就永远做不了(实测踩坑)。
+        video_bytes = fileobj.getvalue() if isinstance(fileobj, BytesIO) else b""
+    fileobj = BytesIO(video_bytes)
    asset_id = uuid.uuid4()
    object_key = f"teams/{task.team_id}/free-create/{asset_id}.mp4"
    stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
@@ -43,8 +43,8 @@ def run_video_digest_task(self, task_id: str) -> str:

@app.task(bind=True, max_retries=0)
 def run_video_replace_digest_task(self, task_id: str) -> str:
-    """商品复刻的第一道工序(参考视频 → 分镜稿,Gemini 半分钟起)在 worker 内跑。
-    失败自己把任务收成 FAILED(见 run_replace_digest),故 max_retries=0,不向上抛重试。"""
+    """Worker:视频复刻第一道工序(参考视频 → 分镜稿,Gemini 半分钟起)在 worker 内跑。
+    商品和角色共用。失败自己把任务收成 FAILED(见 run_replace_digest),故 max_retries=0,不向上抛重试。"""
    from apps.ai.models import AITask
    from apps.ai.video_replace import run_replace_digest

@@ -97,7 +97,7 @@ def poll_free_video_task(self, task_id: str, attempt: int = 0) -> str:
    if task is None:
        return task_id
    try:
-        if is_video_replace_task(task) and task.status == AITask.Status.CREATED:
+        if is_video_replace_task(task):
            task = advance_video_replace(task)
        else:
            task = finalize_free_video(task=task)
@@ -110,9 +110,11 @@ def poll_free_video_task(self, task_id: str, attempt: int = 0) -> str:

    if getattr(dj_settings, "CELERY_TASK_ALWAYS_EAGER", False):
        return task_id
-    if task.status == AITask.Status.CREATED and attempt < 60:
+    # 复刻逐镜生成可能跑 8 段 × 数分钟,60 次(约 30 分钟)不够,拉到约 90 分钟。
+    poll_limit = 180 if is_video_replace_task(task) else 60
+    if task.status == AITask.Status.CREATED and attempt < poll_limit:
        poll_free_video_task.apply_async(args=[task_id, attempt + 1], countdown=8 if attempt < 12 else 30)
-    elif task.status in (AITask.Status.SUBMITTED, AITask.Status.POLLING) and attempt < 60:
+    elif task.status in (AITask.Status.SUBMITTED, AITask.Status.POLLING) and attempt < poll_limit:
        poll_free_video_task.apply_async(args=[task_id, attempt + 1], countdown=30)
    return task_id

@@ -20,6 +20,7 @@ from apps.ai.video_replace import (
    REVIEW_FAILED,
    REVIEW_UNAVAILABLE,
    advance_video_replace,
+    build_shot_plan,
    compact_digest_for_video,
    run_replace_digest,
    submit_video_replace,
@@ -72,6 +73,52 @@ DIGEST_SAMPLE = """片名：《晨间一杯》
 - 第 2 镜标签文字看不清，请校对品牌名。
 """

+DIGEST_FOUR = """片名：加长
+画面比例：9:16 竖屏
+整体风格：写实
+主线：四段卖点。
+
+【镜头 01】
+时间：00:00-00:05
+时长：5秒
+景别：中景
+机位：平视
+运镜：固定
+画面：第一镜开场。
+人物动作：举手。
+台词/旁白：第一句。
+
+【镜头 02】
+时间：00:05-00:10
+时长：5秒
+景别：近景
+机位：平视
+运镜：固定
+画面：第二镜展示。
+人物动作：转瓶。
+台词/旁白：第二句。
+
+【镜头 03】
+时间：00:10-00:15
+时长：5秒
+景别：特写
+机位：平视
+运镜：推近
+画面：第三镜特写。
+人物动作：按下泵头。
+台词/旁白：第三句。
+
+【镜头 04】
+时间：00:15-00:20
+时长：5秒
+景别：中近景
+机位：平视
+运镜：固定
+画面：第四镜收束。
+人物动作：微笑。
+台词/旁白：第四句。
+"""
+

 def _asset(team, user, *, kind=Asset.Type.IMAGE, name="素材", duration_ms=None, preview="http://tos/1.png", review_status="active", review_remote_id=None):
    asset = Asset.objects.create(
@@ -122,13 +169,13 @@ class SubmitVideoReplaceTests(TestCase):
            "model": STANDARD,
            "aspect_ratio": "9:16",
            "resolution": "480p",
-            "duration": 4,
+            "duration": 8,
        }
        params.update(over)
        return submit_video_replace(team=self.team, user=self.user, params=params)

-    def test_product_submit_waits_for_digest_and_keeps_video_out_of_refs(self):
-        """商品复刻第一步:秒回 CREATED,参考视频只留作拆解源,不进火山 references。"""
+    def test_product_submit_waits_for_digest_and_keeps_video_out_of_review_refs(self):
+        """商品复刻第一步:秒回 CREATED。审核列表只有商品图;原片快照另存,出片时再带上。"""
        task = self._submit()
        self.assertEqual(task.status, AITask.Status.CREATED)
        payload = task.request_payload
@@ -145,8 +192,8 @@ class SubmitVideoReplaceTests(TestCase):
        self.assertFalse(self.provider.create_video_task.called)
        self.assertEqual(CreditReservation.objects.filter(task=task).count(), 0)

-    def test_product_digest_becomes_prompt_and_only_images_go_to_ark(self):
-        """第二步:worker 拆完 → 分镜稿变提示词 → 只带商品图提交火山。"""
+    def test_product_digest_becomes_prompt_and_video_goes_to_ark(self):
+        """第二步:worker 拆完 → 按镜拆成生成计划 → 原片 + 商品图一起提交第一镜。"""
        task = self._submit()
        with patch(
            "apps.ai.video_digest.digest_asset_video",
@@ -158,19 +205,28 @@ class SubmitVideoReplaceTests(TestCase):
        payload = task.request_payload
        self.assertFalse(payload["digest_pending"])
        self.assertEqual(payload["digest_shots"], 2)
+        plan = payload["shot_plan"]
+        self.assertEqual(len(plan), 2)
+        self.assertEqual(plan[0]["seconds"], 4)
+        self.assertEqual(plan[1]["seconds"], 4)
        prompt = payload["prompt"]
-        self.assertIn("【镜头 01】", prompt)          # 分镜稿骨架进了提示词
-        self.assertIn("倒进玻璃杯", prompt)            # 画面描述保留
-        self.assertIn("@目标商品", prompt)             # 商品图引用锚点在
-        self.assertIn("净颜精华", prompt)              # 口播里的旧商品名改说自家商品
-        self.assertNotIn("【拆解存疑】", prompt)        # 给用户校对的段落不喂模型
-        self.assertNotIn("字幕：", prompt)             # 花字记录不喂,免得模型画上去
+        self.assertIn("@参考视频", prompt)             # 2.0 锁切镜的锚点
+        self.assertIn("第 1/", prompt)                 # 只喂当前这一镜
+        self.assertIn("倒进玻璃杯", prompt)            # 提炼原文画面保留
+        self.assertIn("台词/旁白", prompt)             # 不改写成「口播」
+        self.assertIn("@目标商品", prompt)
+        self.assertIn("净颜精华", prompt)
+        self.assertNotIn("【拆解存疑】", prompt)
+        self.assertNotIn("字幕：", prompt)
+        self.assertNotIn("转动瓶身", prompt)           # 第二镜的动作不能漏进第一镜
+        self.assertEqual(self.provider.create_video_task.call_count, 1)
+        self.assertEqual(self.provider.create_video_task.call_args.kwargs.get("duration"), 4)

        content = self.provider.create_video_task.call_args.kwargs.get("content_items") or []
        roles = [item.get("role") for item in content]
        self.assertIn("reference_image", roles)
-        self.assertNotIn("reference_video", roles)     # 参考视频不再发给火山
-        self.assertNotIn("video_url", [item.get("type") for item in content])
+        self.assertIn("reference_video", roles)
+        self.assertIn("video_url", [item.get("type") for item in content])

    def _retire(self, task):
        """把任务收成终态,好让单飞闸放行下一次提交。"""
@@ -206,7 +262,7 @@ class SubmitVideoReplaceTests(TestCase):
            run_replace_digest(task)
        task.refresh_from_db()
        self.assertEqual(task.status, AITask.Status.SUBMITTED)
-        self.assertIn("【镜头 01】", task.request_payload["prompt"])
+        self.assertIn("第 1/", task.request_payload["prompt"])
        self.assertIn("@目标商品", task.request_payload["prompt"])

    def test_replace_digest_prompt_matches_standalone(self):
@@ -381,7 +437,7 @@ class SubmitVideoReplaceTests(TestCase):
        tri.save(update_fields=["metadata"])
        return tri

-    def _run_digest(self, task):
+    def _run_digest(self, task, digest=DIGEST_SAMPLE):
        from apps.ai import video_digest

        text_model = ModelConfig.objects.filter(capability="text", status="active").first()
@@ -391,7 +447,7 @@ class SubmitVideoReplaceTests(TestCase):
                 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, {})):
+             patch("apps.ai.services._collect_extract_text", return_value=(digest, {})):
            run_replace_digest(task)
        task.refresh_from_db()
        return task
@@ -408,7 +464,6 @@ class SubmitVideoReplaceTests(TestCase):
        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"]
@@ -457,12 +512,100 @@ class SubmitVideoReplaceTests(TestCase):
        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)
-        self.assertIn("台词/旁白：", compact)
-        self.assertNotIn("字幕：", compact)
-        self.assertNotIn("【拆解存疑】", compact)
+    def test_digest_becomes_a_shot_table_for_the_video_model(self):
+        """分镜稿是给人校对的导演稿。喂给视频模型前要转成只讲「镜头怎么拍、人说什么」
+        的镜头表 —— 元信息和音轨描述留着只会稀释真正的画面指令。"""
+        table = compact_digest_for_video(DIGEST_SAMPLE)
+        # 镜头结构和画面指令留下
+        self.assertIn("镜头 01 | 00:00-00:04", table)
+        self.assertIn("镜头 02 | 00:04-00:08", table)
+        self.assertIn("画面：女生站在窗边", table)
+        self.assertIn("口播：每天早上我都要来一支旧牌精华。", table)
+        self.assertIn("镜头数：2 镜", table)
+        # 整片设定只留对画面有用的
+        self.assertIn("整体风格：写实、暖色调、清晨感", table)
+        self.assertNotIn("片名", table)
+        self.assertNotIn("视频类型", table)
+        self.assertNotIn("结构：", table)
+        # 非画面指令全部剔除
+        for noise in ("字幕：", "音效：", "背景音乐：", "备注：", "【拆解存疑】"):
+            self.assertNotIn(noise, table)
+        # 空值行不占篇幅("表情：不可见" 这种)
+        self.assertNotIn("不可见", table)
+
+    def test_unparseable_digest_falls_back_to_raw_text(self):
+        """模型没按 skill 格式吐时,宁可原文照传也不能把镜头信息弄丢。"""
+        messy = "这条片子讲了一个女生的早晨。\n她拿起瓶子。\n【拆解存疑】\n- 看不清"
+        out = compact_digest_for_video(messy)
+        self.assertIn("这条片子讲了一个女生的早晨。", out)
+        self.assertNotIn("【拆解存疑】", out)
+
+    def test_shot_plan_keeps_one_seedance_job_per_shot(self):
+        plan = build_shot_plan(DIGEST_SAMPLE, "净颜精华", output_seconds=8)
+        self.assertEqual([item["seconds"] for item in plan], [4, 4])
+        self.assertIn("第 1/", plan[0]["prompt"])
+        self.assertIn("@参考视频", plan[0]["prompt"])
+        self.assertIn("倒进玻璃杯", plan[0]["prompt"])
+        self.assertIn("台词/旁白", plan[0]["prompt"])
+        self.assertNotIn("转动瓶身", plan[0]["prompt"])
+        self.assertIn("第 2/", plan[1]["prompt"])
+        self.assertIn("@目标商品", plan[0]["prompt"])
+        self.assertNotIn("@目标角色", plan[0]["prompt"])
+
+    def test_character_shot_plan_swaps_person_not_product(self):
+        plan = build_shot_plan(DIGEST_SAMPLE, "薇薇", replace_mode="character", output_seconds=8)
+        self.assertIn("@目标角色", plan[0]["prompt"])
+        self.assertIn("@参考视频", plan[0]["prompt"])
+        self.assertIn("商品", plan[0]["prompt"])
+        self.assertNotIn("@目标商品", plan[0]["prompt"])
+        self.assertNotIn("净颜精华", plan[0]["prompt"])
+
+    def test_adjacent_sub_four_second_shots_merge(self):
+        digest = """【镜头 01】
+时长：2秒
+画面：第一下。
+【镜头 02】
+时长：2秒
+画面：第二下。
+"""
+        plan = build_shot_plan(digest, "精华", output_seconds=8)
+        self.assertEqual(len(plan), 1)
+        self.assertEqual(plan[0]["seconds"], 4)
+        self.assertIn("切点", plan[0]["prompt"])
+        self.assertIn("第一下", plan[0]["prompt"])
+        self.assertIn("第二下", plan[0]["prompt"])
+
+    def test_shots_are_generated_one_by_one_then_concatenated(self):
+        """2.5 不能一次塞完整镜头表。第一镜出完再提第二镜,全部完成才拼接。"""
+        self.provider.create_video_task.side_effect = [
+            _ark_create_response("ark-1"),
+            _ark_create_response("ark-2"),
+        ]
+        self.provider.poll_video_task.side_effect = [
+            {"status": "succeeded", "usage": {"total_tokens": 1000}},
+            {"status": "succeeded", "usage": {"total_tokens": 1200}},
+        ]
+        self.provider.extract_first_media_url.side_effect = ["http://ark/s1.mp4", "http://ark/s2.mp4"]
+        task = self._run_digest(self._submit())
+        self.assertEqual(task.provider_task_id, "ark-1")
+        self.assertEqual(self.provider.create_video_task.call_count, 1)
+
+        with patch("apps.ai.video_replace._concat_shot_media", return_value=b"concat-bytes") as concat, patch(
+            "apps.ai.free_video._store_free_video_media"
+        ) as store:
+            task = advance_video_replace(task)
+            self.assertEqual(task.provider_task_id, "ark-2")
+            self.assertEqual(self.provider.create_video_task.call_count, 2)
+            self.assertEqual(task.request_payload["shot_plan"][0]["status"], "succeeded")
+            self.assertFalse(store.called)
+
+            task = advance_video_replace(task)
+        self.assertEqual(task.status, AITask.Status.SUCCEEDED)
+        concat.assert_called_once()
+        self.assertEqual(concat.call_args[0][0], ["http://ark/s1.mp4", "http://ark/s2.mp4"])
+        store.assert_called_once()
+        self.assertEqual(store.call_args.kwargs.get("video_bytes"), b"concat-bytes")
+        self.assertEqual(task.request_payload.get("actual_tokens"), 2200)

    def test_character_library_and_temp_are_exclusive(self):
        portrait = _asset(self.team, self.user, name="模特.png", preview="http://tos/model.png")
@@ -472,6 +615,10 @@ class SubmitVideoReplaceTests(TestCase):
        task = self._submit(replace_mode="character", product_id="", model_id=str(model.id))
        self.assertEqual(task.request_payload["replace_mode"], "character")
        self.assertEqual(task.request_payload["subject_name"], "薇薇")
+        self.assertTrue(task.request_payload["digest_pending"])
+        labels = [item["label"] for item in task.request_payload["references"]]
+        self.assertNotIn("参考视频", labels)
+        self.assertIn("目标角色", labels)

    def test_both_modes_accept_up_to_30s(self):
        """商品和角色都走 Seedance 2.5(单次能出 30 秒),参考视频统一按 30 秒收口。"""
@@ -495,29 +642,31 @@ class SubmitVideoReplaceTests(TestCase):
            with self.assertRaisesMessage(ValueError, "不能超过 30 秒"):
                self._submit(replace_mode=mode, video_asset_id=str(too_long.id), **extra)

-    def test_long_source_asks_model_to_condense(self):
-        """60 秒参考稿 + 15 秒成片:必须让模型压缩改编,不然会照着拍到一半戛然而止。"""
+    def test_long_source_drops_trailing_shots(self):
+        """参考片比成片长很多时整镜省略尾镜,不要压缩画面。"""
        long_video = _asset(self.team, self.user, kind=Asset.Type.VIDEO, name="半分钟.mp4", duration_ms=28000)
-        # 显式要一段短成片:参考 28 秒、成片 10 秒,差得多才要求模型压缩改编
-        task = self._run_digest(self._submit(video_asset_id=str(long_video.id), duration=10))
-        prompt = task.request_payload["prompt"]
-        self.assertIn("压缩改编", prompt)
-        self.assertIn("28 秒", prompt)
-        self.assertIn("10 秒", prompt)
+        task = self._run_digest(self._submit(video_asset_id=str(long_video.id), duration=10), digest=DIGEST_FOUR)
+        plan = task.request_payload["shot_plan"]
+        self.assertEqual(len(plan), 2)
+        self.assertIn("第一镜开场", plan[0]["prompt"])
+        self.assertIn("第二镜展示", plan[1]["prompt"])
+        self.assertNotIn("第四镜收束", plan[0]["prompt"] + plan[1]["prompt"])

-    def test_short_source_skips_condense_note(self):
-        """参考视频和成片时长差不多时别啰嗦。"""
+    def test_short_source_keeps_all_shots(self):
+        """时长一致时保留全部镜头。"""
        task = self._run_digest(self._submit(duration=None))   # setUp 的参考视频 8 秒 → 成片也 8 秒
-        self.assertNotIn("压缩改编", task.request_payload["prompt"])
+        self.assertEqual(len(task.request_payload["shot_plan"]), 2)
+        self.assertNotIn("时长不一致时的取舍", task.request_payload["prompt"])
+        self.assertNotIn("省略次要镜头", task.request_payload["prompt"])

-    def test_prompt_always_pins_character_consistency(self):
-        """上传图里可能有人物(服饰模特图/真人手持图),要锁住人物一致;
-        没有人物时也要说清楚「别从商品图凭空套脸」。"""
+    def test_prompt_copies_source_video_instead_of_rewriting_a_script(self):
+        """切镜靠 @参考视频,提示词只钉「换商品」+ 这一镜原文,不要再扩写成导演长文。"""
        prompt = self._run_digest(self._submit()).request_payload["prompt"]
-        self.assertIn("自始至终是同一个人", prompt)
-        self.assertIn("不要中途换脸", prompt)
-        self.assertIn("参考图里如果有人物", prompt)
-        self.assertIn("不要从商品图上凭空套一张脸", prompt)
+        self.assertIn("@参考视频", prompt)
+        self.assertIn("人物", prompt)
+        self.assertIn("台词/旁白", prompt)
+        self.assertNotIn("必须原样保持", prompt)
+        self.assertNotIn("不要从商品图凭空套脸", prompt)

    def test_accepts_15s_video_with_probe_slack(self):
        from apps.ai.media_probe import REF_DURATION_MAX
@@ -658,8 +807,7 @@ class VideoReplaceApiTests(TestCase):


 class VideoReplaceReviewGateTests(TestCase):
-    """审核闸走角色复刻:只有它才把参考视频直传火山,需要真人合规审核。
-    商品复刻改成先提炼分镜稿,参考视频不进火山,自然也不再送审。"""
+    """审核闸只盯人物图/商品图。原片出片时直传火山锁切镜,不进人物素材库送审。"""

    def setUp(self):
        self.user = User.objects.create_user(username="vrrev", password="p")
@@ -704,54 +852,69 @@ class VideoReplaceReviewGateTests(TestCase):
    def test_unreviewed_assets_create_pending_task_without_charging(self):
        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
            "apps.assets.review.get_or_create_team_group"
-        ) as grp, patch("apps.assets.assets_client.create_asset", side_effect=["Asset-vid", "Asset-img"]) as create, patch(
+        ) as grp, patch("apps.assets.assets_client.create_asset", return_value="Asset-img") as create, patch(
            "apps.assets.assets_client.get_asset", return_value={"Status": "Processing"}
        ):
            grp.return_value = MagicMock(remote_group_id="Group-1")
            task = self._submit()
        self.assertEqual(task.status, AITask.Status.CREATED)
-        self.assertTrue((task.request_payload or {}).get("review_pending"))
+        self.assertTrue((task.request_payload or {}).get("digest_pending"))
        self.assertEqual(CreditReservation.objects.filter(task=task).count(), 0)
        self.assertFalse(self.provider.create_video_task.called)
        passed_types = [call.kwargs.get("asset_type") for call in create.call_args_list]
-        self.assertIn("Video", passed_types)
-        self.assertIn("Image", passed_types)
+        self.assertEqual(passed_types, ["Image"])
+        labels = [item["label"] for item in (task.request_payload or {}).get("references") or []]
+        self.assertNotIn("参考视频", labels)

    def test_advance_starts_generation_after_review_passes(self):
        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
            "apps.assets.review.get_or_create_team_group"
-        ) as grp, patch("apps.assets.assets_client.create_asset", side_effect=["Asset-vid", "Asset-img"]), patch(
+        ) as grp, patch("apps.assets.assets_client.create_asset", return_value="Asset-img"), patch(
            "apps.assets.assets_client.get_asset", return_value={"Status": "Processing"}
        ):
            grp.return_value = MagicMock(remote_group_id="Group-1")
            task = self._submit()
-        self.video.refresh_from_db()
        self.image.refresh_from_db()
-        self.video.review_status = "active"
-        self.video.review_remote_id = "asset-vid"
-        self.video.save(update_fields=["review_status", "review_remote_id"])
        self.image.review_status = "active"
        self.image.review_remote_id = "asset-img"
        self.image.save(update_fields=["review_status", "review_remote_id"])
        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
            "apps.assets.assets_client.get_asset", return_value={"Status": "Active"}
+        ), patch(
+            "apps.ai.video_digest.digest_asset_video",
+            return_value=(DIGEST_SAMPLE, {"digest_shots": 2}),
        ):
-            task = advance_video_replace(task)
+            run_replace_digest(task)
+        task.refresh_from_db()
        self.assertEqual(task.status, AITask.Status.SUBMITTED)
        self.assertTrue(CreditReservation.objects.filter(task=task).exists())
        self.assertTrue(self.provider.create_video_task.called)
        content = self.provider.create_video_task.call_args.kwargs.get("content_items") or []
-        urls = [item.get("video_url", item.get("image_url", {})).get("url") for item in content]
-        self.assertTrue(all(str(url).startswith("asset://") for url in urls if url))
+        types = [item.get("type") for item in content]
+        self.assertIn("video_url", types)
+        image_urls = [
+            (item.get("image_url") or {}).get("url")
+            for item in content if item.get("type") == "image_url"
+        ]
+        self.assertTrue(all(str(url).startswith("asset://") for url in image_urls if url))

    def test_advance_fails_review_without_charging(self):
        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
            "apps.assets.review.get_or_create_team_group"
-        ) as grp, patch("apps.assets.assets_client.create_asset", side_effect=["Asset-vid", "Asset-img"]), patch(
+        ) as grp, patch("apps.assets.assets_client.create_asset", return_value="Asset-img"), patch(
            "apps.assets.assets_client.get_asset", return_value={"Status": "Processing"}
        ):
            grp.return_value = MagicMock(remote_group_id="Group-1")
            task = self._submit()
+        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
+            "apps.assets.assets_client.get_asset", return_value={"Status": "Processing"}
+        ), patch(
+            "apps.ai.video_digest.digest_asset_video",
+            return_value=(DIGEST_SAMPLE, {"digest_shots": 2}),
+        ):
+            run_replace_digest(task)
+        task.refresh_from_db()
+        self.assertEqual(task.status, AITask.Status.CREATED)
        with patch("apps.assets.assets_client.is_enabled", return_value=True), patch(
            "apps.assets.assets_client.get_asset", return_value={"Status": "Failed", "ErrorMessage": "real person"}
        ):
@@ -866,9 +866,8 @@ class VideoReplaceView(APIView):

    def post(self, request):
        require_worker()
-        # 商品复刻的拆解在 worker 里跑;worker 还是旧镜像的话消息会被静默丢弃 → 永久「提炼中」。
-        if str((request.data or {}).get("replace_mode") or "") == "product":
-            require_worker_task("apps.ai.tasks.run_video_replace_digest_task")
+        # 复刻的拆解在 worker 里跑;worker 还是旧镜像的话消息会被静默丢弃 → 永久「提炼中」。
+        require_worker_task("apps.ai.tasks.run_video_replace_digest_task")
        from .video_replace import (
            VideoReplaceInProgress,
            serialize_video_replace_task,
@@ -897,8 +896,14 @@ class VideoReplaceView(APIView):
            public_error = classify_generation_error(
                exc, operation="video_generate", internal_kind=internal_kind
            )
+            # detail 用校验本身的原话:submit_video_replace 抛的 ValueError 全都是写给用户看的
+            # 中文("这个商品还没有可用图片""参考视频不能超过 30 秒"…)。套成 invalid_input 那句
+            # 万能文案("请检查描述、参数或素材格式后重试")等于把原因丢了,用户和排查都没法下手。
+            # error 里仍带结构化 code/action,前端要按类型渲染照旧可用。
+            payload = public_error.as_dict()
+            payload["fallback_message"] = message
            return Response(
-                {"detail": public_error.fallback_message, "error": public_error.as_dict()},
+                {"detail": message, "error": payload},
                status=status.HTTP_400_BAD_REQUEST,
            )
        task = _free_video_task_queryset(team).get(id=task.id)