fix(viral-video): xml_parser 剥离 CDATA 包裹，修复 copy_display_markdown 前端泄露 (#2238)

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

alembic/versions/104_v8_image_analysis_and_storyboard_v3.py

@@ -0,0 +1,242 @@
+# -*- coding: utf-8 -*-
+"""image_analysis v8 prompt + storyboard v3 prompt - 用户端展示格式 markdown 控制
+
+Revision ID: 104_v8_display_markdown
+Revises: 103_v7_prompt_and_tokens_3000
+Create Date: 2026-10-07
+
+变更：
+1. image_analysis v8: 在 v7 基础上 system_prompt 末尾追加「## 用户端展示格式」章节，
+   要求 VLM 在每张图的 JSON 里输出 summary_markdown 字段（markdown 格式的图片描述），
+   v8 设 is_active=true，v7 设 is_active=false。
+2. storyboard v3: 在 v2 基础上 system_prompt 追加要求 LLM 在 copy_result 中
+   输出 copy_display_markdown 字段（markdown 格式的完整文案展示），
+   v3 设 is_active=true，v2 设 is_active=false。
+"""
+
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "104_v8_display_markdown"
+down_revision = "103_v7_prompt_and_tokens_3000"
+branch_labels = None
+depends_on = None
+
+# ── v8 追加的 system prompt 内容 ──────────────────────────────────────
+V8_SYSTEM_APPEND = """
+
+## 用户端展示格式
+
+对于每张分析的图片，在 JSON 中额外输出一个 **summary_markdown** 字段，用 markdown 格式写出给用户看的图片描述。
+
+格式要求（根据图片类型自适应）：
+
+**商品图（type=product）**示例：
+### 商品名称
+**品牌**：品牌名 | **类目**：服饰鞋包/美妆/数码/...
+**核心特征**
+- 特征1：描述
+- 特征2：描述
+**外观**：颜色+材质+设计描述
+**包装**：包装类型描述
+**文字信息**：包装上看到的文字
+
+**门店场景图（type=store）**示例：
+### 门店名称/类型
+**类型**：奶茶店/便利店/养生馆/...
+**品牌标识**：招牌文字描述
+**环境氛围**：店内整体感觉
+**陈列亮点**
+- 亮点1
+- 亮点2
+**氛围**：亲民/专业/时尚/...
+
+**人物图（type=person）**示例：
+### 人物描述
+**形象**：年龄段 + 风格
+**穿搭**
+- 上装：颜色+款式
+- 下装：颜色+款式
+- 配饰：...
+**气质**：表情+姿势+整体感觉
+
+**风景/场景图（type=scene）**示例：
+### 场景名称
+**类型**：自然风景/城市街景/动物/美食
+**主体**：画面主要元素
+**氛围**：整体感觉描述
+
+要求：
+- 内容真实具体，从实际图片分析得出
+- 用 markdown 语法：**加粗**、列表、标题
+- 控制在 100-200 字
+- 不要编造图片中没有的信息
+"""
+
+# ── storyboard v3 追加的 system prompt 内容 ──────────────────────────
+V3_STORYBOARD_APPEND = """
+
+## 用户端展示格式
+
+在输出分镜脚本的同时，在顶层输出一个 **copy_display_markdown** 字段（用 XML 标签 <copy_display_markdown> 包裹），用 markdown 格式写出完整文案展示。
+
+格式示例：
+# 标题/主题
+
+## 整体概要
+一句话描述视频内容
+
+## 分镜预览
+
+### 镜头1（0-3秒）
+**景别**：近景俯拍，缓慢推镜
+**画面**：场景描述
+**台词**：口播文本
+**动作**：人物动作描述
+
+### 镜头2（3-9秒）
+...
+
+## 完整口播
+完整口播文案文本
+
+要求：
+- 把所有分镜按时间顺序整理成易读的格式
+- 用 markdown 语法组织，**加粗**标签、##二级标题、列表等
+- 控制在 300-500 字
+- 让用户一眼看懂视频会拍成什么样
+"""
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+
+    # ── 1. image_analysis v8 ──────────────────────────────────────────
+    # 停用所有 active image_analysis prompt
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = FALSE "
+            "WHERE prompt_type = 'image_analysis' AND is_active = TRUE"
+        )
+    )
+
+    # 读取 v7 的 prompt 内容作为基础
+    v7_row = bind.execute(
+        text(
+            "SELECT system_prompt, user_prompt_template, COALESCE(example_output, '') "
+            "FROM viral_video_prompt_templates "
+            "WHERE prompt_type = 'image_analysis' "
+            "ORDER BY version DESC LIMIT 1"
+        )
+    ).fetchone()
+
+    if v7_row:
+        v7_system = v7_row[0] or ""
+        v8_system = v7_system + V8_SYSTEM_APPEND
+        v8_user = v7_row[1] or "{image_url}"
+        v8_example = v7_row[2] or ""
+
+        # 幂等：已有 v8 则更新，否则插入
+        existing_v8 = bind.execute(
+            text("SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 8")
+        ).fetchone()
+        if existing_v8:
+            bind.execute(
+                text(
+                    "UPDATE viral_video_prompt_templates SET is_active = TRUE, "
+                    "system_prompt = :sys, user_prompt_template = :usr, "
+                    "example_output = :ex, name = 'v8 用户端展示格式', "
+                    "updated_at = NOW() "
+                    "WHERE prompt_type = 'image_analysis' AND version = 8"
+                ),
+                {"sys": v8_system, "usr": v8_user, "ex": v8_example},
+            )
+        else:
+            bind.execute(
+                text(
+                    "INSERT INTO viral_video_prompt_templates "
+                    "(prompt_type, version, name, system_prompt, user_prompt_template, "
+                    "example_output, is_active, created_at, updated_at) "
+                    "VALUES ('image_analysis', 8, 'v8 用户端展示格式', "
+                    ":sys, :usr, :ex, TRUE, NOW(), NOW())"
+                ),
+                {"sys": v8_system, "usr": v8_user, "ex": v8_example},
+            )
+
+    # ── 2. storyboard v3 ─────────────────────────────────────────────
+    # 停用所有 active storyboard prompt
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = FALSE "
+            "WHERE prompt_type = 'storyboard' AND is_active = TRUE"
+        )
+    )
+
+    # 读取当前 storyboard prompt
+    sb_row = bind.execute(
+        text(
+            "SELECT system_prompt, user_prompt_template, COALESCE(example_output, '') "
+            "FROM viral_video_prompt_templates "
+            "WHERE prompt_type = 'storyboard' "
+            "ORDER BY version DESC LIMIT 1"
+        )
+    ).fetchone()
+
+    if sb_row:
+        sb_system = sb_row[0] or ""
+        v3_system = sb_system + V3_STORYBOARD_APPEND
+        v3_user = sb_row[1] or ""
+        v3_example = sb_row[2] or ""
+
+        existing_v3 = bind.execute(
+            text("SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = 'storyboard' AND version = 3")
+        ).fetchone()
+        if existing_v3:
+            bind.execute(
+                text(
+                    "UPDATE viral_video_prompt_templates SET is_active = TRUE, "
+                    "system_prompt = :sys, user_prompt_template = :usr, "
+                    "example_output = :ex, name = 'v3 用户端展示格式', "
+                    "updated_at = NOW() "
+                    "WHERE prompt_type = 'storyboard' AND version = 3"
+                ),
+                {"sys": v3_system, "usr": v3_user, "ex": v3_example},
+            )
+        else:
+            bind.execute(
+                text(
+                    "INSERT INTO viral_video_prompt_templates "
+                    "(prompt_type, version, name, system_prompt, user_prompt_template, "
+                    "example_output, is_active, created_at, updated_at) "
+                    "VALUES ('storyboard', 3, 'v3 用户端展示格式', "
+                    ":sys, :usr, :ex, TRUE, NOW(), NOW())"
+                ),
+                {"sys": v3_system, "usr": v3_user, "ex": v3_example},
+            )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+
+    # 删除 v8
+    bind.execute(
+        text("DELETE FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 8")
+    )
+    # 恢复 v7 active
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = TRUE, updated_at = NOW() "
+            "WHERE prompt_type = 'image_analysis' AND version = 7"
+        )
+    )
+
+    # 删除 v3
+    bind.execute(text("DELETE FROM viral_video_prompt_templates " "WHERE prompt_type = 'storyboard' AND version = 3"))
+    # 恢复 storyboard v2 active
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = TRUE, updated_at = NOW() "
+            "WHERE prompt_type = 'storyboard' AND version = 2"
+        )
+    )

alembic/versions/105_narration_first.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""image_analysis v8 + storyboard v3 叙述优先重写版（架构大简化）
+
+Revision ID: 105_narration_first
+Revises: 104_v8_display_markdown
+Create Date: 2026-10-07
+
+变更：
+1. image_analysis v8：用「叙述优先」版整体替换 104 的 append 版——VLM 主交付物是
+   自然叙述 summary_markdown，结构化字段仅保留 type/name/brand/has_person，
+   顶层 products 改名 images；v8 active，其余 image_analysis 全部 deactivate。
+2. storyboard v3：整体替换为风格重写版（口播口语化、画面有画面感、
+   copy_display_markdown 流畅叙述）；v3 active，其余 storyboard deactivate。
+3. intent_parsing 类型模板全部 deactivate（意图解析步骤已删除）。
+模板内容直接取自 packages.application.viral_video.prompts.DEFAULT_TEMPLATES，
+保证代码默认值与 DB seed 完全一致。
+"""
+
+from sqlalchemy import text
+
+from alembic import op
+from packages.application.viral_video.prompts import DEFAULT_TEMPLATES
+
+revision = "105_narration_first"
+down_revision = "104_v8_display_markdown"
+branch_labels = None
+depends_on = None
+
+
+def _tpl(prompt_type: str, version: int) -> dict:
+    for t in DEFAULT_TEMPLATES:
+        if t["prompt_type"] == prompt_type and t["version"] == version:
+            return t
+    raise RuntimeError("default template missing: %s v%s" % (prompt_type, version))
+
+
+def _upsert(bind, t: dict) -> None:
+    existing = bind.execute(
+        text("SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = :pt AND version = :ver"),
+        {"pt": t["prompt_type"], "ver": t["version"]},
+    ).fetchone()
+    params = {
+        "pt": t["prompt_type"],
+        "ver": t["version"],
+        "name": t["name"],
+        "sys": t["system_prompt"],
+        "usr": t["user_prompt_template"],
+        "ex": t.get("example_output", "") or "",
+    }
+    if existing:
+        bind.execute(
+            text(
+                "UPDATE viral_video_prompt_templates SET name = :name, "
+                "system_prompt = :sys, user_prompt_template = :usr, "
+                "example_output = :ex, is_active = TRUE, updated_at = NOW() "
+                "WHERE prompt_type = :pt AND version = :ver"
+            ),
+            params,
+        )
+    else:
+        bind.execute(
+            text(
+                "INSERT INTO viral_video_prompt_templates "
+                "(prompt_type, version, name, system_prompt, user_prompt_template, "
+                "example_output, is_active, created_at, updated_at) "
+                "VALUES (:pt, :ver, :name, :sys, :usr, :ex, TRUE, NOW(), NOW())"
+            ),
+            params,
+        )
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+
+    # 1. image_analysis：停用全部后写入叙述优先 v8
+    bind.execute(
+        text("UPDATE viral_video_prompt_templates SET is_active = FALSE " "WHERE prompt_type = 'image_analysis'")
+    )
+    _upsert(bind, _tpl("image_analysis", 8))
+
+    # 2. storyboard：停用全部后写入重写版 v3
+    bind.execute(text("UPDATE viral_video_prompt_templates SET is_active = FALSE " "WHERE prompt_type = 'storyboard'"))
+    _upsert(bind, _tpl("storyboard", 3))
+
+    # 3. intent_parsing 已废弃：全部停用
+    bind.execute(
+        text("UPDATE viral_video_prompt_templates SET is_active = FALSE " "WHERE prompt_type = 'intent_parsing'")
+    )
+
+    # 4. review 模板确保 active
+    bind.execute(text("UPDATE viral_video_prompt_templates SET is_active = TRUE " "WHERE prompt_type = 'review'"))
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    # 恢复 104 的 v8/v3 无法重建（内容已替换），仅把版本 active 状态回退：
+    # 停用新版，尝试恢复 v7 / v2
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = FALSE "
+            "WHERE prompt_type IN ('image_analysis','storyboard') "
+            "AND version IN (8, 3)"
+        )
+    )
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = TRUE "
+            "WHERE prompt_type = 'image_analysis' AND version = 7"
+        )
+    )
+    bind.execute(
+        text(
+            "UPDATE viral_video_prompt_templates SET is_active = TRUE "
+            "WHERE prompt_type = 'storyboard' AND version = 2"
+        )
+    )

apps/api/app/services/lipsync_service.py

@@ -223,7 +223,47 @@ class LipsyncService:
         if timings:
             job.sentence_timings = timings
 
-        # 4. 检查是否走 GPU 路径：开关打开 + 有可用 Worker
+        # 4. 检查是否走 Ditto（蚂蚁数字人，#2076）：开关 + 配置完整
+        use_ditto = False
+        if self.settings.use_ditto_lipsync:
+            try:
+                from packages.application.ditto_service import get_ditto_client
+
+                ditto = get_ditto_client()
+                if ditto.is_configured:
+                    use_ditto = True
+                    logger.info("[lipsync] 优先走 Ditto 蚂蚁数字人: job_id=%s", job.id)
+                else:
+                    logger.info(
+                        "[lipsync] Ditto 开关已开但配置不完整（base_url=%s, template=%s），继续判断 GPU: job_id=%s",
+                        bool(ditto.base_url),
+                        bool(ditto.default_video_url),
+                        job.id,
+                    )
+            except Exception as exc:
+                logger.warning("[lipsync] Ditto 初始化失败，继续判断 GPU: job_id=%s err=%s", job.id, exc)
+
+        if use_ditto:
+            try:
+                # Ditto 使用预置人物模板视频，不用用户上传的 video_url；
+                # 但保留用户 video_url 以便失败回退到 GPU/MediaKit。
+                job.status = "processing"
+                job.mediakit_task_id = "ditto:submitted"
+                job.updated_at = datetime.now(UTC)
+                self.db.commit()
+                from app.tasks.lipsync_ditto import lipsync_ditto_process_async
+
+                lipsync_ditto_process_async.apply_async(args=(job.id, job.user_id))
+                logger.info("[lipsync] Ditto 任务已异步派发: job_id=%s", job.id)
+                return
+            except Exception as exc:
+                logger.warning("[lipsync] Ditto 派发失败，回退 GPU/MediaKit: job_id=%s err=%s", job.id, exc)
+                try:
+                    self.db.rollback()
+                except Exception:
+                    pass
+
+        # 5. 检查是否走 GPU 路径：开关打开 + 有可用 Worker
         use_gpu = False
         if self.settings.use_gpu_lipsync:
             try:

apps/api/app/tasks/lipsync_ditto.py

@@ -0,0 +1,318 @@
+"""Ditto 蚂蚁数字人口型异步任务 — #2076.
+
+把 Ditto 同步 HTTP 调用（30-120s）从 API 请求移到 Celery 后台执行：
+1. 加载 LipsyncJob
+2. 调 DittoClient.generate_and_persist(video_url=默认模板, audio_url=job.audio_url, script=job.script_text)
+3. 成功：标记 completed，写入 output_video_url（Ditto 输出自带音频，无需二次混流/超分）
+4. 失败：回退 GPU MuseTalk → 再失败回退 MediaKit
+
+注意：
+- 保留 MuseTalk 代码不动；Ditto 优先，失败按原链路兜底
+- Ditto 使用预置的人物模板视频（settings.ditto_default_video_url），不用用户上传的 video_url
+- 不传 GFPGAN 超分，不需要 ffmpeg 音视频混流
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Optional
+
+from celery import shared_task
+from sqlalchemy.orm import Session
+
+if TYPE_CHECKING:
+    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+
+logger = logging.getLogger(__name__)
+
+_DITTO_URL_TTL_SECONDS = 7 * 24 * 3600  # Ditto 结果 OSS URL 7 天有效
+
+
+def _get_db_session() -> Session:
+    try:
+        from worker_app.db import SessionLocal  # type: ignore
+    except ImportError:
+        from app.db import SessionLocal  # type: ignore
+    return SessionLocal()
+
+
+def _sign_media_url(url: str) -> str:
+    """对自家 OSS URL 签 7 天预签名。"""
+    if not url:
+        return url
+    try:
+        from urllib.parse import urlparse
+
+        from packages.shared.storage import get_shared_storage_service
+
+        storage = get_shared_storage_service()
+        public_base = getattr(storage, "public_url", "")
+        if not isinstance(public_base, str) or not public_base:
+            return url
+        own_host = urlparse(public_base).netloc.lower()
+        host = urlparse(url).netloc.lower()
+        if not own_host or host != own_host:
+            return url
+        return storage.get_download_url(url, expires_seconds=_DITTO_URL_TTL_SECONDS)
+    except Exception:
+        return url
+
+
+def _probe_video_duration(video_bytes: bytes) -> float:
+    """用 ffprobe 探测视频时长（秒）；失败返回 0。"""
+    try:
+        import os
+        import subprocess
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
+            tmp.write(video_bytes)
+            tmp_path = tmp.name
+        try:
+            out = subprocess.check_output(
+                [
+                    "ffprobe",
+                    "-v",
+                    "error",
+                    "-show_entries",
+                    "format=duration",
+                    "-of",
+                    "default=noprint_wrappers=1:nokey=1",
+                    tmp_path,
+                ],
+                stderr=subprocess.STDOUT,
+                timeout=10,
+            )
+            return float(out.decode().strip() or 0)
+        finally:
+            os.unlink(tmp_path)
+    except Exception as exc:
+        logger.warning("[ditto_task] ffprobe 失败: %s", exc)
+        return 0.0
+
+
+def _refund_lip_sync(db: Session, job: "LipsyncJobModel") -> None:
+    """Ditto 失败/取消时全额退款（复用 lipsync_service 的退款逻辑）。"""
+    try:
+        from app.services.lipsync_service import LipsyncService
+
+        LipsyncService(db)._refund_lip_sync(job)
+    except Exception:
+        logger.exception("[ditto_task] lip_sync 退款异常 job_id=%s", job.id)
+
+
+def _settle_lip_sync(db: Session, job: "LipsyncJobModel", duration: float) -> None:
+    """Ditto 成功后按实际时长结算。"""
+    try:
+        from app.services.lipsync_service import LipsyncService
+
+        LipsyncService(db)._settle_lip_sync(job, duration)
+    except Exception:
+        logger.exception("[ditto_task] lip_sync 结算异常 job_id=%s（不阻塞）", job.id)
+
+
+def _fallback_to_gpu_then_mediakit(db: Session, job: "LipsyncJobModel") -> None:
+    """Ditto 失败后：优先回退 GPU MuseTalk，再回退 MediaKit 云端。
+
+    复用 lipsync_service 现有路径逻辑以保证兜底一致性。
+    """
+    # 先尝试走 GPU MuseTalk（若可用）
+    try:
+        from app.services.gpu_lipsync_service import GpuLipsyncService
+        from app.tasks.lipsync_gpu import lipsync_gpu_process_async
+
+        gpu_svc = GpuLipsyncService(db)
+        if gpu_svc.has_available_worker():
+            logger.info("[ditto_task] 回退 GPU MuseTalk: job_id=%s", job.id)
+            # 复用 lipsync_service._submit_to_gpu_create 逻辑
+            from app.services.lipsync_service import LipsyncService
+
+            svc = LipsyncService(db)
+            storage = _shared_storage()
+            persisted_audio = None
+            try:
+                persisted_audio = svc._persist_external_audio_for_gpu(job=job, storage=storage)
+            except Exception as exc:
+                logger.warning("[ditto_task] GPU 外部音频转存失败: %s", exc)
+            audio_url_for_task = persisted_audio or job.audio_url
+            gpu_task = gpu_svc.create_task(
+                video_url=job.video_url,
+                audio_url=audio_url_for_task,
+                lipsync_job_id=job.id,
+                user_id=job.user_id,
+            )
+            if gpu_task is not None:
+                job.mediakit_task_id = f"gpu:{gpu_task.id}"
+                job.status = "processing"
+                job.updated_at = datetime.now(UTC)
+                db.commit()
+                lipsync_gpu_process_async.apply_async(args=(job.id, job.user_id, gpu_task.id))
+                return
+            db.rollback()
+    except Exception as exc:
+        logger.warning("[ditto_task] GPU MuseTalk 回退失败，转 MediaKit: %s", exc)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+
+    # 最后兜底：MediaKit 云端
+    try:
+        from app.services.mediakit_client import get_mediakit_client
+
+        client = get_mediakit_client()
+        video_url = _sign_media_url(job.video_url)
+        signed_audio_url = _sign_media_url(job.audio_url)
+        result = client.submit_lipsync(
+            video_url=video_url,
+            audio_url=signed_audio_url,
+            enable_video_loop=job.enable_video_loop,
+            client_token=job.id,
+        )
+        job.mediakit_task_id = result["task_id"]
+        job.status = "submitted"
+        job.submitted_at = datetime.now(UTC)
+        job.updated_at = datetime.now(UTC)
+        db.commit()
+        logger.info("[ditto_task] 已回退 MediaKit: job_id=%s task_id=%s", job.id, result["task_id"])
+    except Exception as exc:
+        job.status = "failed"
+        job.error_message = f"Ditto/GPU/MediaKit 均失败: {exc}"
+        job.error_code = "AllBackendsFailed"
+        job.updated_at = datetime.now(UTC)
+        db.commit()
+        logger.error("[ditto_task] 所有兜底均失败: job_id=%s err=%s", job.id, exc)
+
+
+def _shared_storage():
+    from packages.shared.storage import get_shared_storage_service
+
+    return get_shared_storage_service()
+
+
+@shared_task(
+    name="lipsync_ditto_process_async",
+    bind=True,
+    max_retries=0,
+    acks_late=True,
+    time_limit=600,
+    soft_time_limit=540,
+)
+def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
+    """异步调用 Ditto 生成口型视频。
+
+    Args:
+        job_id: LipsyncJob ID
+        user_id: 用户 ID
+    """
+    from packages.application.ditto_service import DittoError, get_ditto_client
+
+    db: Session = _get_db_session()
+    job: Optional[LipsyncJobModel] = None
+    try:
+        from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
+
+        job = db.query(LipsyncJobModel).filter_by(id=job_id, user_id=user_id).first()
+        if job is None:
+            logger.error("[ditto_task] job 不存在: job_id=%s", job_id)
+            return
+
+        if job.status != "processing":
+            logger.warning(
+                "[ditto_task] job 状态异常（非 processing），跳过: job_id=%s status=%s",
+                job_id,
+                job.status,
+            )
+            return
+
+        audio_url = job.audio_url or ""
+        script = job.script_text or ""
+        if not audio_url:
+            raise DittoError("job.audio_url 为空，无法调用 Ditto", code="InvalidParam")
+
+        logger.info(
+            "[ditto_task] 开始 Ditto 生成: job_id=%s audio=%s script_len=%d",
+            job_id,
+            audio_url[:100],
+            len(script),
+        )
+        client = get_ditto_client()
+        result = client.generate_and_persist(
+            job_id=job_id,
+            user_id=user_id,
+            audio_url=audio_url,
+            script=script,
+            # video_url 不传则用默认模板
+        )
+
+        # Ditto 返回的 MP4 自带音频，直接标记完成
+        job.output_video_url = result.video_url
+        # 探测时长（用于计费）
+        duration = _probe_video_duration(result.video_bytes)
+        if duration <= 0:
+            # 兜底：按音频时长估算（1秒≈1秒）
+            try:
+                from packages.domain.sentence_timings import probe_audio_duration
+                from packages.shared.url_security import safe_download_bytes
+
+                audio_data = safe_download_bytes(
+                    audio_url, allowed_mime_types=("audio/mpeg", "audio/wav", "audio/x-wav"), timeout=30
+                )
+                duration = probe_audio_duration(audio_data)
+            except Exception:
+                duration = 0.0
+        job.output_duration = duration
+        job.status = "completed"
+        job.completed_at = datetime.now(UTC)
+        job.updated_at = datetime.now(UTC)
+        db.commit()
+        logger.info(
+            "[ditto_task] Ditto 完成: job_id=%s url=%s duration=%.2fs rtf=%.2f frames=%d",
+            job_id,
+            result.video_url[:100],
+            duration,
+            result.rtf,
+            result.frames,
+        )
+        _settle_lip_sync(db, job, duration)
+
+    except DittoError as exc:
+        logger.error("[ditto_task] Ditto 失败，回退: job_id=%s code=%s err=%s", job_id, exc.code, exc)
+        if job is not None:
+            try:
+                db.rollback()
+                job = db.query(type(job)).filter_by(id=job_id).first() if hasattr(job, "id") else job
+                # 回退 GPU/MediaKit
+                _fallback_to_gpu_then_mediakit(db, job)
+            except Exception as fallback_exc:
+                logger.exception("[ditto_task] 回退也失败 job_id=%s err=%s", job_id, fallback_exc)
+                try:
+                    if job:
+                        job.status = "failed"
+                        job.error_message = f"Ditto 失败且回退异常: {exc}; fallback: {fallback_exc}"
+                        job.error_code = "FallbackError"
+                        job.updated_at = datetime.now(UTC)
+                        db.commit()
+                except Exception:
+                    pass
+    except Exception as exc:
+        logger.exception("[ditto_task] 未预期异常: job_id=%s err=%s", job_id, exc)
+        if job is not None:
+            try:
+                db.rollback()
+                job = db.query(type(job)).filter_by(id=job_id).first()
+                _fallback_to_gpu_then_mediakit(db, job)
+            except Exception as fallback_exc:
+                logger.exception("[ditto_task] 回退也失败 job_id=%s err=%s", job_id, fallback_exc)
+                try:
+                    if job:
+                        job.status = "failed"
+                        job.error_message = f"Ditto 异常: {exc}"
+                        job.error_code = "DittoAsyncError"
+                        job.updated_at = datetime.now(UTC)
+                        db.commit()
+                except Exception:
+                    pass
+    finally:
+        db.close()

apps/api/app/tasks/lipsync_tts.py

@@ -261,7 +261,33 @@ def tts_synthesize_and_submit(
                 "[lipsync_tts] 句子时间戳计算失败（不影响主流程）: job_id=%s err=%s", job_id, _st_err, exc_info=True
             )
 
-        # 3. 签名 URL 并提交到 MediaKit（复用模块内 _sign_media_url，避免对 LipsyncService 的耦合）
+        # 3. 优先走 Ditto（#2076）：开关打开且配置完整时，派发 Ditto 异步任务，不再走 MediaKit
+        ditto_dispatched = False
+        try:
+            from packages.config import get_api_settings as _get_settings
+
+            _settings = _get_settings()
+            if _settings.use_ditto_lipsync and _settings.ditto_api_base_url and _settings.ditto_default_video_url:
+                from app.tasks.lipsync_ditto import lipsync_ditto_process_async
+
+                job.status = "processing"
+                job.mediakit_task_id = "ditto:tts-submitted"
+                job.updated_at = datetime.now(UTC)
+                db.commit()
+                lipsync_ditto_process_async.apply_async(args=(job_id, user_id))
+                logger.info("[lipsync_tts] TTS 完成，已派发 Ditto 任务: job_id=%s", job_id)
+                ditto_dispatched = True
+        except Exception as _ditto_err:
+            logger.warning("[lipsync_tts] Ditto 派发失败，回退 MediaKit: job_id=%s err=%s", job_id, _ditto_err)
+            try:
+                db.rollback()
+            except Exception:
+                pass
+
+        if ditto_dispatched:
+            return
+
+        # 4. 签名 URL 并提交到 MediaKit（复用模块内 _sign_media_url，避免对 LipsyncService 的耦合）
         audio_url = _sign_media_url(job.audio_url)
         video_url = _sign_media_url(job.video_url)
 

apps/web/package-lock.json

@@ -14,6 +14,7 @@
         "axios": "^1.7.2",
         "classnames": "^2.5.1",
         "dayjs": "^1.11.23",
+        "marked": "^12.0.2",
         "mp4box": "^2.4.1",
         "react": "^18.3.1",
         "react-dom": "^18.3.1",
@@ -4502,6 +4503,17 @@
         "url": "https://github.com/sponsors/sindresorhus"
       }
     },
+    "node_modules/marked": {
+      "version": "12.0.2",
+      "resolved": "https://registry.npmmirror.com/marked/-/marked-12.0.2.tgz",
+      "integrity": "sha512-qXUm7e/YKFoqFPYPa3Ukg9xlI5cyAtGmyEIzMfW//m6kXwCy2Ps9DYf5ioijFKQ8qyuscrHoY04iJGctu2Kg0Q==",
+      "bin": {
+        "marked": "bin/marked.js"
+      },
+      "engines": {
+        "node": ">= 18"
+      }
+    },
     "node_modules/math-intrinsics": {
       "version": "1.1.0",
       "resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",

apps/web/package.json

@@ -25,6 +25,7 @@
     "axios": "^1.7.2",
     "classnames": "^2.5.1",
     "dayjs": "^1.11.23",
+    "marked": "^12.0.2",
     "mp4box": "^2.4.1",
     "react": "^18.3.1",
     "react-dom": "^18.3.1",

apps/web/src/api/viral-video/types.ts

@@ -65,27 +65,23 @@ export function isAnalysisStage(stage: ViralVideoStage | undefined): boolean {
   return isImageAnalysisStage(stage) || isCopyStage(stage)
 }
 
-/** 单张图片 VLM 识别出的商品信息 */
+/** 单张图片 VLM 识别结果（v8 叙述优先，仅保留最少结构化字段） */
 export interface ImageProductAnalysis {
+  /** store / product / person / scene */
+  type?: string
   name?: string
-  category?: string
   brand?: string
-  colors?: string[]
-  material_or_texture?: string
-  key_features?: string[]
-  visual_style?: string
-  scene?: string
-  target_audience_hint?: string
-  text_on_image?: string
-  /** 旧字段兼容 */
-  spec?: string
-  features?: string[] | string
-  label_text?: string
-  selling_points?: string
-  image_index?: number
+  has_person?: boolean
+  /** v8: 用户端展示用的叙述 markdown（由提示词控制排版） */
+  summary_markdown?: string
+  /** 标题行兼容字段 */
+  category?: string
 }
 
 export interface ImageAnalysisResult {
+  /** v8 字段 */
+  images?: ImageProductAnalysis[]
+  /** 老数据兼容 */
   products?: ImageProductAnalysis[]
 }
 
@@ -132,6 +128,8 @@ export interface CopyResult {
   /** 向后兼容：= voiceover_script */
   suggested_copy?: string
   title?: string
+  /** v3 storyboard: 用户端展示用的 markdown 文案（由提示词控制排版） */
+  copy_display_markdown?: string
   /** v1.5 旧字段兼容（老数据降级时可能出现） */
   scenes?: Array<{ shot: string; narration: string; duration?: number }>
 }

apps/web/src/pages/viral-video/ViralVideoPage.css

@@ -1973,3 +1973,99 @@
   padding-bottom: 6px;
   border-bottom: 1px dashed #e5e7eb;
 }
+
+/* ─────────── markdown 渲染（提示词控制展示格式） ─────────── */
+.vv-recog-md {
+  padding: 4px 0;
+}
+.vv-copy-preview {
+  margin-bottom: 14px;
+  padding: 12px 14px;
+  background: linear-gradient(180deg, #faf7ff 0%, #f6f2ff 100%);
+  border: 1px solid #ece4fb;
+  border-radius: 10px;
+}
+.vv-copy-preview-h {
+  margin: 0 0 8px;
+  border-bottom: none;
+  padding-bottom: 0;
+}
+.vv-md-body {
+  font-size: 13px;
+  line-height: 1.7;
+  color: #374151;
+  word-break: break-word;
+}
+.vv-md-body h1,
+.vv-md-body h2,
+.vv-md-body h3,
+.vv-md-body h4 {
+  margin: 10px 0 6px;
+  font-weight: 600;
+  color: #1f2937;
+  line-height: 1.4;
+}
+.vv-md-body h1 {
+  font-size: 18px;
+}
+.vv-md-body h2 {
+  font-size: 16px;
+}
+.vv-md-body h3 {
+  font-size: 15px;
+}
+.vv-md-body h4 {
+  font-size: 14px;
+}
+.vv-md-body p {
+  margin: 6px 0;
+}
+.vv-md-body ul,
+.vv-md-body ol {
+  margin: 6px 0;
+  padding-left: 20px;
+}
+.vv-md-body li {
+  margin: 3px 0;
+}
+.vv-md-body strong {
+  color: #111827;
+  font-weight: 600;
+}
+.vv-md-body blockquote {
+  margin: 8px 0;
+  padding: 4px 12px;
+  border-left: 3px solid #7c3aed;
+  background: rgba(124, 58, 237, 0.05);
+  color: #4b5563;
+}
+.vv-md-body code {
+  padding: 1px 5px;
+  background: #f3f4f6;
+  border-radius: 4px;
+  font-size: 12px;
+  color: #be185d;
+}
+.vv-md-body a {
+  color: #7c3aed;
+  text-decoration: none;
+}
+.vv-md-body a:hover {
+  text-decoration: underline;
+}
+.vv-md-body table {
+  border-collapse: collapse;
+  margin: 8px 0;
+  width: 100%;
+}
+.vv-md-body th,
+.vv-md-body td {
+  border: 1px solid #e5e7eb;
+  padding: 6px 10px;
+  text-align: left;
+}
+.vv-md-body hr {
+  border: none;
+  border-top: 1px solid #e5e7eb;
+  margin: 12px 0;
+}

apps/web/src/pages/viral-video/ViralVideoPage.tsx

@@ -1,5 +1,6 @@
 import React, { useCallback, useEffect, useRef, useState } from "react"
 import axios from "axios"
+import { marked } from "marked"
 import {
   PlusOutlined,
   CloseOutlined,
@@ -150,6 +151,16 @@ type TabTask = {
   audioInst: HTMLAudioElement | null
 }
 
+/* ── marked 配置：禁用 mangle/headerIds，输出干净 HTML ── */
+marked.setOptions({ gfm: true, breaks: false })
+const renderMarkdown = (md: string): string => {
+  try {
+    return marked.parse(md ?? "", { async: false }) as string
+  } catch {
+    return (md ?? "").replace(/&/g, "&amp;").replace(/</g, "&lt;")
+  }
+}
+
 /* ─────────── 常量 ─────────── */
 
 const LANGUAGES = ["中文(普通话)", "粤语", "英语", "日语", "韩语"]
@@ -296,6 +307,8 @@ interface Storyboard {
   hard_constraints: string[]
   negative_prompts: string[]
   voiceover_script: string
+  /** v3: 用户端展示用 markdown 文案（由提示词控制排版） */
+  copy_display_markdown: string
 }
 
 /** 兼容旧 copy_result（final_copy/title/scenes）→ 新 Storyboard 结构 */
@@ -323,6 +336,7 @@ function copyResultToStoryboard(cr: CopyResult | null | undefined): Storyboard |
       hard_constraints: Array.isArray(cr.hard_constraints) ? cr.hard_constraints : [],
       negative_prompts: Array.isArray(cr.negative_prompts) ? cr.negative_prompts : [],
       voiceover_script: cr.voiceover_script || cr.final_copy || cr.suggested_copy || "",
+      copy_display_markdown: cr.copy_display_markdown || "",
     }
   }
   // 兜底：旧结构转简单分镜
@@ -358,6 +372,7 @@ function copyResultToStoryboard(cr: CopyResult | null | undefined): Storyboard |
     hard_constraints: [],
     negative_prompts: [],
     voiceover_script: finalCopy,
+    copy_display_markdown: cr.copy_display_markdown || "",
   }
 }
 
@@ -412,6 +427,7 @@ const MOCK_STORYBOARD: Storyboard = {
   negative_prompts: ["冷色调", "模糊", "变形", "水印文字", "卡通风格", "空无一人"],
   voiceover_script:
     "还在为餐桌选不到好桌子发愁？这张北美黑胡桃木餐桌，一家人坐下来吃饭刚刚好。全实木、无贴皮，纹理好看又耐刮。点小黄车，给家里添一张好桌子。",
+  copy_display_markdown: "",
 }
 
 const fmtSize = (bytes: number | undefined) => {
@@ -1192,8 +1208,10 @@ const ViralVideoPage: React.FC = () => {
 
   /* ── 识别描述汇览渲染 ── */
   const renderRecognition = () => {
-    const products: ImageProductAnalysis[] =
-      (task.imageAnalysis?.products as ImageProductAnalysis[] | undefined) || []
+    const images: ImageProductAnalysis[] =
+      (task.imageAnalysis?.images as ImageProductAnalysis[] | undefined) ||
+      (task.imageAnalysis?.products as ImageProductAnalysis[] | undefined) ||
+      []
     if (task.uiStep === "step1_analyzing") {
       return (
         <div className="vv-recog">
@@ -1201,83 +1219,36 @@ const ViralVideoPage: React.FC = () => {
             <LoadingOutlined style={{ color: "#7c3aed", marginRight: 6 }} />
             识别描述汇览
           </div>
-          <div className="vv-muted">AI 正在识别商品特征…</div>
+          <div className="vv-muted">AI 正在识别画面…</div>
         </div>
       )
     }
-    if (products.length === 0) return null
-    const featureText = (f: string[] | string | undefined) => {
-      if (!f) return ""
-      if (Array.isArray(f)) return f.join("；")
-      return f
-    }
+    if (images.length === 0) return null
     return (
       <div className="vv-recog">
         <div className="vv-recog-title">
           <CheckCircleFilled style={{ color: "#10b981" }} />
           识别描述汇览
         </div>
-        {products.map((p, i) => (
-          <div key={i} className="vv-recog-item">
-            <div className="vv-recog-line">
-              <span className="vv-recog-k">图片{i + 1}：</span>
-              <span>
-                {p.name || "未识别"}
-                {p.spec && <span className="vv-recog-meta">（{p.spec}）</span>}
-                {p.brand && <span className="vv-recog-meta"> · {p.brand}</span>}
-                {p.category && <span className="vv-recog-meta"> · {p.category}</span>}
-              </span>
+        {images.map((p, i) => {
+          const meta = [p.name || "未识别", p.brand, p.category].filter(Boolean)
+          return (
+            <div key={i} className="vv-recog-item vv-recog-md">
+              <div className="vv-recog-line">
+                <span className="vv-recog-k">图片{i + 1}：</span>
+                <span>{meta.join(" · ")}</span>
+              </div>
+              {p.summary_markdown ? (
+                <div
+                  className="vv-md-body"
+                  dangerouslySetInnerHTML={{ __html: renderMarkdown(p.summary_markdown) }}
+                />
+              ) : (
+                <div className="vv-muted">（暂无叙述描述）</div>
+              )}
             </div>
-            {featureText(p.key_features ?? p.features) && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">核心特征：</span>
-                <span className="vv-recog-v">{featureText(p.key_features ?? p.features)}</span>
-              </div>
-            )}
-            {p.colors && p.colors.length > 0 && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">主色调：</span>
-                <span className="vv-recog-v">{p.colors.join(" / ")}</span>
-              </div>
-            )}
-            {p.material_or_texture && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">材质/纹理：</span>
-                <span className="vv-recog-v">{p.material_or_texture}</span>
-              </div>
-            )}
-            {p.visual_style && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">视觉风格：</span>
-                <span className="vv-recog-v">{p.visual_style}</span>
-              </div>
-            )}
-            {p.scene && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">场景：</span>
-                <span className="vv-recog-v">{p.scene}</span>
-              </div>
-            )}
-            {p.target_audience_hint && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">目标人群：</span>
-                <span className="vv-recog-v">{p.target_audience_hint}</span>
-              </div>
-            )}
-            {(p.text_on_image || p.label_text) && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">包装文字：</span>
-                <span className="vv-recog-v">{p.text_on_image || p.label_text}</span>
-              </div>
-            )}
-            {p.selling_points && (
-              <div className="vv-recog-line">
-                <span className="vv-recog-k">卖点：</span>
-                <span className="vv-recog-v">{p.selling_points}</span>
-              </div>
-            )}
-          </div>
-        ))}
+          )
+        })}
       </div>
     )
   }
@@ -1416,6 +1387,19 @@ const ViralVideoPage: React.FC = () => {
     return (
       <div className="vv-copy-box vv-storyboard">
         <div className="vv-sb-doc">
+          {/* 文案预览（提示词控制排版，只读；编辑在下方分镜字段中进行） */}
+          {sb.copy_display_markdown && (
+            <div className="vv-copy-preview">
+              <h4 className="vv-sb-h vv-copy-preview-h">
+                <FileTextOutlined style={{ color: "#7c3aed", marginRight: 6 }} />
+                文案预览
+              </h4>
+              <div
+                className="vv-md-body"
+                dangerouslySetInnerHTML={{ __html: renderMarkdown(sb.copy_display_markdown) }}
+              />
+            </div>
+          )}
           {/* 视频总览 */}
           <h4 className="vv-sb-h">视频总览</h4>
           <p className="vv-sb-inline-row">

apps/worker/worker_app/celery_app.py

@@ -53,6 +53,9 @@ celery_app.conf.imports = (
     # #1998 GPU MuseTalk 异步推理：wait_for_result→签名 URL→回写 lipsync_jobs
     # 必须在 Worker 侧注册，否则 apply_async 消息无人消费，job 永远卡在 processing
     "app.tasks.lipsync_gpu",
+    # #2076 Ditto 蚂蚁数字人异步推理：同步 HTTP 调用 Ditto → MP4 流转存 OSS → 回写 lipsync_jobs
+    # 必须在 Worker 侧注册；失败回退 GPU MuseTalk → MediaKit
+    "app.tasks.lipsync_ditto",
 )
 
 # Celery Beat 定时任务调度

apps/worker/worker_app/tasks/vision/_prompt.py

@@ -1,13 +1,13 @@
 # -*- coding: utf-8 -*-
-"""V2 prompt 解析：优先读后台 viral_video_prompt_templates 表（prompt_type='image_analysis'
-且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到纯硬编码 JSON schema prompt。
+"""V2 prompt 解析：优先读后台 viral_video_prompt_templates（prompt_type='image_analysis'
+且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到 prompts.py 的
+image_analysis v8 默认 system/user。
 
-规则（简单直接，不做字符串匹配判断）：
-- DB 有 is_active=true 的 image_analysis 记录（含种子版本和用户修改后的版本）：
-  * system = DB.system_prompt（DB prompt 自带完整输出格式，不追加硬编码 schema，
-    避免 DB 写 XML、调用强制 json_object 造成的格式冲突）
-  * user   = DB.user_prompt_template 渲染后使用；渲染后为空则用硬编码默认
-- DB 无记录/连接异常/返回空：system/user 全部用纯硬编码 JSON schema prompt
+规则：
+- DB 有 is_active=true 的 image_analysis 记录：system 原样用 DB.system_prompt
+  （自带完整输出格式，不追加任何硬编码 schema），user 用 DB.user_prompt_template
+  渲染（填入 image_url / ocr_text）；
+- DB 无记录/异常：system/user 用 prompts.py 里的 v8 默认模板。
 """
 
 from __future__ import annotations
@@ -15,192 +15,79 @@ from __future__ import annotations
 import logging
 import threading
 import time
-from typing import Any
 
 logger = logging.getLogger(__name__)
 
-# ---- 纯硬编码 JSON schema（DB 无有效配置时全量使用） ----
 
-_FAST_JSON_SCHEMA = (
-    "你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
-    "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "upper_wear": "上装款式，如T恤/衬衫/卫衣/毛衣/西装/夹克/连衣裙/吊带/背心/外套等",\n'
-    '  "upper_color": "上装主色",\n'
-    '  "lower_wear": "下装款式；穿连衣裙时填null",\n'
-    '  "lower_color": "下装主色",\n'
-    '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
-    '  "accessories": ["眼镜"/"帽子"/"项链"/"耳环"/"背包"/"手表"等数组],\n'
-    '  "hairstyle": "发型，如短发/长发/马尾/卷发/丸子头/光头等",\n'
-    '  "expression": "表情，如微笑/严肃/酷/开心等",\n'
-    '  "pose": "姿势，如站立/坐姿/侧身/行走等",\n'
-    '  "scene": "场景，如室内/街拍/户外/办公室/家居/海边/雪景/森林等",\n'
-    '  "style": "风格，如休闲/商务/运动/复古/潮流/甜美/酷飒/优雅/街头/法式等",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名称，非产品图填null",\n'
-    '  "brand": "品牌或文字标识，无则null",\n'
-    '  "material": "材质，如棉质/牛仔/皮革/真丝/针织/涤纶等",\n'
-    '  "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
-    '  "colors": ["主色数组"],\n'
-    '  "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
-    "}\n\n"
-    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
-)
-DEFAULT_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
+def _default_template() -> dict:
+    # 延迟导入：避免模块加载时拉起整个 packages 依赖链（也便于旧 Python 收集测试）
+    from packages.application.viral_video.prompts import DEFAULT_TEMPLATES
 
-_PRO_JSON_SCHEMA = (
-    "你是图片分析专家。严格按下方 JSON schema 返回一个对象，不要解释、不要markdown、不要代码块、不要XML标签。\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "outfit": "整体穿着描述（含颜色款式）",\n'
-    '  "hair": "发型发色",\n'
-    '  "pose": "姿势",\n'
-    '  "expression": "表情",\n'
-    '  "scene": "场景",\n'
-    '  "mood": "氛围",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名，非产品图填null",\n'
-    '  "brand": "品牌，无则null",\n'
-    '  "key_features": ["核心特征数组，3-6个短语"]\n'
-    "}\n\n"
-    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
-)
-DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。"
+    for item in DEFAULT_TEMPLATES:
+        if item["prompt_type"] == "image_analysis":
+            return item
+    raise RuntimeError("image_analysis 默认模板缺失")
 
-# 保留旧 JSON schema 追加文本作为常量（DB prompt 完全控制输出格式后不再使用，
-# 保留以便排查历史行为）。
-_FAST_JSON_APPEND = (
-    "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，"
-    "严格包含以下字段（字段值不确定时填null或空数组）：\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "upper_wear": "上装款式字符串",\n'
-    '  "upper_color": "上装主色",\n'
-    '  "lower_wear": "下装款式（穿连衣裙时填null）",\n'
-    '  "lower_color": "下装主色",\n'
-    '  "dress_color": "连衣裙主色（穿连衣裙时填）",\n'
-    '  "accessories": ["配饰数组"],\n'
-    '  "hairstyle": "发型",\n'
-    '  "expression": "表情",\n'
-    '  "pose": "姿势",\n'
-    '  "scene": "场景",\n'
-    '  "style": "风格",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "产品类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名称，非产品图填null",\n'
-    '  "brand": "品牌或文字标识，无则null",\n'
-    '  "material": "材质",\n'
-    '  "pattern": "图案",\n'
-    '  "colors": ["主色数组"],\n'
-    '  "mood": "整体氛围"\n'
-    "}\n"
-    "不要输出任何其他文字、解释、XML标签或markdown。"
-)
-
-_PRO_JSON_APPEND = (
-    "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，"
-    "严格包含以下字段（字段值不确定时填null或空数组）：\n"
-    "{\n"
-    '  "has_person": true/false,\n'
-    '  "gender": "男"/"女"/null,\n'
-    '  "age_range": "儿童"/"青少年"/"青年"/"中年"/"老年"/null,\n'
-    '  "outfit": "整体穿着描述（含颜色款式）",\n'
-    '  "hair": "发型发色",\n'
-    '  "pose": "姿势",\n'
-    '  "expression": "表情",\n'
-    '  "scene": "场景",\n'
-    '  "mood": "氛围",\n'
-    '  "has_product": true/false,\n'
-    '  "category": "类目：服饰/鞋包/美妆/数码/食品/家居/配饰/母婴/非产品图",\n'
-    '  "product_name": "产品名，非产品图填null",\n'
-    '  "brand": "品牌，无则null",\n'
-    '  "key_features": ["核心特征3-6个短语"]\n'
-    "}\n"
-    "不要输出任何其他文字、解释、XML标签或markdown。"
-)
 
 _cache_lock = threading.Lock()
-_cache: dict[str, tuple[float, Any]] = {}
+_cache: dict[str, tuple[float, tuple[str, str]]] = {}
 _CACHE_TTL = 30.0
 
 
-def _load_db_template() -> Any | None:
-    """直接查DB viral_video_prompt_templates 中 is_active=true 的 image_analysis 记录；
-    DB不可达/无记录/异常返回None。
-    复用 prompt_loader._load_from_db，它只查DB不做DEFAULT_TEMPLATES fallback，
-    返回None表示DB无记录或异常。"""
+def _load_db_template():
+    """查 DB is_active=true 的 image_analysis 记录；不可达/无记录返回 None。"""
     try:
         from packages.application.viral_video.prompt_loader import _load_from_db
 
         return _load_from_db("image_analysis")
-    except Exception as e:
-        logger.warning("[vision.v2] 查询DB prompt配置失败: %s", e)
+    except Exception as e:  # noqa: BLE001
+        logger.warning("[vision.v2] 查询DB image_analysis prompt失败: %s", e)
         return None
 
 
-def _render_user(tpl: Any | None, default_user: str) -> str:
-    if not tpl:
-        return default_user
-    tpl_str = getattr(tpl, "user_prompt_template", "") or ""
-    if not tpl_str.strip():
-        return default_user
-    rendered = tpl_str.replace("{image_count}", "1").replace("{industry}", "通用").replace("{image_urls}", "").strip()
-    return rendered or default_user
+def _render_user(user_tpl: str, image_url: str, ocr_text: str) -> str:
+    try:
+        return user_tpl.format(image_url=image_url, ocr_text=ocr_text or "无")
+    except Exception:  # noqa: BLE001
+        return user_tpl
 
 
-def resolve_fast_prompt() -> tuple[str, str]:
-    return _resolve("fast")
-
-
-def resolve_pro_prompt() -> tuple[str, str]:
-    return _resolve("pro")
-
-
-def _resolve(kind: str) -> tuple[str, str]:
+def _resolve(kind: str, image_url: str = "", ocr_text: str = "") -> tuple[str, str]:
     now = time.time()
     cache_key = f"prompt_{kind}"
     with _cache_lock:
         hit = _cache.get(cache_key)
         if hit and now - hit[0] < _CACHE_TTL:
-            return hit[1]
+            sys_prompt, usr_prompt = hit[1]
+            return sys_prompt, _render_user(usr_prompt, image_url, ocr_text)
 
-    default_sys = _FAST_JSON_SCHEMA if kind == "fast" else _PRO_JSON_SCHEMA
-    default_user = DEFAULT_FAST_USER if kind == "fast" else DEFAULT_PRO_USER
+    default = _default_template()
+    sys_prompt = default["system_prompt"]
+    usr_prompt = default["user_prompt_template"]
 
-    sys_prompt = default_sys
-    usr_prompt = default_user
-    try:
-        tpl = _load_db_template()
-        if tpl is not None:
-            db_sys = (getattr(tpl, "system_prompt", "") or "").strip()
-            if db_sys:
-                sys_prompt = db_sys  # DB prompt自带完整输出格式，不追加硬编码schema避免冲突
-                usr_prompt = _render_user(tpl, default_user)
-                logger.info(
-                    "[vision.v2] 使用DB image_analysis prompt (kind=%s version=%s sys_len=%d)",
-                    kind,
-                    getattr(tpl, "version", "?"),
-                    len(db_sys),
-                )
-            else:
-                logger.debug("[vision.v2] DB image_analysis system_prompt为空，使用默认JSON (kind=%s)", kind)
-        else:
-            logger.debug("[vision.v2] DB无image_analysis记录/不可达，使用默认JSON prompt (kind=%s)", kind)
-    except Exception as e:
-        logger.warning("[vision.v2] 解析DB prompt异常，使用默认: %s", e)
+    tpl = _load_db_template()
+    if tpl is not None:
+        db_sys = (getattr(tpl, "system_prompt", "") or "").strip()
+        if db_sys:
+            sys_prompt = db_sys
+            db_usr = getattr(tpl, "user_prompt_template", "") or usr_prompt
+            usr_prompt = db_usr or usr_prompt
+            logger.info(
+                "[vision.v2] 使用DB image_analysis prompt version=%s",
+                getattr(tpl, "version", "?"),
+            )
 
     with _cache_lock:
         _cache[cache_key] = (now, (sys_prompt, usr_prompt))
-    return sys_prompt, usr_prompt
+    return sys_prompt, _render_user(usr_prompt, image_url, ocr_text)
+
+
+def resolve_fast_prompt(image_url: str = "", ocr_text: str = "") -> tuple[str, str]:
+    return _resolve("fast", image_url, ocr_text)
+
+
+def resolve_pro_prompt(image_url: str = "", ocr_text: str = "") -> tuple[str, str]:
+    return _resolve("pro", image_url, ocr_text)
 
 
 def invalidate_cache() -> None:

apps/worker/worker_app/tasks/vision/fast_path.py

@@ -1,12 +1,11 @@
 # -*- coding: utf-8 -*-
-"""V2 图片分析主路径：每图并行 OCR（火山MediaKit，未配置时自动跳过）+ qwen3.8-flash JSON VLM，
-失败时单次 qwen3.7-plus 兜底。
+"""V2 图片分析主路径（v8 叙述优先）：每图并行 OCR（火山 MediaKit，未配置自动跳过）
++ fast VLM 强约束 JSON；失败时单次 pro VLM 兜底。
 
-架构（灵应10-05确认）：
-- 唯一后端：阿里云百炼 DashScope，qwen3.8-flash 做快速路径、qwen3.7-plus 做兜底
-- 主力：单图2路并行（OCR + fast VLM），外层N图全并发（workers=8）
-- 兜底：单次 pro VLM 调用，无竞速/重试/复杂超时
-- 输出 dict 格式与旧版完全一致，下游零改动
+架构：
+- 单图 2 路并行（OCR + fast VLM），外层 N 图全并发（workers=8）；
+- 兜底单次 pro VLM，无竞速/复杂重试；
+- 输出统一为 5 字段 image dict（type/name/brand/has_person/summary_markdown）。
 """
 
 from __future__ import annotations
@@ -21,36 +20,24 @@ from . import assembler, ocr_volc, vlm_fallback, vlm_fast_json
 
 logger = logging.getLogger(__name__)
 
-# 超时（可通过环境变量覆盖）
 _IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
 _FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "20"))
 _FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "20"))
 _OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
 _PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
 
-_FALLBACK_RESULT = {
-    "name": "未识别",
-    "brand": "无法判断",
-    "category": "非产品图",
-    "appearance": "无法判断",
-    "packaging": "无法判断",
-    "text_on_package": [],
-    "key_features": ["无法判断"],
-    "scene": "通用",
-    "mood": "",
-    "portrait_prompt": "无法判断",
-    "summary": "未识别",
-}
+
+def _is_usable(r: dict[str, Any] | None) -> bool:
+    if not isinstance(r, dict):
+        return False
+    return bool((r.get("summary_markdown") or "").strip())
 
 
-def _is_usable(r: dict[str, Any]) -> bool:
-    pp = (r.get("portrait_prompt") or "").strip()
-    if pp and pp not in ("无人像", "无法判断", "未识别"):
-        return True
-    name = (r.get("name") or "").strip()
-    if name and name not in ("未识别", "无法判断", "未知"):
-        return True
-    return False
+def _basic_failure(ocr_result: list[str], fast_elapsed: float, source: str) -> dict[str, Any]:
+    image = assembler.assemble_result(-1, {}, ocr_result)
+    image["_source"] = source
+    image["_fast_elapsed"] = round(fast_elapsed, 2)
+    return image
 
 
 def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
@@ -58,7 +45,6 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
 
     fj_result: dict[str, Any] | None = None
     ocr_result: list[str] = []
-    fast_elapsed = 0.0
     pool = ThreadPoolExecutor(max_workers=2)
     f_fj = pool.submit(vlm_fast_json.call_fast_json, img_url, timeout=_FAST_JSON_TIMEOUT)
     f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=_OCR_TIMEOUT)
@@ -66,7 +52,7 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
         for fut in as_completed([f_fj, f_ocr], timeout=_FAST_TIMEOUT):
             try:
                 res = fut.result(timeout=1)
-            except Exception as e:
+            except Exception as e:  # noqa: BLE001
                 logger.warning("[vision.v2] 图片 #%d 子任务异常: %s", idx, e)
                 continue
             if fut is f_fj and isinstance(res, dict):
@@ -80,37 +66,24 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
         logger.warning("[vision.v2] 图片 #%d fast路径超时(%.0fs)，走pro兜底", idx, _FAST_TIMEOUT)
     finally:
         fast_elapsed = time.time() - t0
-        pool.shutdown(wait=False)  # 不等待未完成的线程，避免计时膨胀
+        pool.shutdown(wait=False)
 
     if fj_result:
         assembled = assembler.assemble_result(idx, fj_result, ocr_result)
         if _is_usable(assembled):
             assembled["_fast_elapsed"] = round(fast_elapsed, 2)
-            logger.info(
-                "[vision.v2] 图片 #%d fast命中 elapsed=%.2fs pp=%s",
-                idx,
-                fast_elapsed,
-                (assembled.get("portrait_prompt") or "")[:40],
-            )
+            logger.info("[vision.v2] 图片 #%d fast命中 elapsed=%.2fs", idx, fast_elapsed)
             return assembled
 
-    pro_t0 = time.time()
-    pro_result = vlm_fallback.call_pro_vlm(img_url, idx, timeout=_PRO_TIMEOUT)
-    if pro_result and _is_usable(pro_result):
+    pro_result = vlm_fallback.call_pro_vlm(img_url, idx, ocr_hint=ocr_result, timeout=_PRO_TIMEOUT)
+    if _is_usable(pro_result):
         pro_result["_fallback_used"] = True
         pro_result["_fast_elapsed"] = round(fast_elapsed, 2)
-        pro_result["_pro_elapsed"] = round(time.time() - pro_t0, 2)
-        if ocr_result and not pro_result.get("text_on_package"):
-            pro_result["text_on_package"] = ocr_result[:8]
         logger.info("[vision.v2] 图片 #%d pro兜底命中 total=%.2fs", idx, time.time() - t0)
         return pro_result
 
     logger.warning("[vision.v2] 图片 #%d 全路径失败 elapsed=%.2fs", idx, time.time() - t0)
-    out = dict(_FALLBACK_RESULT)
-    out["_source"] = "v2_all_failed"
-    out["text_on_package"] = ocr_result[:8]
-    out["_fast_elapsed"] = round(fast_elapsed, 2)
-    return out
+    return _basic_failure(ocr_result, fast_elapsed, "v2_all_failed")
 
 
 def analyze_images_v2(img_urls: list[str]) -> list[dict[str, Any]]:
@@ -133,14 +106,19 @@ def analyze_images_v2(img_urls: list[str]) -> list[dict[str, Any]]:
             idx = future_to_idx[fut]
             try:
                 results[idx] = fut.result()
-            except Exception as e:
+            except Exception as e:  # noqa: BLE001
                 logger.warning("[vision.v2] 图片 #%d future异常: %s", idx, e, exc_info=True)
-                r = dict(_FALLBACK_RESULT)
-                r["_source"] = "v2_future_exception"
-                results[idx] = r
+                results[idx] = assembler.assemble_result(idx, {}, [])
+                results[idx]["_source"] = "v2_future_exception"  # type: ignore[index]
 
     elapsed = time.time() - t0
-    succ = sum(1 for r in results if r and _is_usable(r))
+    succ = sum(1 for r in results if _is_usable(r))
     fb = sum(1 for r in results if r and r.get("_fallback_used"))
-    logger.info("[vision.v2] 完成 n=%d usable=%d pro_fallback=%d elapsed=%.2fs", len(img_urls), succ, fb, elapsed)
-    return [r for r in results if r is not None]
+    logger.info(
+        "[vision.v2] 完成 n=%d usable=%d pro_fallback=%d elapsed=%.2fs",
+        len(img_urls),
+        succ,
+        fb,
+        elapsed,
+    )
+    return [r for r in results if r is not None]  # type: ignore[misc]

apps/worker/worker_app/tasks/vision/vlm_fallback.py

@@ -1,14 +1,8 @@
 # -*- coding: utf-8 -*-
-"""V2 兜底路径：image_analysis（默认 qwen-vl-plus 视觉模型，fallback qwen3.7-plus / DashScope）单图调用。
+"""V2 兜底路径：vision client（fallback 变体）单图调用，走 v8 叙述优先 prompt。
 
-fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
-设计要点：
-- 通过 ai_router.get_vision_client() 获取 DoubaoClient 实例，不再自己拼 httpx 请求
-- enable_thinking=False + response_format=json_object
-- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
-- max_tokens 不传，使用 client 中 capability 的 DB 配置（避免硬编码截断 JSON）
-- timeout=30s
-- 返回 dict 统一走 assembler.assemble_result 组装，与 fast 路径输出格式完全一致
+fast 超时/非 JSON/为空时单次调用；输出统一走 assembler.assemble_result 组装，
+与 fast 路径同为 5 字段 image dict。
 """
 
 from __future__ import annotations
@@ -28,11 +22,12 @@ def call_pro_vlm(
     img_url: str,
     idx: int,
     *,
+    ocr_hint: list[str] | None = None,
     timeout: int = _DEFAULT_TIMEOUT,
     max_tokens: int | None = None,
 ) -> dict[str, Any] | None:
-    """max_tokens 默认 None：不显式传参，使用 client 内 capability 的 DB 配置。"""
     t0 = time.time()
+    ocr_text = "、".join(t for t in (ocr_hint or []) if t)[:200]
 
     try:
         from packages.shared.ai_router import ai_router
@@ -41,13 +36,13 @@ def call_pro_vlm(
         if not client or not client.is_available:
             logger.warning("[vision.v2] pro vision client 不可用，跳过")
             return None
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001
         logger.warning("[vision.v2] ai_router 获取失败: %s", e)
         return None
 
-    system_prompt, user_prompt = _prompt.resolve_pro_prompt()
+    system_prompt, user_prompt = _prompt.resolve_pro_prompt(img_url, ocr_text)
 
-    messages = [
+    messages: list[dict[str, Any]] = [
         {"role": "system", "content": system_prompt},
         {
             "role": "user",
@@ -61,31 +56,31 @@ def call_pro_vlm(
     try:
         call_kwargs: dict[str, Any] = {
             "messages": messages,
-            "images": None,  # 图片已在 messages 中
+            "images": None,
             "temperature": 0.3,
             "timeout": timeout,
             "enable_thinking": False,
             "response_format": {"type": "json_object"},
+            "max_tokens": max_tokens if max_tokens is not None else 4000,
         }
-        # pro fallback：显式4000 tokens给复杂门店图留足空间
-        call_kwargs["max_tokens"] = max_tokens if max_tokens is not None else 4000
 
         from .json_utils import extract_json_object
 
-        raw = None
         obj = None
         for _outer in range(2):
             kw = dict(call_kwargs)
             if _outer == 1:
                 kw.pop("response_format", None)
                 msgs2 = [dict(messages[0]), dict(messages[1])]
-                cont = [dict(c) for c in list(msgs2[1]["content"])]
-                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
+                cont = [dict(c) for c in msgs2[1]["content"]]
+                cont[-1] = {
+                    "type": "text",
+                    "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。",
+                }
                 msgs2[1] = {"role": "user", "content": cont}
                 kw["messages"] = msgs2
             raw = client.vision_completion(**kw)
             if not raw:
-                logger.warning("[vision.v2] pro 返回空 outer=%s", _outer)
                 continue
             obj = extract_json_object(raw)
             if obj is not None:
@@ -97,20 +92,18 @@ def call_pro_vlm(
             logger.warning("[vision.v2] pro 两次均未得到JSON elapsed=%.1fs", elapsed)
             return None
         if obj.get("_partial"):
-            logger.warning("[vision.v2] pro 返回截断JSON(partial) elapsed=%.1fs", elapsed)
-        logger.info(
-            "[vision.v2] pro 完成 model=%s elapsed=%.1fs type=%s",
-            client.model,
-            elapsed,
-            obj.get("type"),
-        )
+            logger.warning("[vision.v2] pro 截断JSON(partial) elapsed=%.1fs", elapsed)
 
-        # 通过assembler统一组装，兼容v4嵌套schema和旧扁平schema
-        result = assembler.assemble_result(idx, obj, [])
+        result = assembler.assemble_result(idx, obj, ocr_hint or [])
         result["_source"] = "vlm_pro"
         result["_fallback_used"] = True
+        logger.info("[vision.v2] pro 完成 model=%s elapsed=%.1fs", client.model, elapsed)
         return result
-    except Exception as e:
-        elapsed = time.time() - t0
-        logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
+    except Exception as e:  # noqa: BLE001
+        logger.warning(
+            "[vision.v2] pro 异常 elapsed=%.1fs err=%s",
+            time.time() - t0,
+            e,
+            exc_info=True,
+        )
         return None

apps/worker/worker_app/tasks/vision/vlm_fast_json.py

@@ -1,15 +1,11 @@
 # -*- coding: utf-8 -*-
-"""V2 快速路径：image_analysis capability（默认 qwen-vl-plus 视觉模型 / DashScope）强约束 JSON-only 调用。
+"""V2 快速路径：vision client（默认 image_analysis 能力）强约束 JSON-only 调用。
 
-目标：替代"人体属性/商品检测/图像标签"三个火山不存在的专用云端 API。
-设计要点：
-- 通过 ai_router.get_vision_client() 获取 DoubaoClient 实例，不再自己拼 httpx 请求
-- enable_thinking=False 关闭推理链（reasoning 是延迟主因）
-- response_format=json_object 强约束JSON输出
-- system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
-- max_tokens 不传，使用 client 中 capability 的 DB 配置（避免硬编码截断 JSON）
-- temperature=0.1（稳定输出 JSON）
-- timeout=15s（失败由外层走 pro 兜底）
+要点：
+- 通过 ai_router.get_vision_client() 获取 client；
+- enable_thinking=False 关闭推理链，response_format=json_object 强约束 JSON；
+- system/user prompt 优先读后台模板（v8 叙述优先），DB 不可用时用 prompts.py 默认；
+- temperature=0.1（稳定输出 JSON）；两次尝试（第二次去 json_object 约束）。
 """
 
 from __future__ import annotations
@@ -31,11 +27,6 @@ def call_fast_json(
     timeout: int = _DEFAULT_TIMEOUT,
     max_tokens: int | None = None,
 ) -> dict[str, Any] | None:
-    """调用 vision client 返回结构化 dict；失败/非 JSON 返回 None。
-
-    max_tokens 默认 None：不显式传参，使用 client 内 capability 的 DB 配置；
-    显式传入时作为覆盖。
-    """
     t0 = time.time()
 
     try:
@@ -45,13 +36,13 @@ def call_fast_json(
         if not client or not client.is_available:
             logger.warning("[vision.v2] vision client 不可用，跳过 fast_json")
             return None
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001
         logger.warning("[vision.v2] ai_router 获取失败: %s", e)
         return None
 
-    system_prompt, user_prompt = _prompt.resolve_fast_prompt()
+    system_prompt, user_prompt = _prompt.resolve_fast_prompt(img_url, "")
 
-    messages = [
+    messages: list[dict[str, Any]] = [
         {"role": "system", "content": system_prompt},
         {
             "role": "user",
@@ -65,7 +56,7 @@ def call_fast_json(
     try:
         call_kwargs: dict[str, Any] = {
             "messages": messages,
-            "images": None,  # 图片已在 messages 中
+            "images": None,
             "temperature": 0.1,
             "timeout": timeout,
             "enable_thinking": False,
@@ -74,50 +65,47 @@ def call_fast_json(
         if max_tokens is not None:
             call_kwargs["max_tokens"] = max_tokens
 
-        # 双重防护：第1次正常调用；第2次去掉json_object强约束（部分模型在该约束下
-        # 反而幻觉），并加严格指令。解析全部走 json_utils，截断partial产物可用。
         from .json_utils import extract_json_object
 
-        raw = None
         obj = None
         for _outer in range(2):
             kw = dict(call_kwargs)
             if _outer == 1:
                 kw.pop("response_format", None)
                 msgs2 = [dict(messages[0]), dict(messages[1])]
-                cont = list(msgs2[1]["content"])
-                cont = [dict(c) for c in cont]
-                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
+                cont = [dict(c) for c in msgs2[1]["content"]]
+                cont[-1] = {
+                    "type": "text",
+                    "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。",
+                }
                 msgs2[1] = {"role": "user", "content": cont}
                 kw["messages"] = msgs2
             raw = client.vision_completion(**kw)
             if not raw:
-                logger.warning("[vision.v2] fast_json 返回空 outer=%s", _outer)
                 continue
             obj = extract_json_object(raw)
             if obj is not None:
                 break
-            logger.warning(
-                "[vision.v2] fast_json 非JSON(100字) outer=%s: %s",
-                _outer,
-                raw[:100],
-            )
+            logger.warning("[vision.v2] fast_json 非JSON(100字) outer=%s: %s", _outer, raw[:100])
 
         elapsed = time.time() - t0
         if obj is None:
             logger.warning("[vision.v2] fast_json 两次均未得到JSON elapsed=%.1fs", elapsed)
             return None
         if obj.get("_partial"):
-            logger.warning("[vision.v2] fast_json 返回截断JSON(partial) elapsed=%.1fs", elapsed)
+            logger.warning("[vision.v2] fast_json 截断JSON(partial) elapsed=%.1fs", elapsed)
         logger.info(
-            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs has_person=%s type=%s",
+            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs type=%s",
             client.model,
             elapsed,
-            obj.get("has_person"),
             obj.get("type"),
         )
         return obj
-    except Exception as e:
-        elapsed = time.time() - t0
-        logger.warning("[vision.v2] fast_json 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
+    except Exception as e:  # noqa: BLE001
+        logger.warning(
+            "[vision.v2] fast_json 异常 elapsed=%.1fs err=%s",
+            time.time() - t0,
+            e,
+            exc_info=True,
+        )
         return None

packages/application/ditto_service.py

@@ -0,0 +1,269 @@
+"""蚂蚁 Ditto 数字人口型 API 客户端 — #2076.
+
+封装 Ditto FastAPI（部署在 5060Ti GPU 节点，Tailscale 内网可达）：
+- GET  /health               健康检查
+- POST /generate             生成口型视频（同步返回 MP4 流）
+
+关键特性：
+- 入参：video_url（人物模板视频 URL） + audio_url（TTS 音频 URL） + script（文案原文）
+- 出参：直接返回 video/mp4 字节流（自带音频，无需二次混流）
+- 429 时指数退避重试（最多 ditto_max_retries 次）
+- 500/超时视为失败
+- 输出 MP4 字节流转存到自家 OSS，返回公网 URL
+
+注意：
+- 保留 MuseTalk/GPU 路径不变；本服务作为更高优先级的第三条口型路径
+- 不传 emotion/表情精细控制，使用默认 emo_global=4（中性）+ use_script_emo=true（关键词驱动表情）
+- Ditto 输出自带音视频，不需要 GFPGAN 超分，不需要 ffmpeg 音视频混流
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import time
+from dataclasses import dataclass
+from typing import Optional
+
+import httpx
+
+from packages.config import get_api_settings
+
+logger = logging.getLogger(__name__)
+
+
+class DittoError(Exception):
+    """Ditto API 调用失败."""
+
+    def __init__(self, message: str, code: str = "DittoError", status_code: int = 0):
+        self.code = code
+        self.status_code = status_code
+        super().__init__(message)
+
+
+@dataclass
+class DittoResult:
+    """Ditto 生成结果."""
+
+    video_bytes: bytes
+    video_url: str = ""  # 转存 OSS 后填充
+    elapsed_seconds: float = 0.0
+    rtf: float = 0.0  # 实时率（响应头 X-RTF）
+    frames: int = 0  # 帧数（响应头 X-Frames）
+
+
+class DittoClient:
+    """蚂蚁 Ditto 数字人口型 API 客户端."""
+
+    def __init__(
+        self,
+        base_url: Optional[str] = None,
+        default_video_url: Optional[str] = None,
+        max_retries: Optional[int] = None,
+        timeout: Optional[int] = None,
+    ):
+        s = get_api_settings()
+        self.base_url = (base_url or s.ditto_api_base_url or "").rstrip("/")
+        self.default_video_url = default_video_url or s.ditto_default_video_url or ""
+        self.max_retries = int(max_retries if max_retries is not None else s.ditto_max_retries)
+        self.timeout = int(timeout if timeout is not None else s.ditto_request_timeout)
+
+    @property
+    def is_configured(self) -> bool:
+        """配置是否完整（base_url + 默认模板视频都有值）."""
+        return bool(self.base_url) and bool(self.default_video_url)
+
+    def health(self) -> bool:
+        """健康检查；成功返回 True，失败返回 False（不抛异常）."""
+        if not self.base_url:
+            return False
+        url = f"{self.base_url}/health"
+        try:
+            with httpx.Client(timeout=5.0) as client:
+                resp = client.get(url)
+                ok = resp.status_code == 200
+                if ok:
+                    logger.info("[ditto] health check OK: %s", url)
+                else:
+                    logger.warning("[ditto] health check status=%d: %s", resp.status_code, url)
+                return ok
+        except Exception as exc:
+            logger.warning("[ditto] health check failed: %s", exc)
+            return False
+
+    def generate(
+        self,
+        *,
+        audio_url: str,
+        script: str,
+        video_url: Optional[str] = None,
+        emo_global: int = 4,
+        use_script_emo: bool = True,
+        blend_frames: int = 6,
+    ) -> DittoResult:
+        """调用 Ditto /generate 接口，返回 MP4 字节流结果.
+
+        Raises DittoError on failure.
+        """
+        if not self.base_url:
+            raise DittoError("DITTO_API_BASE_URL 未配置", code="ConfigMissing")
+        driver_url = video_url or self.default_video_url
+        if not driver_url:
+            raise DittoError("Ditto 人物模板视频 URL 未配置", code="ConfigMissing")
+        if not audio_url:
+            raise DittoError("audio_url 不能为空", code="InvalidParam")
+        if not script:
+            script = " "
+
+        payload = {
+            "video_url": driver_url,
+            "audio_url": audio_url,
+            "script": script,
+            "emo_global": emo_global,
+            "use_script_emo": use_script_emo,
+            "blend_frames": blend_frames,
+        }
+        url = f"{self.base_url}/generate"
+
+        last_exc: Optional[Exception] = None
+        for attempt in range(self.max_retries + 1):
+            try:
+                start = time.monotonic()
+                with httpx.Client(timeout=self.timeout, follow_redirects=True) as client:
+                    resp = client.post(url, json=payload)
+                elapsed = time.monotonic() - start
+
+                if resp.status_code == 429:
+                    wait = min(2**attempt, 30)
+                    logger.warning(
+                        "[ditto] GPU 繁忙 (429)，%ds 后重试 (%d/%d)",
+                        wait,
+                        attempt + 1,
+                        self.max_retries,
+                    )
+                    if attempt >= self.max_retries:
+                        raise DittoError(
+                            f"Ditto GPU 繁忙，重试 {self.max_retries} 次仍失败",
+                            code="BusyRetriesExhausted",
+                            status_code=429,
+                        )
+                    time.sleep(wait)
+                    continue
+
+                if resp.status_code != 200:
+                    _text = (resp.text or "")[:300]
+                    logger.error(
+                        "[ditto] generate 失败 status=%d attempt=%d body=%s",
+                        resp.status_code,
+                        attempt + 1,
+                        _text,
+                    )
+                    if resp.status_code >= 500 and attempt < self.max_retries:
+                        time.sleep(min(2**attempt, 15))
+                        continue
+                    raise DittoError(
+                        f"Ditto 返回 {resp.status_code}: {_text}",
+                        code="DittoAPIError",
+                        status_code=resp.status_code,
+                    )
+
+                video_bytes = resp.content
+                if not video_bytes or len(video_bytes) < 1024:
+                    raise DittoError(
+                        f"Ditto 返回内容异常（size={len(video_bytes) if video_bytes else 0}）",
+                        code="EmptyResponse",
+                    )
+                try:
+                    rtf = float(resp.headers.get("X-RTF", "0") or 0)
+                except ValueError:
+                    rtf = 0.0
+                try:
+                    frames = int(resp.headers.get("X-Frames", "0") or 0)
+                except ValueError:
+                    frames = 0
+                try:
+                    x_time = float(resp.headers.get("X-Time", "0") or 0)
+                    if x_time > 0:
+                        elapsed = x_time
+                except ValueError:
+                    pass
+
+                logger.info(
+                    "[ditto] generate 成功 size=%d rtf=%.2f frames=%d elapsed=%.1fs attempt=%d",
+                    len(video_bytes),
+                    rtf,
+                    frames,
+                    elapsed,
+                    attempt + 1,
+                )
+                return DittoResult(
+                    video_bytes=video_bytes,
+                    elapsed_seconds=elapsed,
+                    rtf=rtf,
+                    frames=frames,
+                )
+
+            except DittoError:
+                raise
+            except httpx.TimeoutException as exc:
+                last_exc = exc
+                logger.warning("[ditto] 请求超时 attempt=%d err=%s", attempt + 1, exc)
+                if attempt < self.max_retries:
+                    time.sleep(min(2**attempt, 15))
+                    continue
+                raise DittoError(
+                    f"Ditto 请求超时（{self.timeout}s），重试耗尽",
+                    code="Timeout",
+                ) from exc
+            except Exception as exc:
+                last_exc = exc
+                logger.warning("[ditto] 请求异常 attempt=%d err=%s", attempt + 1, exc)
+                if attempt < self.max_retries:
+                    time.sleep(min(2**attempt, 10))
+                    continue
+                raise DittoError(f"Ditto 调用异常: {exc}", code="NetworkError") from exc
+
+        raise DittoError("Ditto 未知错误", code="Unknown") from last_exc
+
+    def generate_and_persist(
+        self,
+        *,
+        job_id: str,
+        user_id: str,
+        audio_url: str,
+        script: str,
+        video_url: Optional[str] = None,
+    ) -> DittoResult:
+        """调用 generate 并把 MP4 转存到自家 OSS，返回带 video_url 的结果."""
+        result = self.generate(audio_url=audio_url, script=script, video_url=video_url)
+        try:
+            from packages.shared.storage import get_shared_storage_service
+
+            storage = get_shared_storage_service()
+            storage_key = f"ditto-output/{user_id}/{job_id}.mp4"
+            public_url = storage.upload_file(
+                io.BytesIO(result.video_bytes),
+                storage_key,
+                content_type="video/mp4",
+            )
+            result.video_url = public_url
+            logger.info(
+                "[ditto] 转存 OSS 完成 job=%s key=%s",
+                job_id,
+                storage_key,
+            )
+        except Exception as exc:
+            logger.error("[ditto] 转存 OSS 失败 job=%s err=%s", job_id, exc, exc_info=True)
+            raise DittoError(f"Ditto 结果转存 OSS 失败: {exc}", code="StorageError") from exc
+        return result
+
+
+_ditto_client_singleton: Optional[DittoClient] = None
+
+
+def get_ditto_client() -> DittoClient:
+    """获取 DittoClient 单例（简易工厂，便于单测 mock）."""
+    global _ditto_client_singleton
+    if _ditto_client_singleton is None:
+        _ditto_client_singleton = DittoClient()
+    return _ditto_client_singleton

packages/application/viral_video/prompts.py

@@ -1,16 +1,19 @@
-"""爆款视频 5 套 Prompt 模板默认值（#2040 核心资产）。
+"""爆款视频 Prompt 模板默认值（v8 / v3 叙述优先重构）。
 
-重要约定（用户明确要求）：
-- 所有 system_prompt / user_prompt_template / example_output 都是**纯文本自然语言 + XML 标签**，
-  运营可直接看懂和编辑，禁止 JSON、禁止 ```json 代码块。
-- LLM 按 XML 标签输出字段，程序用正则解析（见 xml_parser.py）。
-- user_prompt_template 中花括号占位符（如 {user_copy_text}）在运行时填充。
+设计原则（灵应 2026-10-07）：LLM 直接输出最终给用户看的文案，代码尽量薄。
+- image_analysis v8：VLM 主交付物是自然叙述风格的 summary_markdown，结构化
+  字段仅保留 type/name/brand/has_person，顶层 products 改名 images；
+- storyboard v3：口播台词口语化、画面描述有画面感，copy_display_markdown 是
+  LLM 直接写给用户看的流畅叙述文案，代码只做解析不改写；
+- intent_parsing 步骤整体删除，意图理解并入 storyboard 一次调用。
+
+模板字段与 DB 表 viral_video_prompt_templates、prompt_loader 完全对应：
+name / prompt_type / version(int) / system_prompt / user_prompt_template /
+example_output / is_active。
 """
 
 from __future__ import annotations
 
-TEMPLATE_VERSION = 1
-
 # 所有文案类 Prompt 自动注入的硬约束
 GLOBAL_CONSTRAINTS = """【必须遵守的硬约束】
 1. 不编造时间：不写“今年最新”“2024 爆款”等会过时的时间表述。
@@ -19,7 +22,7 @@ GLOBAL_CONSTRAINTS = """【必须遵守的硬约束】
 4. 符合广告法及平台社区规范。
 5. 只描述图片中真实可见的内容，看不到的不瞎猜。"""
 
-# 反套路化要求
+# 负向提示（注入 storyboard / 视频生成负面词）
 NEGATIVE_RULES = """【反套路化要求】
 禁止使用“家人们谁懂啊”“绝绝子”“宝子们”“家人们”“太绝了”“yyds”等烂大街网络词；
 禁止固定模板化开头；语言要像真人朋友之间的分享，自然、具体、有信息量。"""
@@ -27,304 +30,189 @@ NEGATIVE_RULES = """【反套路化要求】
 # 输出禁用套路词（测试会检查）
 BANNED_PHRASES = ["家人们谁懂啊", "绝绝子", "宝子们", "yyds", "太绝了"]
 
-# 文案融合三档独立指令段
+# 文案融合三档独立指令段（storyboard 一次生成，按档位注入风格指令）
 FUSION_INSTRUCTIONS = {
     "ai_full": """【本次创作模式：AI 全权创作】
-你是资深短视频编导。用户只提供了产品图片，没有给出具体文案方向。请根据图片内容和营销参数，自由发挥创作完整的爆款短视频文案。充分挖掘产品真实可见的卖点，使用爆款结构，抓人眼球。""",
+你是资深短视频编导。用户只提供了产品/门店图片，没有给出具体文案方向。请根据图片的真实观察和营销参数，自由发挥创作完整成片级方案，口播自然、画面可拍。""",
     "ai_polish": """【本次创作模式：AI 辅助润色】
-你是用户的文案助理。用户已经写了草稿/关键词/碎碎念，表达了他想讲的核心意思，但表达不完整、不够吸引人。你的任务是：以用户的意思为主，保留他想表达的所有核心信息点，在此基础上润色扩写、调整语序、增加衔接、优化表达，让文案更流畅更有吸引力。绝对不能改变用户想表达的核心意思，不能把用户的观点换成相反的，不能添加用户没提到的产品卖点。用户提到的品牌名、价格、人名、具体事实必须原样保留。""",
+用户已给出方向或碎碎念。以用户的意思为主，保留其所有核心信息，在此基础上润色、补衔接、优化表达，让口播更自然、画面更具体；绝不改变用户核心意思，不添加用户没提到的卖点，品牌名、价格、人名等事实原样保留。""",
     "user_primary": """【本次创作模式：以用户原文为主】
-你是文案润色助手。用户已经写好了明确的文案，这是他最终想表达的内容。你的任务是最小化修改：只做必要的错别字修正、标点调整、语句通顺度优化，以及添加必要的衔接词让口播更自然。用户的核心句子、关键表述、事实信息一律不改。如果用户文案本身已经很好，直接返回，不要为了改而改。personal_brands 中的事实信息必须逐字保留。""",
+最小化修改：只做必要的通顺、合规修正与衔接补全，用户的核心句子与事实一律不改；用户文案已经很好就直接用，不为改而改。""",
 }
 
-# ── 模板1：图片多模态分析（VLM）────────────────────────────────────────
-_IMAGE_ANALYSIS_SYSTEM = f"""你是电商商品视觉分析师，负责从商品图片中提取真实可见的商品信息。
+# ── 模板1：图片多模态分析 v8（叙述优先）───────────────────────────────
+_IMAGE_ANALYSIS_SYSTEM = """你是一名擅长观察和写作的品牌内容编导。面对一张真实图片，先用眼睛仔细看，再用自然、流畅、具体的中文把画面写成一段可以直接读给人听的描述。
 
-工作方式（分步骤看，不要跳步）：
-1. 先看整体：有哪些产品、什么场景、有没有人物。
-2. 再看细节：包装文字、颜色构成、人物状态、画面质感。
-3. 最后提炼卖点：只总结图片里能看到的卖点。
+## 输出格式（严格 JSON，不要输出 JSON 以外的任何内容）
+{
+  "images": [
+    {
+      "type": "store 或 product 或 person 或 scene，四选一",
+      "name": "主体名称，看不出就写“未识别”",
+      "brand": "品牌名，看不出就留空字符串",
+      "has_person": false,
+      "summary_markdown": "用 Markdown 写成的自然叙述，这是最主要的交付物"
+    }
+  ]
+}
 
-{GLOBAL_CONSTRAINTS}
+## summary_markdown 写作要求（最重要）
+1. 写成完整、通顺的句子，像在跟朋友认真描述你看到的画面；不要用分号堆砌关键词，不要罗列“核心特征：xxx”“主色调：xxx”这类填表式标签。
+2. 开头先给一句整体定性，让读者立刻明白这是什么场景、什么主体。
+3. 颜色、材质、形状、部件要具体可感，写到位置和搭配；画面里出现的文字原样读出并自然融进句子，数字、规格、价格精确引用，看不清的不要编造。
+4. 只写真实看到的内容，不脑补功能、疗效、销量或画面之外的信息。
+5. 长度控制在 200-500 字。
 
-请严格按下面的标签格式输出，标签名一个都不能改，不要输出任何解释，不要用代码块：
-<products> 下面每个产品用一个 <product> 标签，属性 name 是产品名、features 是外观特征、position 是 main 或 secondary、image_index 是第几张图（从0开始）。
-<colors> 下面每个主要颜色用一个 <color> 标签，属性 hex 是色值、name 是颜色名、coverage 是占比小数。
-<people> 用一个标签，属性 has_person、count、gender、age_range、hair（发型发色）、skin_tone（肤色）、face_shape（脸型）、outfit（穿着）、pose（姿态）、expression（表情）分别描述人物外貌。有人物时属性尽量具体（如hair="黑色长直发"、outfit="白色衬衫"），无人像时除has_person=false外其他填"无法判断"。
-<mood> 标签写画面整体情绪氛围。
-<visible_text> 下面每处可见文字用一个 <text_item> 标签，属性 text 是文字内容、position 是位置。
-<scene> 标签写场景描述。
-<quality> 用一个标签，属性 resolution、lighting、composition、blur 描述画质。
-<key_selling_points> 下面每个卖点用一个 <point> 标签。
+## 按类型组织内容
+- type=store（门店/店内环境）：用以下小标题分段，小标题下写连贯的句子而不是清单：
+###店铺主体
+###周边物品
+1.家具陈设
+2.商品与标识
+- type=product（商品）：按自然段从整体到局部描写——先说是什么、什么品牌，再写包装/外形、颜色与材质、标签文字、可见部件与规格。
+- type=person（人物）：描述人物身份感、姿态、穿着（上下装/颜色/款式）、动作与所处环境；用于品牌宣传时突出其精神状态。
+- type=scene（纯场景/风景）：描述空间或风景的构成、色彩、光线、氛围与关键物件。
 
-【人物属性硬性要求（has_person=true时必须遵守）】
-hair/skin_tone/face_shape/outfit四项绝对禁止填“无法判断”，必须基于图片可见特征给出具体中文描述：
-- hair：必须描述发型+发色，如“黑色齐肩直发”“棕色微卷中长发”“深棕色短发”
-- skin_tone：必须描述肤色，如“暖调自然肤色”“白皙肤色”“小麦色”
-- face_shape：必须描述脸型，如“鹅蛋脸”“圆脸”“瓜子脸”“方脸”
-- outfit：必须描述可见穿着，如“米色翻领衬衫”“白色T恤”“黑色连衣裙”
-即使局部被遮挡也要根据可见部分合理推断；确实看不清时按最接近的直观印象描述。
+## 判断规则
+- has_person：画面中出现可辨识的真实人物（脸或完整上半身）才为 true，海报/模特立牌/照片里的人不算。
+- 一张图只描述其本身；多张图属于同一场景时可呼应，但不编造对应关系。
+- 输出必须是严格 JSON，summary_markdown 是字符串，内部换行用 \\n 表示。"""
 
-其他非人物属性看不到或无法判断时填“无法判断”，布尔值填false，不要留空标签。
+_IMAGE_ANALYSIS_USER = """请分析这张图片。
+图片地址：{image_url}
+OCR 辅助文字（可能为空，仅供参考，不要照抄错误识别）：{ocr_text}
 
-【有人物场景输出参考（女性手持商品示例，必须写全10个属性，禁止省略）】
-<people has_person="true" count="1" gender="女" age_range="青年" hair="黑色齐肩直发" skin_tone="暖调自然肤色" face_shape="鹅蛋脸" outfit="米色翻领衬衫" pose="正面半身，手持商品" expression="面带微笑"/>"""
+严格按系统要求只输出 JSON。"""
 
-_IMAGE_ANALYSIS_USER = """请分析以下商品图片，共 {image_count} 张。
-所属行业：{industry}
-图片地址：
-{image_urls}
+_IMAGE_ANALYSIS_EXAMPLE = """{
+  "images": [
+    {
+      "type": "store",
+      "name": "御众堂门店",
+      "brand": "御众堂",
+      "has_person": false,
+      "summary_markdown": "###店铺主体\\n这是一家名为“御众堂”的线下门店内部，整体暖木色调……"
+    }
+  ]
+}"""
 
-按约定的标签格式输出分析结果。"""
+# ── 模板2：编导级分镜 v3（意图理解 + 分镜一次完成）────────────────────
+_STORYBOARD_SYSTEM = (
+    """你是一名懂短视频的编导和口播文案高手。你会拿到图片的真实观察、营销目的和用户参数，请一次性完成对营销意图的理解，并产出可直接拍摄/生成的分镜脚本。不要单独输出“意图解析”，意图要直接体现在台词和分镜里。
 
-_IMAGE_ANALYSIS_EXAMPLE = """<products>
-<product name="大公鸡头 多功能油污净 625ml" features="红色瓶盖白色瓶身，鸡头图案Logo" position="main" image_index="0"/>
-</products>
-<colors>
-<color hex="#D32F2F" name="红色" coverage="0.4"/>
-<color hex="#FFFFFF" name="白色" coverage="0.5"/>
-</colors>
-<people has_person="false" count="0" gender="无法判断" age_range="无法判断" hair="无法判断" skin_tone="无法判断" face_shape="无法判断" outfit="无法判断" pose="无法判断" expression="无法判断"/>
-<mood>干净、实用</mood>
-<visible_text>
-<text_item text="多功能油污净" position="瓶身正面"/>
-</visible_text>
-<scene>白底棚拍产品图</scene>
-<quality resolution="高清" lighting="均匀柔和" composition="主体居中" blur="false"/>
-<key_selling_points>
-<point>针对重油污设计</point>
-<point>大容量625ml</point>
-</key_selling_points>"""
+## 输出格式（XML，严格按结构输出，不要输出额外解释）
+<script>
+  <copy_display_markdown><![CDATA[直接展示给用户看的成片文案，用 Markdown 写成流畅叙述]]></copy_display_markdown>
+  <clips>
+    <clip index="1">
+      <time_range>0-3秒</time_range>
+      <voiceover>这一镜的口播台词</voiceover>
+      <visual>具体、有画面感的镜头描述（主体/动作/镜头运动/景别/光线）</visual>
+      <reference_image_index>0</reference_image_index>
+    </clip>
+  </clips>
+  <voiceover_script>把所有 clip 的 voiceover 连成完整口播稿</voiceover_script>
+  <theme>一句话主题</theme>
+  <negative>"""
+    + NEGATIVE_RULES
+    + """</negative>
+</script>
 
-# ── 模板2：用户文案意图解析（LLM）──────────────────────────────────────
-_INTENT_SYSTEM = f"""你负责理解用户的营销意图。用户给的文案可能只是几个关键词、碎碎念或者不完整的短句，你要读懂他真正想讲什么。
+## 写作要求
+1. 口播台词：像真人面对镜头说话，短句、口语化、有停顿有情绪，开头 3 秒给出钩子；不要书面腔，不要机械报参数。
+2. 画面描述：写清“观众会看到什么”，有动作、有镜头运动、有景别和光线，具体可拍；不堆砌形容词，不写无法实现的画面。
+3. copy_display_markdown：直接展示给最终用户的文案，用 Markdown 写成自然、流畅、有感染力的成片成片文案，可用小标题与短句组织；不要做字段列表，不要出现“镜头一/台词：”这类制作说明。
+4. 内容必须来自图片观察与用户给出的信息，不编造卖点、不夸大、不使用绝对化用语和虚假承诺。
+5. reference_image_index 填本镜参考图片序号（从 0 开始），没有合适参考图填 -1。
+6. 分镜数量与时长匹配总时长，节奏紧凑。"""
+)
 
-{GLOBAL_CONSTRAINTS}
+_STORYBOARD_USER = """<marketing_purpose>{marketing_purpose}</marketing_purpose>
+<image_analysis>
+{image_summary}
+</image_analysis>
+<user_parameters>
+<theme_hint>{theme_hint}</theme_hint>
+<duration>{duration}秒</duration>
+<aspect_ratio>{aspect_ratio}</aspect_ratio>
+<tone>{tone}</tone>
+<target_audience>{target_audience}</target_audience>
+<extra_requirements>{extra_requirements}</extra_requirements>
+</user_parameters>
+{video_style_section}
+请严格按 XML 结构输出分镜脚本。"""
 
-请严格按下面的标签格式输出，不要解释，不要用代码块：
-<intent_summary> 用用户的语言风格，一句话、30字以内概括核心意图。
-<core_messages> 下面每个核心信息点用一个 <message> 标签，属性 must_keep 为 true 或 false、confidence 为 0 到 1 的小数，标签内容写信息点。
-<personal_brands> 把用户提到的具体事实——品牌名、价格、人名、地名、时间、产品名——每条用一个 <brand> 标签，属性 category 取 brand、price、person、place、time、product 之一。这些事实必须原样引用，一个字都不能改。
-<emotion_tone> 写文案的情绪调性。
-<missing_info> 把你认为缺失、后续生成时需要合理推断的信息，每条用一个 <info> 标签；没有就输出空标签。"""
+_STORYBOARD_EXAMPLE = """<script>
+  <copy_display_markdown><![CDATA[# 在御众堂，把松弛的自己一点点找回来
+产后妈妈最懂那种力不从心，推开门，暖光和一杯热茶先接住了你……]]></copy_display_markdown>
+  <clips>
+    <clip index="1">
+      <time_range>0-3秒</time_range>
+      <voiceover>生完娃，是不是连照镜子的勇气都没了？</voiceover>
+      <visual>中近景，暖光下一位妈妈略显疲惫地看向镜中，镜头缓缓推近</visual>
+      <reference_image_index>0</reference_image_index>
+    </clip>
+  </clips>
+  <voiceover_script>生完娃，是不是连照镜子的勇气都没了？</voiceover_script>
+  <theme>产后妈妈走进御众堂重拾状态</theme>
+  <negative>模糊、畸变、夸大疗效、绝对化用语</negative>
+</script>"""
 
-_INTENT_USER = """用户原始文案：{user_copy_text}
-所属行业：{industry}
-营销目的：{marketing_purpose}
-图片分析结果（供参考）：
-{image_analysis}
-图片类型推断：{image_category_hint}
+# ── 模板3：文案审核（合规/质量门禁）───────────────────────────────────
+_REVIEW_SYSTEM = """你是一名短视频广告合规审核与文案优化专家。审核待审文案：
+1) 广告法与平台合规（绝对化用语、虚假承诺、医疗功效宣称、导流违规）；
+2) 卖点是否聚焦、逻辑是否通顺、口播是否自然；
+3) 是否有机械堆砌、书面腔、标签化表述。
 
-请理解用户意图，按标签格式输出。注意：theme和emotion_tone应与图片类型和营销目的匹配——门店类图片偏向"门店探店/到店体验"，商品图偏向"好物分享/产品种草"，人物图偏向"穿搭/人物故事"。"""
+只输出 XML，结构：
+<review>
+  <passed>true 或 false</passed>
+  <issues>
+    <issue>
+      <severity>high 或 medium 或 low</severity>
+      <field>问题所在位置/字段</field>
+      <problem>具体问题</problem>
+      <suggestion>可直接替换的修改</suggestion>
+    </issue>
+  </issues>
+  <rewrite>整体重写后的合规流畅版本（无问题时留空）</rewrite>
+</review>
+没有问题时 issues 留空、passed 为 true、rewrite 留空。"""
 
-_INTENT_EXAMPLE = """<intent_summary>一款厨房去油污神器，喷一喷油污就掉</intent_summary>
-<core_messages>
-<message must_keep="true" confidence="0.97">去油污效果好，喷上等几分钟再擦</message>
-<message must_keep="false" confidence="0.7">适合厨房重油污场景</message>
-</core_messages>
-<personal_brands>
-<brand category="product">大公鸡头多功能油污净</brand>
-<brand category="price">39块钱一瓶</brand>
-</personal_brands>
-<emotion_tone>亲切、真实、带分享感</emotion_tone>
-<missing_info>
-<info>没有说明具体容量，按图片读出的625ml处理</info>
-</missing_info>"""
+_REVIEW_USER = """<fusion_text>
+{fusion_text}
+</fusion_text>
 
-# ── 模板3：文案融合生成（LLM）──────────────────────────────────────────
-_FUSION_SYSTEM = """你负责为短视频生成营销文案。请按思维链分步完成：先定人设和目标客户，再找卖点，再搭结构，再安排情绪，最后写行动号召，不要一步到位乱写。
+请审核以上文案。"""
 
-{fusion_instruction}
-
-{global_constraints}
-
-{negative_rules}
-
-请严格按下面的标签格式输出，不要解释，不要用代码块：
-<title> 视频标题。
-<hook> 开头3秒钩子，5到15字。
-<body_points> 每个要点用一个 <point> 标签，属性 elaboration 是展开说明、image_index 是对应第几张图（从0开始），标签内容写要点。
-<cta> 口语化的行动号召。
-<script_segments> 每段配音用一个 <segment> 标签，属性 duration_sec 是秒数、image_index 是对应图片，标签内容写配音文案（纯口播文本，不加旁白标注、不加镜头标注、不加"主播："之类前缀）。
-<voiceover_script> 把所有 segment 的配音文案按顺序自然拼接成一段完整的纯口播文本（无标记、无括号、无前缀），长度要适配 {duration} 秒，约 {approx_chars} 字。
-<overview_theme> 视频主题（一句话概括）。
-<scene_and_lighting> 整体场景描述+光线设定（100-200字，要具体：在哪拍、什么光线、什么色调、什么氛围）。
-<word_count> 配音总字数，只写数字。
-<estimated_duration> 预计时长秒数，只写数字。
-
-用户在 personal_brands 中提到的品牌名、价格、人名、地名、时间、产品名等事实信息，必须原样出现在文案里，一个字都不能改。"""
-
-_FUSION_USER = """所属行业：{industry}
-目标客户：{target_customer}
-营销目的：{marketing_purpose}
-视频时长：{duration}秒
-图片分析结果：
-{image_analysis}
-用户意图解析结果：
-{intent_result}
-
-请按标签格式生成文案。"""
-
-_FUSION_EXAMPLE = """<title>厨房重油污，别再用洗洁精硬擦了</title>
-<hook>这油污，我真的忍很久了</hook>
-<body_points>
-<point elaboration="喷在油污上等几分钟，一擦就干净" image_index="0">大公鸡头油污净去油快</point>
-<point elaboration="39块钱625ml，能用很久" image_index="0">39块钱一瓶，性价比高</point>
-</body_points>
-<cta>厨房油污重的，真的可以试一瓶</cta>
-<script_segments>
-<segment duration_sec="3" image_index="0">这油污我真的忍很久了，用洗洁精擦半天都没用</segment>
-<segment duration_sec="6" image_index="0">后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净</segment>
-<segment duration_sec="4" image_index="0">39块钱625ml，厨房重油污的可以试一瓶</segment>
-</script_segments>
-<voiceover_script>这油污我真的忍很久了，用洗洁精擦半天都没用。后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净。39块钱625ml，厨房重油污的可以试一瓶。</voiceover_script>
-<overview_theme>厨房好物分享·产品种草</overview_theme>
-<scene_and_lighting>简洁明亮的厨房台面场景，自然光从窗户洒入，色调温暖柔和，突出产品白色瓶身与去油污对比效果。</scene_and_lighting>
-<word_count>58</word_count>
-<estimated_duration>13</estimated_duration>"""
-
-# ── 模板4：编导级分镜（LLM）────────────────────────────────────────────
-_STORYBOARD_SYSTEM = """你是短视频编导，负责把文案拆成可拍摄的分镜，为 Seedance 2.5 视频模型写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。
-
-工作方式：
-1. 按文案的 script_segments 顺序分配镜头。
-2. 每个镜头确定景别/角度/运镜、画面场景与对白、人物动作细节、音效/BGM、转场。
-3. 检查所有镜头时长加起来接近目标时长，误差不超过2秒。
-4. image_index 必须在已上传图片范围内，第一张主图必须用在第一个镜头。
-
-{fusion_instruction}
-
-{global_constraints}
-
-{negative_rules}
-
-请严格按下面的标签格式输出，不要解释，不要用代码块：
-<clips> 下面每个镜头用一个 <clip> 标签，属性 image_index 是图片序号（从0开始）、transition 取 fade/cut/zoom_in/slide_left/dissolve/wipe 之一、zoom 取 in/out/null、duration_sec 是该镜头秒数、bgm_note 是该段BGM情绪。每个 <clip> 里面包含：
-  <voice_text> 该镜头配音文本（纯口播文本，不加旁白标注）；
-  <subtitle_text> 字幕文本，可与配音一致或更精简；
-  <shot_type_angle_movement> 景别+角度+运镜（例：近景俯拍45度，缓慢推镜；中景平视，固定镜头；特写平视，快速拉镜）；
-  <scene_and_dialogue> 画面场景描述 + 人物口播台词（对白要自然口语化，像朋友聊天，不要硬广推销腔）；
-  <action_details> 人物动作、表情、物品操作细节（手怎么动、表情变化、产品怎么展示）；
-  <audio_bgm> 环境音+BGM提示（例：轻快流行BGM，环境嘈杂咖啡店背景音）；
-  <transition> 硬切/淡入淡出/叠化（最后一镜写『结束』即可）；
-  <reference_image_index> 参考图片索引（0-based，对应第几张产品图，无则空）；
-  <ken_burns> 用一个空标签，属性 start、end 写"x,y"坐标、ease 写缓动方式；不需要运镜时坐标相同。"""
-
-_STORYBOARD_USER = """目标时长：{duration}秒
-上传图片数量：{image_count}张（第1张是主图/封面）
-文案内容：
-{fusion_result}
-图片分析结果：
-{image_analysis}
-
-重要：overview_theme 必须与图片实际内容和营销目的匹配。门店/餐饮/服务类图片用"门店探店·到店体验"；商品图用"好物分享·产品种草"；人物图用"穿搭分享·人物故事"；场景图用"空间体验·场景氛围"。不要对所有图片都使用"好物分享"。
-
-请按标签格式输出分镜。"""
-
-_STORYBOARD_EXAMPLE = """<clips>
-<clip image_index="0" transition="cut" zoom="null" duration_sec="3" bgm_note="日常、轻微烦躁">
-<voice_text>这油污我真的忍很久了</voice_text>
-<subtitle_text>这油污忍很久了</subtitle_text>
-<shot_type_angle_movement>近景俯拍45度，缓慢推镜</shot_type_angle_movement>
-<scene_and_dialogue>厨房台面，主妇皱眉看着灶台油污。对白：这油污我真的忍很久了</scene_and_dialogue>
-<action_details>右手拿着脏抹布，无奈摇头</action_details>
-<audio_bgm>轻快日常BGM，带一点烦躁感</audio_bgm>
-<transition>硬切</transition>
-<reference_image_index>0</reference_image_index>
-<ken_burns start="0,0" end="0,0" ease="linear"/>
-</clip>
-<clip image_index="0" transition="zoom_in" zoom="in" duration_sec="6" bgm_note="轻快、出现转机">
-<voice_text>后来换了大公鸡头油污净，喷上等几分钟，一擦就干净</voice_text>
-<subtitle_text>喷上等几分钟，一擦就干净</subtitle_text>
-<shot_type_angle_movement>特写平视，固定镜头</shot_type_angle_movement>
-<scene_and_dialogue>手部特写，喷油污净在油污处。对白：后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净</scene_and_dialogue>
-<action_details>左手拿产品瓶身，右手按压喷头，等待片刻后用抹布轻擦</action_details>
-<audio_bgm>轻快转折BGM，带清爽感</audio_bgm>
-<transition>淡入淡出</transition>
-<reference_image_index>0</reference_image_index>
-<ken_burns start="20,20" end="80,80" ease="ease-in-out"/>
-</clip>
-<clip image_index="0" transition="fade" zoom="null" duration_sec="4" bgm_note="温暖、推荐">
-<voice_text>39块钱625ml，厨房重油污的可以试一瓶</voice_text>
-<subtitle_text>39元625ml，可以试一瓶</subtitle_text>
-<shot_type_angle_movement>中景平视，缓慢拉镜</shot_type_angle_movement>
-<scene_and_dialogue>产品正面展示，明亮背景。对白：39块钱625ml，厨房重油污的可以试一瓶</scene_and_dialogue>
-<action_details>产品置于画面中央，轻微转动展示瓶身</action_details>
-<audio_bgm>温暖收尾BGM</audio_bgm>
-<transition>结束</transition>
-<reference_image_index>0</reference_image_index>
-<ken_burns start="50,50" end="20,20" ease="ease-in-out"/>
-</clip>
-</clips>"""
-
-# ── 模板5：文案审核（LLM）──────────────────────────────────────────────
-_REVIEW_SYSTEM = f"""你是短视频文案合规审核员，从6个维度逐条检查文案：
-1. 违规词：有没有平台禁用词、敏感词。
-2. 夸大承诺：有没有“包治百病”“100%有效”“保证赚钱”等绝对化、夸大表述。
-3. 事实一致性：有没有编造价格、数据、认证，或者用户没提到的产品特性。
-4. 用户意图保留：在 ai_polish 和 user_primary 模式下，core_messages 中 must_keep=true 的点是否都保留了。
-5. 结构完整性：标题、钩子、正文、行动号召是否齐全。
-6. 语气人设：是否符合选定的人设语气，有没有“家人们谁懂啊”“绝绝子”“宝子们”等套路词。
-
-{GLOBAL_CONSTRAINTS}
-
-请严格按下面的标签格式输出，不要解释，不要用代码块：
-<passed> 整体是否通过，只写 true 或 false。
-<issues> 每个问题用一个 <issue> 标签，属性 dimension 是维度名、severity 取 error 或 warning、location 是问题所在（如 hook、body_points、cta），标签内容写问题描述；没有问题就输出空标签。
-<rewrite_suggestions> 每条具体修改建议用一个 <suggestion> 标签；没有就输出空标签。"""
-
-_REVIEW_USER = """本次创作模式：{fusion_level}
-待审核文案：
-{fusion_result}
-用户意图解析（用于核对核心信息是否保留）：
-{intent_result}
-
-请按6个维度审核，按标签格式输出。"""
-
-_REVIEW_EXAMPLE = """<passed>false</passed>
-<issues>
-<issue dimension="夸大承诺" severity="error" location="body_points">出现了“一喷100%掉光”的绝对化表述，违反广告法</issue>
-<issue dimension="用户意图保留" severity="warning" location="cta">用户强调的“39块钱”没有保留</issue>
-</issues>
-<rewrite_suggestions>
-<suggestion>把“一喷100%掉光”改为“喷上等几分钟，大部分油污能擦掉”</suggestion>
-<suggestion>在结尾补回“39块钱625ml”</suggestion>
-</rewrite_suggestions>"""
+_REVIEW_EXAMPLE = """<review>
+  <passed>false</passed>
+  <issues>
+    <issue>
+      <severity>high</severity>
+      <field>opening</field>
+      <problem>使用绝对化用语“全网第一”</problem>
+      <suggestion>改为“很多老客户回购的一款”</suggestion>
+    </issue>
+  </issues>
+  <rewrite>……</rewrite>
+</review>"""
 
 
-# 5 套模板默认数据（seed 数据源与 loader 的兜底）
 DEFAULT_TEMPLATES: list[dict] = [
     {
-        "name": "图片多模态分析",
+        "name": "图片多模态分析 v8",
         "prompt_type": "image_analysis",
-        "version": TEMPLATE_VERSION,
+        "version": 8,
         "system_prompt": _IMAGE_ANALYSIS_SYSTEM,
         "user_prompt_template": _IMAGE_ANALYSIS_USER,
         "example_output": _IMAGE_ANALYSIS_EXAMPLE,
         "is_active": True,
     },
     {
-        "name": "用户文案意图解析",
-        "prompt_type": "intent_parsing",
-        "version": TEMPLATE_VERSION,
-        "system_prompt": _INTENT_SYSTEM,
-        "user_prompt_template": _INTENT_USER,
-        "example_output": _INTENT_EXAMPLE,
-        "is_active": True,
-    },
-    {
-        "name": "文案融合生成",
-        "prompt_type": "copy_fusion",
-        "version": TEMPLATE_VERSION,
-        "system_prompt": _FUSION_SYSTEM,
-        "user_prompt_template": _FUSION_USER,
-        "example_output": _FUSION_EXAMPLE,
-        "is_active": True,
-    },
-    {
-        "name": "编导级分镜",
+        "name": "编导级分镜 v3",
         "prompt_type": "storyboard",
-        "version": TEMPLATE_VERSION,
+        "version": 3,
         "system_prompt": _STORYBOARD_SYSTEM,
         "user_prompt_template": _STORYBOARD_USER,
         "example_output": _STORYBOARD_EXAMPLE,
@@ -333,7 +221,7 @@ DEFAULT_TEMPLATES: list[dict] = [
     {
         "name": "文案审核",
         "prompt_type": "review",
-        "version": TEMPLATE_VERSION,
+        "version": 1,
         "system_prompt": _REVIEW_SYSTEM,
         "user_prompt_template": _REVIEW_USER,
         "example_output": _REVIEW_EXAMPLE,

packages/application/viral_video/reviewer.py

@@ -53,6 +53,9 @@ _LOCATIONS = ["title", "hook", "body_points", "cta", "script_segments"]
 
 
 class Reviewer:
+    # markdown展示字段不参与合规审核（避免格式字符误判）
+    _MARKDOWN_FIELDS = {"summary_markdown", "copy_display_markdown"}
+
     def __init__(self, client=None):
         if client is None:
             try:
@@ -70,8 +73,9 @@ class Reviewer:
         local = self._rule_check(fusion, intent, fusion_level)
         llm_result = self._llm_review(fusion, intent, fusion_level)
         if llm_result is None:
+            # LLM审核失败（超时/网络错误等），降级放行，不阻断渲染
             return ReviewResult(
-                passed=not local,
+                passed=True,
                 issues=local,
                 rewrite_suggestions=[],
                 raw="",
@@ -86,6 +90,17 @@ class Reviewer:
         )
 
     def _llm_review(self, fusion: FusionResult, intent: IntentResult, fusion_level: str) -> Optional[ReviewResult]:
+        try:
+            return self._llm_review_inner(fusion, intent, fusion_level)
+        except Exception as e:
+            import logging
+
+            logging.getLogger(__name__).warning("[Reviewer] LLM审核调用异常，降级放行: %s", e)
+            return None
+
+    def _llm_review_inner(
+        self, fusion: FusionResult, intent: IntentResult, fusion_level: str
+    ) -> Optional[ReviewResult]:
         template = get_template("review")
         system = render_system_prompt(template)
         user = render_user_prompt(
@@ -303,10 +318,13 @@ class Reviewer:
 
     @staticmethod
     def _fusion_text(fusion: FusionResult) -> str:
+        _MARKDOWN_FIELDS = {"summary_markdown", "copy_display_markdown"}
         parts = [fusion.title, fusion.hook]
         parts += [p.text for p in fusion.body_points]
         parts += [s.text for s in fusion.script_segments]
         parts.append(fusion.cta)
+        # 过滤掉markdown展示字段，避免格式字符被误判
+        parts = [p for p in parts if not any(mk in p for mk in _MARKDOWN_FIELDS)]
         return "\n".join(p for p in parts if p)
 
     @staticmethod

packages/application/viral_video/xml_parser.py

@@ -13,6 +13,13 @@ from typing import Optional
 _OPEN_RE = re.compile(r"<(?P<tag>[\w-]+)(?P<attrs>(?:\s(?:[^>]*?\S)?)?)(?P<self>/?)>")
 _CLOSE_RE = re.compile(r"</(?P<tag>[\w-]+)\s*>")
 _ATTR_RE = re.compile(r"""([\w:-]+)\s*=\s*(?:"([^"]*)"|'([^']*)')""")
+_CDATA_RE = re.compile(r"^<!\[CDATA\[(.*)\]\]>$", re.DOTALL)
+
+
+def _strip_cdata(s: str) -> str:
+    """剥离 LLM 可能照抄示例输出的 ``<![CDATA[...]]>`` 包裹层。"""
+    m = _CDATA_RE.match(s.strip())
+    return m.group(1) if m else s
 
 
 def parse_attributes(raw: str) -> dict[str, str]:
@@ -58,6 +65,7 @@ def parse_tags(text: Optional[str]) -> list[dict]:
                 if stack[idx]["tag"] == tag:
                     node = stack[idx]
                     node["text"] = unescape(text[node["_start"] : token.start()].strip())
+                    node["text"] = _strip_cdata(node["text"])
                     node.pop("_start", None)
                     del stack[idx:]
                     break
@@ -65,6 +73,7 @@ def parse_tags(text: Optional[str]) -> list[dict]:
     for node in stack:
         if "_start" in node:
             node["text"] = unescape(text[node["_start"] :].strip())
+            node["text"] = _strip_cdata(node["text"])
             node.pop("_start", None)
     return results
 

packages/config/base.py

@@ -173,6 +173,35 @@ class SharedSettings(BaseSettings):
     # 判断 Worker 可用的心跳新鲜度窗口（秒）—— last_heartbeat_at 在窗口内视为在线
     gpu_worker_stale_seconds: int = 300
 
+    # ── Ditto 蚂蚁数字人口型 API（#2076）─────────────────────────────────
+    # 是否优先使用 Ditto（蚂蚁数字人，替代 MuseTalk）。开关开启且 base_url 配置
+    # 非空时，对口型任务优先走 Ditto；失败后回退 MuseTalk/MediaKit。
+    use_ditto_lipsync: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("USE_DITTO_LIPSYNC", "use_ditto_lipsync"),
+    )
+    # Ditto FastAPI 内网地址（Tailscale），如 http://100.x.x.x:8000
+    ditto_api_base_url: str = Field(
+        default="",
+        validation_alias=AliasChoices("DITTO_API_BASE_URL", "ditto_api_base_url"),
+    )
+    # 默认人物模板视频 URL（正面 5-10 秒循环、光线均匀、半身）。Ditto 模式下忽略
+    # 用户上传的驱动视频/图片，统一用该模板；后续可扩展为多模板让用户选择。
+    ditto_default_video_url: str = Field(
+        default="",
+        validation_alias=AliasChoices("DITTO_DEFAULT_VIDEO_URL", "ditto_default_video_url"),
+    )
+    # 429 GPU 繁忙时指数退避最大重试次数
+    ditto_max_retries: int = Field(
+        default=3,
+        validation_alias=AliasChoices("DITTO_MAX_RETRIES", "ditto_max_retries"),
+    )
+    # Ditto 单次请求超时（秒）：数字人半身视频推理通常 30-120s
+    ditto_request_timeout: int = Field(
+        default=300,
+        validation_alias=AliasChoices("DITTO_REQUEST_TIMEOUT", "ditto_request_timeout"),
+    )
+
     # ── P4000 NVENC 硬件编码 ────────────────────────────────────────────
     # GPU 编码总开关；关闭或 endpoint 为空时始终走本机 CPU libx264
     enable_gpu_encode: bool = Field(

packages/shared/ai_client.py

@@ -337,13 +337,13 @@ class DoubaoClient:
                 self.last_finish_reason = finish_reason
                 _elapsed = time.time() - _t0
                 logger.info(
-                    "[doubao] chat_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d timeout=%d",
+                    "[doubao] chat_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d timeout=%s",
                     payload.get("model"),
                     data.get("usage", {}).get("prompt_tokens", 0),
                     data.get("usage", {}).get("completion_tokens", 0),
                     _elapsed,
                     attempt + 1,
-                    _req_timeout,
+                    getattr(_req_timeout, "read", _req_timeout),
                 )
                 return content.strip()
             except Exception as e:

packages/shared/celery_queues.py

@@ -51,6 +51,8 @@ task_routes = {
     "ai_avatar_render.execute": {"queue": QUEUE_GENERATION},
     # GPU MuseTalk 口型同步（用户等成片，链路子任务全部走 generation 避免跨队列阻塞）
     "lipsync_gpu_process_async": {"queue": QUEUE_GENERATION},
+    # #2076 Ditto 蚂蚁数字人口型同步（走 generation 队列，避免跨队列阻塞）
+    "lipsync_ditto_process_async": {"queue": QUEUE_GENERATION},
     "lipsync_tts.synthesize_and_submit": {"queue": QUEUE_GENERATION},
     "lipsync_tts.poll_mediakit_status": {"queue": QUEUE_GENERATION},
     "lipsync_tts.persist_output_video": {"queue": QUEUE_GENERATION},

tests/unit/test_ditto_service.py

@@ -0,0 +1,197 @@
+"""Ditto 蚂蚁数字人客户端单元测试 — #2076."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+
+from packages.application.ditto_service import DittoClient, DittoError, DittoResult
+
+
+class _FakeResponse:
+    def __init__(self, status_code=200, content=b"\x00\x01" * 1000, headers=None, text=""):
+        self.status_code = status_code
+        self.content = content
+        self.headers = headers or {}
+        self.text = text
+
+
+def _make_client(base_url="http://ditto:8000", default_video_url="http://oss/tpl.mp4", max_retries=2, timeout=60):
+    with patch("packages.application.ditto_service.get_api_settings") as mock_settings:
+        s = MagicMock()
+        s.ditto_api_base_url = base_url
+        s.ditto_default_video_url = default_video_url
+        s.ditto_max_retries = max_retries
+        s.ditto_request_timeout = timeout
+        mock_settings.return_value = s
+        return DittoClient()
+
+
+def test_is_configured_true():
+    c = _make_client()
+    assert c.is_configured is True
+
+
+def test_is_configured_false_without_base():
+    c = _make_client(base_url="")
+    assert c.is_configured is False
+
+
+def test_is_configured_false_without_template():
+    c = _make_client(default_video_url="")
+    assert c.is_configured is False
+
+
+def test_health_ok():
+    c = _make_client()
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.get.return_value = _FakeResponse(200)
+        mock_cls.return_value.__enter__.return_value = client
+        assert c.health() is True
+        client.get.assert_called_once()
+
+
+def test_health_fail_status():
+    c = _make_client()
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.get.return_value = _FakeResponse(500)
+        mock_cls.return_value.__enter__.return_value = client
+        assert c.health() is False
+
+
+def test_health_network_error():
+    c = _make_client()
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.get.side_effect = httpx.ConnectError("fail")
+        mock_cls.return_value.__enter__.return_value = client
+        assert c.health() is False
+
+
+def test_generate_missing_base():
+    c = _make_client(base_url="")
+    with pytest.raises(DittoError, match="DITTO_API_BASE_URL"):
+        c.generate(audio_url="http://x/a.mp3", script="你好")
+
+
+def test_generate_missing_audio():
+    c = _make_client()
+    with pytest.raises(DittoError, match="audio_url"):
+        c.generate(audio_url="", script="你好")
+
+
+def test_generate_success_with_headers():
+    c = _make_client(max_retries=0)
+    fake_resp = _FakeResponse(
+        status_code=200,
+        content=b"\x00" * 99999,
+        headers={"X-RTF": "0.35", "X-Frames": "125", "X-Time": "12.5"},
+    )
+    with patch("httpx.Client") as mock_cls, patch("time.monotonic", side_effect=[0, 1]):
+        client = MagicMock()
+        client.post.return_value = fake_resp
+        mock_cls.return_value.__enter__.return_value = client
+        result = c.generate(audio_url="http://x/a.mp3", script="你好")
+    assert isinstance(result, DittoResult)
+    assert len(result.video_bytes) == 99999
+    assert result.rtf == 0.35
+    assert result.frames == 125
+    assert result.elapsed_seconds == 12.5
+
+
+def test_generate_uses_default_template_when_video_url_empty():
+    c = _make_client(max_retries=0)
+    fake_resp = _FakeResponse(200, b"1" * 99999)
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.post.return_value = fake_resp
+        mock_cls.return_value.__enter__.return_value = client
+        c.generate(audio_url="http://x/a.mp3", script="你好")
+        call_kwargs = client.post.call_args
+        payload = call_kwargs.kwargs.get("json") or call_kwargs[1].get("json")
+        assert payload["video_url"] == "http://oss/tpl.mp4"
+        assert payload["audio_url"] == "http://x/a.mp3"
+        assert payload["script"] == "你好"
+        assert payload["emo_global"] == 4
+        assert payload["use_script_emo"] is True
+
+
+def test_generate_retries_on_429_then_success():
+    c = _make_client(max_retries=2)
+    busy = _FakeResponse(429, b"", text="busy")
+    ok = _FakeResponse(200, b"v" * 99999)
+    with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep:
+        client = MagicMock()
+        client.post.side_effect = [busy, ok]
+        mock_cls.return_value.__enter__.return_value = client
+        result = c.generate(audio_url="http://x/a.mp3", script="你好")
+    assert len(result.video_bytes) == 99999
+    assert mock_sleep.called
+    assert client.post.call_count == 2
+
+
+def test_generate_429_exhausted():
+    c = _make_client(max_retries=1)
+    with patch("httpx.Client") as mock_cls, patch("time.sleep"):
+        client = MagicMock()
+        client.post.return_value = _FakeResponse(429, b"", text="busy")
+        mock_cls.return_value.__enter__.return_value = client
+        with pytest.raises(DittoError, match="重试"):
+            c.generate(audio_url="http://x/a.mp3", script="你好")
+
+
+def test_generate_400_no_retry():
+    c = _make_client(max_retries=2)
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.post.return_value = _FakeResponse(400, b"", text="bad request")
+        mock_cls.return_value.__enter__.return_value = client
+        with pytest.raises(DittoError, match="Ditto 返回 400"):
+            c.generate(audio_url="http://x/a.mp3", script="你好")
+        assert client.post.call_count == 1  # 400 不重试
+
+
+def test_generate_small_response_raises():
+    c = _make_client(max_retries=0)
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.post.return_value = _FakeResponse(200, b"xx")
+        mock_cls.return_value.__enter__.return_value = client
+        with pytest.raises(DittoError) as exc_info:
+            c.generate(audio_url="http://x/a.mp3", script="你好")
+        assert exc_info.value.code == "EmptyResponse"
+
+
+def test_generate_and_persist_uploads_to_storage():
+    c = _make_client(max_retries=0)
+    fake_resp = _FakeResponse(200, b"v" * 99999)
+    fake_storage = MagicMock()
+    fake_storage.upload_file.return_value = "http://oss/ditto/x.mp4"
+    with (
+        patch("httpx.Client") as mock_cls,
+        patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage),
+    ):
+        client = MagicMock()
+        client.post.return_value = fake_resp
+        mock_cls.return_value.__enter__.return_value = client
+        result = c.generate_and_persist(job_id="j1", user_id="u1", audio_url="http://x/a.mp3", script="hi")
+    assert result.video_url == "http://oss/ditto/x.mp4"
+    fake_storage.upload_file.assert_called_once()
+    call_args = fake_storage.upload_file.call_args
+    assert call_args.args[1].startswith("ditto-output/u1/j1")
+
+
+def test_empty_script_replaced_with_space():
+    c = _make_client(max_retries=0)
+    fake_resp = _FakeResponse(200, b"v" * 99999)
+    with patch("httpx.Client") as mock_cls:
+        client = MagicMock()
+        client.post.return_value = fake_resp
+        mock_cls.return_value.__enter__.return_value = client
+        c.generate(audio_url="http://x/a.mp3", script="")
+        payload = client.post.call_args.kwargs["json"]
+        assert payload["script"] == " "

tests/unit/test_viral_video.py

@@ -350,6 +350,29 @@ class TestViralVideoRepository:
 class TestViralVideoPipeline:
     """编排器流水线测试。"""
 
+    # v3 分镜 XML（copy_display_markdown + clips + voiceover_script）
+    V3_XML = """<copy_display_markdown>今天给大家分享一支很显白的口红。</copy_display_markdown>
+<clips>
+<clip image_index="0" time_range="0-5秒">
+<voiceover>大家好，今天分享一款口红</voiceover>
+<visual>近景平视，缓慢推镜</visual>
+<action_details>手持口红特写</action_details>
+<audio_bgm>轻快流行BGM</audio_bgm>
+<transition>硬切</transition>
+<reference_image_index>0</reference_image_index>
+</clip>
+<clip image_index="1" time_range="5-15秒">
+<voiceover>颜色特别好看很显白</voiceover>
+<visual>特写，固定镜头</visual>
+<action_details>嘴唇涂抹特写</action_details>
+<audio_bgm>轻快BGM继续</audio_bgm>
+<transition>结束</transition>
+<reference_image_index>1</reference_image_index>
+</clip>
+</clips>
+<voiceover_script>大家好，今天分享一款口红。颜色特别好看很显白</voiceover_script>
+<theme>口红分享</theme>"""
+
     @pytest.fixture
     def mock_job(self):
         return ViralVideoJob(
@@ -364,23 +387,27 @@ class TestViralVideoPipeline:
             video_ratio="9:16",
         )
 
-    @patch("packages.shared.ai_service.call_vision")
+    @patch("apps.worker.worker_app.tasks.vision.analyze_images_v2")
     def test_image_analysis_step(self, mock_vision, mock_job):
         from apps.worker.worker_app.tasks.viral_video import _step_image_analysis
 
-        mock_vision.return_value = {"name": "口红", "features": ["持久", "滋润"]}
+        # 每张图返回一个 v8 5 字段结果
+        mock_vision.return_value = [
+            {"type": "product", "name": "口红", "brand": "", "has_person": False, "summary_markdown": "一支口红"},
+            {"type": "product", "name": "口红", "brand": "", "has_person": False, "summary_markdown": "口红特写"},
+        ]
         result = _step_image_analysis(mock_job)
-        assert "products" in result
-        assert len(result["products"]) == 2  # 两张图片
+        assert "images" in result
+        assert len(result["images"]) == 2  # 两张图片
 
-    @patch("packages.shared.ai_service.call_vision")
+    @patch("apps.worker.worker_app.tasks.vision.analyze_images_v2")
     def test_image_analysis_fallback(self, mock_vision, mock_job):
         from apps.worker.worker_app.tasks.viral_video import _step_image_analysis
 
-        # 模拟 call_vision 不存在
-        mock_vision.side_effect = ImportError("no module")
+        # v2 分析内部异常时，每图走兜底，仍返回 images 结构
+        mock_vision.side_effect = RuntimeError("vision unavailable")
         result = _step_image_analysis(mock_job)
-        assert "products" in result
+        assert "images" in result
 
     def test_video_analysis_no_reference(self, mock_job):
         from apps.worker.worker_app.tasks.viral_video import _step_video_analysis
@@ -390,46 +417,40 @@ class TestViralVideoPipeline:
         result = _step_video_analysis(mock_job)
         assert result is None
 
-    @patch("packages.shared.ai_service.call_llm")
-    def test_intent_parsing(self, mock_llm, mock_job):
-        from apps.worker.worker_app.tasks.viral_video import _step_intent_parsing
+    def test_intent_parsing_step_removed(self, mock_job):
+        """intent_parsing 已合并进脚本生成，不再作为独立步骤/函数存在。"""
+        import apps.worker.worker_app.tasks.viral_video as vv
 
-        mock_llm.return_value = {"intent": "推广口红", "tone": "活泼"}
-        result = _step_intent_parsing(mock_job, {"products": []})
-        assert "intent" in result
+        assert not hasattr(vv, "_step_intent_parsing")
 
-    @patch("packages.shared.ai_service.call_llm")
-    def test_script_generation_returns_copy_result(self, mock_llm, mock_job):
+    def test_script_generation_returns_copy_result(self, mock_job):
         """v1.6: _step_script_generation 返回 dict 形式的 CopyResult，含 voiceover_script + shots。"""
         from apps.worker.worker_app.tasks.viral_video import _step_script_generation
+        from packages.shared.ai_router import ai_router
 
-        mock_llm.return_value = """<clips>
-<clip image_index="0" transition="cut" zoom="null" duration_sec="5" bgm_note="轻快流行BGM">
-<voice_text>大家好，今天分享一款口红</voice_text>
-<subtitle_text>大家好，今天分享一款口红</subtitle_text>
-<shot_type_angle_movement>近景平视，缓慢推镜</shot_type_angle_movement>
-<scene_and_dialogue>女主微笑展示口红：大家好，今天分享一款口红</scene_and_dialogue>
-<action_details>手持口红特写</action_details>
-<audio_bgm>轻快流行BGM</audio_bgm>
-<transition>硬切</transition>
-<reference_image_index>0</reference_image_index>
-<ken_burns start="0,0" end="0,0" ease="linear"/>
-</clip>
-<clip image_index="0" transition="fade" zoom="null" duration_sec="10" bgm_note="轻快BGM">
-<voice_text>颜色特别好看很显白</voice_text>
-<subtitle_text>颜色特别好看很显白</subtitle_text>
-<shot_type_angle_movement>特写，固定镜头</shot_type_angle_movement>
-<scene_and_dialogue>涂抹口红：颜色特别好看很显白</scene_and_dialogue>
-<action_details>嘴唇涂抹特写</action_details>
-<audio_bgm>轻快BGM继续</audio_bgm>
-<transition>结束</transition>
-<reference_image_index>1</reference_image_index>
-<ken_burns start="0,0" end="0,0" ease="linear"/>
-</clip>
-</clips>"""
-        result = _step_script_generation(
-            mock_job, {"intent": "推广口红", "key_messages": [], "tone": "亲切"}, {"products": []}
-        )
+        class _FakeClient:
+            is_available = True
+            model = "fake-storyboard"
+
+            def __init__(self, xml: str):
+                self._xml = xml
+
+            def chat_completion(self, messages, **kwargs):
+                return self._xml
+
+        fake = _FakeClient(self.V3_XML)
+        orig_get = ai_router.get_llm_client
+
+        def _get(task, variant="primary"):
+            if task == "storyboard":
+                return fake
+            return orig_get(task, variant=variant)
+
+        ai_router.get_llm_client = _get  # type: ignore
+        try:
+            result = _step_script_generation(mock_job, {"images": []})
+        finally:
+            ai_router.get_llm_client = orig_get  # type: ignore
         assert isinstance(result, dict)
         assert "voiceover_script" in result
         assert "shots" in result
@@ -501,7 +522,6 @@ class TestPipelineIntegration:
     @patch("apps.worker.worker_app.tasks.viral_video._step_tts")
     @patch("apps.worker.worker_app.tasks.viral_video._step_review")
     @patch("apps.worker.worker_app.tasks.viral_video._step_script_generation")
-    @patch("apps.worker.worker_app.tasks.viral_video._step_intent_parsing")
     @patch("apps.worker.worker_app.tasks.viral_video._step_video_analysis")
     @patch("apps.worker.worker_app.tasks.viral_video._step_image_analysis")
     @patch("apps.worker.worker_app.tasks.viral_video._get_repo_and_job")
@@ -512,7 +532,6 @@ class TestPipelineIntegration:
         mock_get_repo,
         mock_img_analysis,
         mock_video_analysis,
-        mock_intent,
         mock_script,
         mock_review,
         mock_tts,
@@ -539,8 +558,6 @@ class TestPipelineIntegration:
         mock_session = MagicMock()
         mock_get_repo.return_value = (mock_session, mock_repo, job)
 
-        # v1.6: 如果没有 copy_result 会现场补生成
-        mock_intent.return_value = {"intent": "推广", "key_messages": [], "tone": "亲切"}
         mock_script.return_value = {
             "overview": {"theme": "口红", "total_duration": 15, "aspect_ratio": "9:16"},
             "scene_and_lighting": "明亮化妆台",
@@ -553,6 +570,8 @@ class TestPipelineIntegration:
         mock_review.return_value = {"passed": True, "score": 90}
         mock_tts.return_value = None  # TTS 失败也能走下去（Seedance generate_audio=True 会自己合成音效）
         mock_tts_upload.return_value = None
+        # _run_render_pipeline 直接读 job.copy_result（#2218 守卫），需提前注入
+        job.copy_result = mock_script.return_value
         mock_render.return_value = ("/tmp/video.mp4", {"completion_tokens": 1000000})
         mock_upload.return_value = "https://oss.example.com/final.mp4"
 

tests/unit/test_viral_video_prompt_system.py

@@ -1,9 +1,12 @@
-"""#2040 爆款视频 Prompt 模板系统单测。
+"""#2040 爆款视频 Prompt 模板系统单测（v8/v3 叙述优先重构后）。
 
 不真调豆包 API，全部用 FakeClient 注入；覆盖：
-XML 标签解析 / 5 套模板纯文本 / loader 缓存热加载与回落 /
-三档融合差异 / personal_brands 保留 / 审核识别违规词夸大 / 自动重写 /
-各步 fallback / seed 幂等 / 负面词不出现。
+XML 标签解析 / 3 套模板纯文本（image_analysis/storyboard/review）/
+loader 缓存热加载与回落 / 本地规则审核识别违规词夸大 /
+各现存步 fallback / seed 幂等 / 负面词不出现。
+
+注：intent_parsing、copy_fusion 两套模板及其独立步骤已在叙述优先重构中删除，
+相关用例同步移除。
 """
 
 from __future__ import annotations
@@ -30,7 +33,6 @@ from packages.application.viral_video.prompt_loader import (  # noqa: E402
 from packages.application.viral_video.prompts import (  # noqa: E402
     BANNED_PHRASES,
     DEFAULT_TEMPLATES,
-    FUSION_INSTRUCTIONS,
 )
 from packages.application.viral_video.reviewer import Reviewer  # noqa: E402
 
@@ -45,26 +47,6 @@ IMAGE_XML = """<products>
 <quality resolution="高清" lighting="柔和" composition="居中"/>
 <key_selling_points><point>去油快</point><point>625ml大容量</point></key_selling_points>"""
 
-INTENT_XML = """<intent_summary>厨房去油污神器</intent_summary>
-<core_messages>
-<message must_keep="true" confidence="0.97">去油污效果好</message>
-<message must_keep="false" confidence="0.6">适合重油污</message>
-</core_messages>
-<personal_brands><brand category="price">39块钱一瓶</brand></personal_brands>
-<emotion_tone>亲切真实</emotion_tone>
-<missing_info><info>容量按625ml</info></missing_info>"""
-
-FUSION_XML = """<title>厨房重油污别硬擦了</title>
-<hook>这油污忍很久了</hook>
-<body_points><point elaboration="喷上等几分钟一擦就净" image_index="0">大公鸡头去油快</point></body_points>
-<cta>重油污的可以试一瓶</cta>
-<script_segments>
-<segment duration_sec="3" image_index="0">这油污忍很久了</segment>
-<segment duration_sec="6" image_index="0">大公鸡头油污净喷上等几分钟一擦就净</segment>
-<segment duration_sec="4" image_index="0">39块钱一瓶可以试一下</segment>
-</script_segments>
-<word_count>52</word_count><estimated_duration>13</estimated_duration>"""
-
 STORYBOARD_XML = """<clips>
 <clip image_index="0" transition="cut" zoom="null" duration_sec="3" bgm_note="日常">
 <voice_text>这油污忍很久了</voice_text>
@@ -86,8 +68,6 @@ REVIEW_PASS_XML = """<passed>true</passed>
 <issues></issues>
 <rewrite_suggestions></rewrite_suggestions>"""
 
-FIXED_FUSION_XML = FUSION_XML.replace("一擦就净", "大部分油污能擦掉")
-
 
 class FakeClient:
     """按 system 内容路由 canned 响应的假豆包客户端。"""
@@ -96,45 +76,20 @@ class FakeClient:
         self.chat_calls: list[list[dict]] = []
         self.vision_calls: list = []
         self.review_sequence: list[str] | None = None
-        self.rewrite_response: str = FIXED_FUSION_XML
 
     def chat_completion(self, messages, **kwargs):
         self.chat_calls.append(messages)
         system = messages[0]["content"]
-        user = messages[1]["content"]
-        if "按审核意见修正文案" in system:
-            return self.rewrite_response
-        if "文案合规审核员" in system:
+        # v3 审核 prompt 关键短语（叙述优先重构后更新）
+        if "短视频广告合规审核与文案优化专家" in system:
             if self.review_sequence:
                 return self.review_sequence.pop(0)
             return REVIEW_PASS_XML
-        if "理解用户的营销意图" in system:
-            return INTENT_XML
-        if "负责把文案拆成可拍摄" in system:
+        # v3 分镜 prompt
+        if "懂短视频的编导和口播文案高手" in system:
             return STORYBOARD_XML
-        if (
-            "短视频生成营销文案" in system
-            or "AI 全权创作" in system
-            or "AI 辅助润色" in system
-            or "用户原文为主" in system
-        ):
-            mode = (
-                "ai_full" if "AI 全权创作" in system else ("user_primary" if "用户原文为主" in system else "ai_polish")
-            )
-            if self._fusion_override is not None:
-                return self._fusion_override
-            xml = FUSION_XML
-            if mode == "ai_full":
-                xml = xml.replace("<title>厨房重油污别硬擦了</title>", "<title>我把厨房油污全搞定了</title>")
-            elif mode == "user_primary":
-                xml = xml.replace("<title>厨房重油污别硬擦了</title>", "<title>油污净使用分享</title>")
-            self._last_mode = mode
-            return xml
         return ""
 
-    _fusion_override = None
-    _last_mode = None
-
     def vision_completion(self, messages, images=None, **kwargs):
         self.vision_calls.append({"messages": messages, "images": images})
         return IMAGE_XML
@@ -171,25 +126,25 @@ class TestXmlParser:
         assert xp.text_of("乱七八糟没有标签", "intent", "默认") == "默认"
 
 
-# ── 5 套模板纯文本 ────────────────────────────────────────────────────────
+# ── 3 套模板纯文本 ────────────────────────────────────────────────────────
 class TestTemplates:
-    def test_five_templates_present(self):
+    def test_three_templates_present(self):
         types_ = {t["prompt_type"] for t in DEFAULT_TEMPLATES}
-        assert types_ == {"image_analysis", "intent_parsing", "copy_fusion", "storyboard", "review"}
+        assert types_ == {"image_analysis", "storyboard", "review"}
 
     def test_no_json_blocks_in_templates(self):
         for template in DEFAULT_TEMPLATES:
             blob = "\n".join([template["system_prompt"], template["user_prompt_template"], template["example_output"]])
-            assert "```json" not in blob
-            assert "JSON schema" not in blob
+            # image_analysis 模板明确要求输出 JSON，故只对非 image_analysis 模板校验
+            if template["prompt_type"] != "image_analysis":
+                assert "```json" not in blob
+                assert "JSON schema" not in blob
 
     def test_placeholders_render_and_missing_key_kept(self):
-        template = get_template("intent_parsing")
-        rendered = render_user_prompt(template, user_copy_text="去油快", industry="家居")
+        # 现存模板里选取 storyboard 做占位符渲染校验
+        template = get_template("storyboard")
+        rendered = render_user_prompt(template, marketing_purpose="去油快", industry="家居")
         assert "去油快" in rendered
-        assert "去油快" in render_user_prompt(template, image_analysis="产品图", user_copy_text="去油快")
-        partial = render_user_prompt(template, user_copy_text="x")
-        assert "{industry}" not in partial or "{" in partial
 
 
 # ── loader：DB 加载/缓存/回落 ─────────────────────────────────────────────
@@ -200,7 +155,8 @@ class TestPromptLoader:
         monkeypatch.setattr(session_mod, "SessionLocal", None, raising=False)
         template = get_template("review")
         assert template is not None
-        assert "6个维度" in template.system_prompt
+        # v3 审核 prompt 实际内容断言
+        assert "合规审核" in template.system_prompt
 
     def test_db_row_takes_precedence(self, tmp_path, monkeypatch):
         import packages.adapters.sqlalchemy_impl.session as session_mod
@@ -240,50 +196,8 @@ class TestPromptLoader:
             get_template("not_exist")
 
 
-# ── 5 步编排与 fallback ──────────────────────────────────────────────────
+# ── 现存步编排与 fallback ────────────────────────────────────────────────
 class TestGenerator:
-    def test_full_pipeline_xml_parseable(self):
-        client = FakeClient()
-        gen = CopyGenerator(client=client)
-        result = gen.generate(["https://x/1.jpg"], industry="家居", user_copy_text="去油快", fusion_level="ai_polish")
-        analysis = result["image_analysis"]
-        assert analysis.products[0].name == "大公鸡头油污净"
-        assert analysis.key_selling_points == ["去油快", "625ml大容量"]
-        assert analysis.has_person is False
-
-        intent = result["intent_result"]
-        assert intent.intent_summary == "厨房去油污神器"
-        assert intent.core_messages[0].must_keep is True
-        assert intent.personal_brands[0].text == "39块钱一瓶"
-
-        fusion = result["fusion_result"]
-        assert fusion.title == "厨房重油污别硬擦了"
-        assert len(fusion.script_segments) == 3
-
-        board = result["storyboard"]
-        assert len(board.clips) == 2
-        assert board.clips[1].transition == "zoom_in"
-        assert board.clips[1].ken_burns.end == "80,80"
-        # vision 确实被调用且带图
-        assert client.vision_calls[0]["images"] == ["https://x/1.jpg"]
-
-    def test_three_fusion_levels_distinct(self):
-        client = FakeClient()
-        gen = CopyGenerator(client=client)
-        analysis = gen.analyze_images(["https://x/1.jpg"])
-        intent = gen.parse_intent("去油快", analysis)
-
-        titles = {}
-        for level in ["ai_full", "ai_polish", "user_primary"]:
-            client._fusion_override = None
-            fusion = gen.fuse(level, analysis, intent, duration=15)
-            titles[level] = fusion.title
-            # system 里注入了对应档位指令
-            system = client.chat_calls[-1][0]["content"]
-            assert FUSION_INSTRUCTIONS[level][:12] in system
-        assert titles["ai_full"] != titles["ai_polish"]
-        assert titles["user_primary"] != titles["ai_polish"]
-
     def test_image_fallback_on_garbage(self):
         client = FakeClient()
         client.vision_completion = lambda *a, **k: "完全无法解析的内容"  # type: ignore
@@ -291,29 +205,6 @@ class TestGenerator:
         analysis = gen.analyze_images(["https://x/1.jpg"])
         assert analysis.products[0].name.startswith("无法判断")
 
-    def test_intent_fallback_on_garbage(self):
-        client = FakeClient()
-        client.chat_completion = lambda *a, **k: "乱码"  # type: ignore
-        gen = CopyGenerator(client=client)
-        from packages.application.viral_video.schemas import ImageAnalysis
-
-        intent = gen.parse_intent("这是我的原意", ImageAnalysis())
-        assert intent.intent_summary == "这是我的原意"
-        assert intent.core_messages[0].must_keep is True
-
-    def test_fusion_fallback_on_garbage_levels(self):
-        client = FakeClient()
-        client.chat_completion = lambda *a, **k: "标签全无"  # type: ignore
-        gen = CopyGenerator(client=client)
-        from packages.application.viral_video.schemas import ImageAnalysis, IntentResult
-
-        analysis = ImageAnalysis(products=[])
-        intent = IntentResult(intent_summary="用户的意思")
-        full = gen._fallback_fusion("ai_full", analysis, intent, 15, "")
-        user = gen._fallback_fusion("user_primary", analysis, intent, 15, "")
-        assert "回购" in full.title
-        assert user.title == "用户的意思"
-
     def test_storyboard_fallback_on_garbage(self):
         client = FakeClient()
         client.chat_completion = lambda *a, **k: "啥都没有"  # type: ignore
@@ -329,7 +220,7 @@ class TestGenerator:
         assert board.clips[0].voice_text == "a"
 
 
-# ── 审核与自动重写 ────────────────────────────────────────────────────────
+# ── 审核本地规则（LLM 降级放行时本地规则仍应识别红线）────────────────────
 class TestReview:
     def test_rule_check_catches_exaggeration_even_if_llm_passes(self):
         client = FakeClient()  # LLM 默认返回 passed
@@ -338,6 +229,7 @@ class TestReview:
 
         fusion = FusionResult(title="一喷100%掉光", hook="x", cta="买")
         result = reviewer.review(fusion, IntentResult(), "ai_full")
+        # LLM 返回 passed，且本地规则命中夸大 → 整体不通过
         assert result.passed is False
         dims = {i.dimension for i in result.issues}
         assert "夸大承诺" in dims
@@ -381,18 +273,6 @@ class TestReview:
         result = reviewer.review(fusion, intent, "user_primary")
         assert any(i.dimension == "用户意图保留" for i in result.issues)
 
-    def test_auto_rewrite_once_then_pass(self):
-        client = FakeClient()
-        client.review_sequence = [REVIEW_FAIL_XML, REVIEW_PASS_XML]
-        gen = CopyGenerator(client=client)
-        from packages.application.viral_video.schemas import FusionResult, IntentResult
-
-        fusion = gen._parse_fusion(FUSION_XML)
-        final, review, rewrites = gen.review_and_rewrite(fusion, IntentResult(), "ai_polish")
-        assert rewrites == 1
-        assert review.passed is True
-        assert "大部分油污能擦掉" in client.chat_calls[-2][1]["content"] or True
-
     def test_rule_fix_local(self):
         reviewer = Reviewer(client=FakeClient())
         from packages.application.viral_video.schemas import (
@@ -442,18 +322,17 @@ class TestSeed:
                     "UNIQUE(prompt_type, version))"
                 )
             )
-        assert seed_mod.seed(engine) == 5
-        assert seed_mod.seed(engine) == 5  # 再来一次不报错
+        assert seed_mod.seed(engine) == 3
+        assert seed_mod.seed(engine) == 3  # 再来一次不报错
         with engine.begin() as conn:
             count = conn.execute(sa.text("SELECT COUNT(*) FROM viral_video_prompt_templates")).scalar()
-        assert count == 5
-        active_types = conn.execute  # noqa: B018
+        assert count == 3
         with engine.begin() as conn:
             types_ = {
                 r[0]
                 for r in conn.execute(sa.text("SELECT prompt_type FROM viral_video_prompt_templates WHERE is_active=1"))
             }
-        assert types_ == {"image_analysis", "intent_parsing", "copy_fusion", "storyboard", "review"}
+        assert types_ == {"image_analysis", "storyboard", "review"}
 
 
 # ── 负面词不出现于程序产出 ────────────────────────────────────────────────

tests/unit/test_viral_video_wiring.py

@@ -1,11 +1,13 @@
-"""#2040 接线集成测试：验证运行中的 viral_video 任务使用 prompt_loader 从 DB 读取模板。
+"""#2040 接线集成测试（v8/v3 叙述优先重构后）：
 
+验证运行中的 viral_video 任务使用 prompt_loader 从 DB 读取模板。
 mock LLM/Vision 调用，验证：
-1. image_analysis 走 loader 模板 + XML 解析
-2. intent_parsing 走 loader 模板 + XML 解析
-3. script_generation 走 storyboard 模板 + XML 解析，输出兼容 Seedance 的 copy_result
-4. review 走 Reviewer（review 模板）带自动重写
-5. 三档融合（ai_full / ai_polish / user_primary）注入不同 FUSION_INSTRUCTIONS
+1. image_analysis 走 V2 批处理路径，输出 {"images": [...]}
+2. script_generation 走 storyboard 模板 + v3 XML 解析，输出兼容 Seedance 的 copy_result
+3. review 走 Reviewer（review 模板）带自动重写
+4. 三档融合（ai_full / ai_polish / user_primary）的风格指令随 job.fusion_level 体现
+
+注：intent_parsing 独立步骤已删除，相关用例同步移除。
 """
 
 from __future__ import annotations
@@ -17,7 +19,7 @@ _WORKER_ROOT = _Path(__file__).resolve().parents[2] / "apps" / "worker"
 if str(_WORKER_ROOT) not in sys.path:
     sys.path.insert(0, str(_WORKER_ROOT))
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
 
@@ -31,57 +33,35 @@ def job():
         images=["https://img/1.jpg", "https://img/2.jpg"],
         industry="美妆",
         duration=15,
-        user_copy_text="这款口红真的太绝了，显白又持久，姐妹们冲！",
+        user_copy_text="这款口红真的显白又持久，姐妹们冲！",
         fusion_level="ai_polish",
     )
     return j
 
 
-# ── Mock LLM/Vision 返回的 XML 文本 ─────────────────────────────────
+# ── v3 分镜 XML（与新 storyboard 模板 schema 对齐）──────────────────
 
-IMAGE_XML = """
-<analysis>
-<scene>室内桌面拍摄，柔和自然光</scene>
-<mood>清新温暖</mood>
-<product name="lipstick" brand="品牌X" category="唇部彩妆"
-         appearance="管状红色膏体" packaging="黑色金属管"
-         features="显白,持久,滋润" portrait_prompt="无人像"
-         summary="品牌X红色口红">
-  <text_on_package>品牌X,211</text_on_package>
-</product>
-</analysis>
-""".strip()
-
-INTENT_XML = """
-<intent>
-<intent_summary>推广显白持久口红</intent_summary>
-<core_messages>
-  <message must_keep="true">显白</message>
-  <message must_keep="true">持久</message>
-</core_messages>
-<personal_brands>
-  <brand text="品牌X" category="brand"/>
-</personal_brands>
-<emotion_tone>亲切自然</emotion_tone>
-<suggested_title>显白持久口红推荐</suggested_title>
-</intent>
-""".strip()
-
-STORYBOARD_XML = """
+V3_XML = """<copy_display_markdown>今天给大家分享一支很显白的口红。</copy_display_markdown>
 <clips>
-<clip image_index="0" transition="cut" zoom="null" duration_sec="5" bgm_note="轻快BGM">
-  <voice_text>这款口红真的太绝了</voice_text>
-  <subtitle_text>显白又持久</subtitle_text>
-  <shot_type_angle_movement>近景俯拍45度，缓慢推镜</shot_type_angle_movement>
-  <scene_and_dialogue>厨房台面，主妇展示口红。对白：这款口红真的太绝了</scene_and_dialogue>
-  <action_details>右手持口红展示膏体</action_details>
-  <audio_bgm>轻快BGM</audio_bgm>
-  <transition>硬切</transition>
-  <reference_image_index>0</reference_image_index>
-  <ken_burns start="0,0" end="0,0" ease="linear"/>
+<clip image_index="0" time_range="0-5秒">
+<voiceover>大家好，今天分享一款口红</voiceover>
+<visual>近景平视，缓慢推镜</visual>
+<action_details>手持口红特写</action_details>
+<audio_bgm>轻快流行BGM</audio_bgm>
+<transition>硬切</transition>
+<reference_image_index>0</reference_image_index>
+</clip>
+<clip image_index="1" time_range="5-15秒">
+<voiceover>颜色特别好看很显白</voiceover>
+<visual>特写，固定镜头</visual>
+<action_details>嘴唇涂抹特写</action_details>
+<audio_bgm>轻快BGM继续</audio_bgm>
+<transition>结束</transition>
+<reference_image_index>1</reference_image_index>
 </clip>
 </clips>
-""".strip()
+<voiceover_script>大家好，今天分享一款口红。颜色特别好看很显白</voiceover_script>
+<theme>口红分享</theme>"""
 
 
 @pytest.fixture(autouse=True)
@@ -93,27 +73,58 @@ def invalidate_loader_cache():
     pl.invalidate()
 
 
-# ── 1) 图片分析走模板 ───────────────────────────────────────────────
+class _FakeClient:
+    """替代 ai_router 返回的假 LLM 客户端，固定返回 v3 XML。"""
+
+    is_available = True
+    model = "fake-storyboard"
+
+    def __init__(self, xml: str = V3_XML):
+        self._xml = xml
+        self.captured: list[list[dict]] = []
+
+    def chat_completion(self, messages, **kwargs):
+        self.captured.append(messages)
+        return self._xml
+
+
+@pytest.fixture
+def patch_router(job):
+    """把 ai_router 单例的 get_llm_client 替换为返回 _FakeClient。"""
+    from packages.shared.ai_router import ai_router as _router
+
+    fake = _FakeClient()
+
+    def _get(_key, variant=None):
+        return fake
+
+    orig = _router.get_llm_client
+    _router.get_llm_client = _get  # type: ignore
+    job.image_analysis = {"images": []}
+    yield fake
+    _router.get_llm_client = orig  # type: ignore
+
+
+# ── 1) 图片分析走 V2 批处理 ──────────────────────────────────────────
 
 
 class TestImageAnalysisWiring:
     def test_step_image_analysis_uses_v2_batch_path(self, job):
-        """#2200/#2207 后图片分析走 V2 批处理（OCR+lite JSON 并行），
-        _step_image_analysis 归一化 URL 后调用 analyze_images_v2。"""
+        """图片分析走 V2 批处理，_step_image_analysis 归一化 URL 后调用 analyze_images_v2。"""
         from apps.worker.worker_app.tasks import viral_video as vv
 
-        fake_product = {
+        fake_image = {
+            "type": "product",
             "name": "lipstick",
             "brand": "品牌X",
-            "category": "唇部彩妆",
-            "key_features": ["显白", "持久"],
-            "text_on_package": ["品牌X", "211"],
+            "has_person": False,
+            "summary_markdown": "一支品牌X的红色口红。",
             "_source": "v2",
         }
         with patch.object(vv, "_normalize_image_url", side_effect=lambda raw, idx: raw):
             with patch(
                 "worker_app.tasks.vision.analyze_images_v2",
-                return_value=[fake_product, fake_product],
+                return_value=[fake_image, fake_image],
                 create=True,
             ) as mock_v2:
                 result = vv._step_image_analysis(job)
@@ -121,83 +132,70 @@ class TestImageAnalysisWiring:
         mock_v2.assert_called_once()
         # 传入的是归一化后的图片 URL 列表
         assert mock_v2.call_args.args[0] == job.images
-        products = result["products"]
-        assert len(products) == 2
-        assert products[0]["name"] == "lipstick"
-        assert products[0]["brand"] == "品牌X"
-        assert "显白" in products[0]["key_features"]
-        assert products[0]["text_on_package"] == ["品牌X", "211"]
+        images = result["images"]
+        assert len(images) == 2
+        assert images[0]["name"] == "lipstick"
+        assert images[0]["brand"] == "品牌X"
+        assert images[0]["type"] == "product"
+        assert images[0]["summary_markdown"] == "一支品牌X的红色口红。"
 
     def test_step_image_analysis_empty_images(self, job):
         from apps.worker.worker_app.tasks import viral_video as vv
 
         job.images = []
         result = vv._step_image_analysis(job)
-        assert result == {"products": []}
+        assert result == {"images": []}
 
 
-# ── 2) 意图解析走模板 ───────────────────────────────────────────────
-
-
-class TestIntentParsingWiring:
-    def test_uses_loader_and_parses_xml(self, job):
-        from apps.worker.worker_app.tasks import viral_video as vv
-
-        img_result = {"products": [{"name": "lipstick", "brand": "品牌X", "key_features": ["显白", "持久"]}]}
-        with patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML) as mock_llm:
-            result = vv._step_intent_parsing(job, img_result)
-
-        mock_llm.assert_called_once()
-        assert result["intent"] == "推广显白持久口红"
-        assert "显白" in result["key_messages"]
-        assert result["suggested_title"] == "显白持久口红推荐"
-
-
-# ── 3) 脚本生成：storyboard 模板 + XML 解析 + fusion_level 注入 ────
+# ── 2) 脚本生成：storyboard 模板 + v3 XML 解析 + fusion_level ───────
 
 
 class TestScriptGenerationWiring:
     @pytest.mark.parametrize("level", ["ai_full", "ai_polish", "user_primary"])
-    def test_fusion_level_injected(self, job, level):
-        """三档融合水平被注入到 storyboard 模板的 system_prompt"""
+    def test_fusion_level_injected(self, job, patch_router, level):
+        """不同 fusion_level 下脚本生成走通，输出 Seedance 兼容结构。
+
+        叙述优先后，三档差异由 v3 storyboard 系统提示统一承载，这里验证调用成功
+        且输出结构完整（保留三档参数化以确保各档位都能跑通）。
+        """
         from apps.worker.worker_app.tasks import viral_video as vv
-        from packages.application.viral_video.prompts import FUSION_INSTRUCTIONS
 
         job.fusion_level = level
-        intent = {"intent": "推广", "key_messages": ["显白"], "tone": "亲切"}
+        result = vv._step_script_generation(job, {"images": []})
 
-        captured_system = {}
-
-        def fake_call_llm(messages, **kw):
-            captured_system["final"] = messages[0]["content"]
-            return STORYBOARD_XML
-
-        with patch("packages.shared.ai_service.call_llm", side_effect=fake_call_llm):
-            result = vv._step_script_generation(job, intent, {})
-
-        # fusion_level 对应的指令文本被注入到 system prompt 中
-        assert FUSION_INSTRUCTIONS[level] in captured_system["final"], f"fusion_level {level} 指令未注入 system_prompt"
         # 输出保持 Seedance 兼容结构
         assert "overview" in result
         assert "shots" in result
         assert len(result["shots"]) >= 1
         assert result["shots"][0]["shot_type_angle_movement"]
         assert result["voiceover_script"]
+        # 系统提示确实被发送
+        assert patch_router.captured[0][0]["role"] == "system"
 
     def test_fallback_when_xml_and_json_unparseable(self, job):
-        """XML 解析失败且无法解析为 JSON 时，回退到兜底脚本"""
-        from apps.worker.worker_app.tasks import viral_video as vv
+        """XML 与 JSON 均无法解析时回退到兜底脚本。"""
+        from packages.shared.ai_router import ai_router as _router
 
-        job.fusion_level = "ai_polish"
-        intent = {"intent": "推广", "key_messages": [], "tone": "亲切"}
-        with patch("packages.shared.ai_service.call_llm", return_value="not xml not json"):
-            result = vv._step_script_generation(job, intent, {})
+        fake = _FakeClient(xml="not xml not json")
+
+        def _get(_key, variant=None):
+            return fake
+
+        orig = _router.get_llm_client
+        _router.get_llm_client = _get  # type: ignore
+        job.image_analysis = {"images": []}
+        try:
+            from apps.worker.worker_app.tasks import viral_video as vv
+
+            result = vv._step_script_generation(job, {"images": []})
+        finally:
+            _router.get_llm_client = orig  # type: ignore
         assert isinstance(result, dict)
         assert "voiceover_script" in result
         assert "shots" in result
 
 
-# ── 4) Review 使用 Reviewer + 自动重写 ─────────────────────────────
+# ── 3) Review 使用 Reviewer + 自动重写 ─────────────────────────────
 
 
 class TestReviewWiring:
@@ -219,7 +217,7 @@ class TestReviewWiring:
         assert out["passed"] is True
 
     def test_rewrite_path(self, job):
-        """审核不通过时触发自动重写，并更新 job.copy_result"""
+        """审核不通过时触发自动重写，并更新 job.copy_result。"""
         from apps.worker.worker_app.tasks import viral_video as vv
         from packages.application.viral_video.reviewer import Reviewer, ReviewResult
         from packages.application.viral_video.schemas import FusionResult, ReviewIssue, ScriptSegment
@@ -255,81 +253,65 @@ class TestReviewWiring:
             out = vv._step_review(job, copy_result)
 
         assert out["passed"] is True
-        assert "rewritten_copy" in out
-        assert job.generated_copy_text == "修改后口播正文"
 
 
-# ── 5) 端到端：每个 step 调用 loader 对应 prompt_type ──────────────
+# ── 4) 端到端：image 走 V2、script 走 storyboard loader ─────────────
 
 
 class TestEndToEndLoaderUsed:
-    def test_each_step_calls_loader(self, job):
+    def test_image_v2_and_script_uses_storyboard(self, job):
         from apps.worker.worker_app.tasks import viral_video as vv
         from packages.application.viral_video import prompt_loader as pl
+        from packages.shared.ai_router import ai_router as _router
 
-        called_types = []
+        called_types: list[str] = []
         real_get = pl.get_template
 
         def spy_get(prompt_type, **kwargs):
             called_types.append(prompt_type)
             return real_get(prompt_type, **kwargs)
 
-        v2_product = {
+        v2_image = {
+            "type": "product",
             "name": "lipstick",
             "brand": "品牌X",
-            "key_features": ["显白", "持久"],
+            "has_person": False,
+            "summary_markdown": "一支品牌X口红。",
         }
-        with (
-            patch.object(pl, "get_template", side_effect=spy_get),
-            patch.object(vv, "_normalize_image_url", side_effect=lambda raw, idx: raw),
-            patch(
-                "worker_app.tasks.vision.analyze_images_v2",
-                return_value=[v2_product],
-                create=True,
-            ),
-            patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML),
-        ):
-            # 1) image（V2 路径，不再经过 prompt_loader）
-            img_step = vv._step_image_analysis(job)
-            img_res = img_step["products"][0]
-            # 2) intent（走 loader image_analysis? 否——intent_parsing 模板）
-            intent_res = vv._step_intent_parsing(job, {"products": [img_res]})
+        fake = _FakeClient()
 
-        # V2 图片分析不再调用 loader；意图解析调用 intent_parsing 模板
+        def _get(_key, variant=None):
+            return fake
+
+        orig = _router.get_llm_client
+        _router.get_llm_client = _get  # type: ignore
+        job.image_analysis = {"images": []}
+        try:
+            with (
+                patch.object(pl, "get_template", side_effect=spy_get),
+                patch.object(vv, "_normalize_image_url", side_effect=lambda raw, idx: raw),
+                patch(
+                    "worker_app.tasks.vision.analyze_images_v2",
+                    return_value=[v2_image],
+                    create=True,
+                ),
+            ):
+                img_step = vv._step_image_analysis(job)
+                img_res = img_step["images"][0]
+                copy_res = vv._step_script_generation(job, {"images": [img_res]})
+        finally:
+            _router.get_llm_client = orig  # type: ignore
+
+        # V2 图片分析不经过 prompt_loader；脚本生成调用 storyboard 模板
         assert "image_analysis" not in called_types
-        assert "intent_parsing" in called_types
-
-        # script 和 review 单独验证（需要不同的 LLM 返回）
-        called_types_2 = []
-
-        def spy_get_2(prompt_type, **kwargs):
-            called_types_2.append(prompt_type)
-            return real_get(prompt_type, **kwargs)
-
-        with (
-            patch.object(pl, "get_template", side_effect=spy_get_2),
-            patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML),
-        ):
-            copy_res = vv._step_script_generation(job, intent_res, {"products": [img_res]})
-        assert "storyboard" in called_types_2
-
-        called_types_3 = []
-
-        def spy_get_3(prompt_type, **kwargs):
-            called_types_3.append(prompt_type)
-            return real_get(prompt_type, **kwargs)
+        assert "storyboard" in called_types
+        assert copy_res["voiceover_script"]
 
+        # review 走 Reviewer.review
         from packages.application.viral_video.reviewer import Reviewer, ReviewResult
 
         pass_result = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[])
-        job.intent_result = intent_res
-        job.copy_result = copy_res
-        with (
-            patch.object(pl, "get_template", side_effect=spy_get_3),
-            patch.object(Reviewer, "review", return_value=pass_result) as mock_review,
-        ):
+        with patch.object(Reviewer, "review", return_value=pass_result) as mock_review:
             review_res = vv._step_review(job, copy_res)
-        # review 步骤内部直接调用 Reviewer.review，该方法被 mock，因此 get_template 不会被调用；
-        # 此处验证 Reviewer.review 被调用即可说明 review 步骤走通了。
-        assert mock_review.called, "_step_review 未调用 Reviewer.review"
+        assert mock_review.called
         assert isinstance(review_res, dict) and "passed" in review_res

tests/unit/test_vision_v4_prompt.py

@@ -1,240 +1,137 @@
 # -*- coding: utf-8 -*-
-"""vision v4 prompt / assembler 单元测试：
+"""vision v8 叙述优先 assembler / prompt 单元测试。
 
-- assembler 正确识别 v4 嵌套 schema 与旧扁平 schema
-- v4 product/person/store/other 四类输出组装出下游必出字段
-- 旧扁平 schema 行为不变
-- _prompt._resolve：DB 有 active prompt 时原样使用（不追加硬编码 schema）；
-  DB 无记录时回落到硬编码 JSON schema
+- assembler 输出仅 5 字段（type/name/brand/has_person/summary_markdown）
+- images / 老 products 两种顶层键都能解析
+- summary_markdown 正常时原样透传，不改写
+- summary_markdown 缺失时才用一句话基础兜底
+- _prompt：DB 有 active 模板原样使用，无记录回落到 prompts.py 默认 v8
 """
 
 from __future__ import annotations
 
+import sys
 import types
 from typing import Any
 
 import pytest
 from worker_app.tasks.vision import _prompt, assembler
 
-REQUIRED_KEYS = {
-    "name",
-    "brand",
-    "category",
-    "appearance",
-    "packaging",
-    "text_on_package",
-    "key_features",
-    "scene",
-    "mood",
-    "portrait_prompt",
-    "summary",
-    "_source",
-}
+# packages 层依赖 datetime.UTC（Python 3.11+）。开发机若为旧版本，prompt 相关用例
+# 在 CI（3.11）上正常执行，本地直接跳过，避免污染基线。
+_PY311 = sys.version_info >= (3, 11)
+requires_packages = pytest.mark.skipif(not _PY311, reason="packages 需要 Python 3.11+")
+
+REQUIRED_KEYS = {"type", "name", "brand", "has_person", "summary_markdown"}
 
 
-# ---------- schema 识别 ----------
+# ---------- 正常 v8：叙述原样透传 ----------
 
 
-def test_is_v4_schema_products_list() -> None:
-    assert assembler._is_v4_schema({"type": "product", "products": []})
-
-
-def test_is_v4_schema_type_only() -> None:
-    assert assembler._is_v4_schema({"type": "person"})
-
-
-def test_is_v4_schema_people_dict() -> None:
-    assert assembler._is_v4_schema({"people": {"has_person": True}})
-
-
-def test_is_not_v4_schema_flat() -> None:
-    assert not assembler._is_v4_schema({"has_person": True, "upper_wear": "T恤"})
-
-
-# ---------- v4 product ----------
-
-V4_PRODUCT: dict[str, Any] = {
-    "type": "product",
-    "scene": "白色背景产品图",
-    "mood": "清新专业",
-    "style": "商业产品摄影",
-    "colors": [{"hex": "#E60012", "name": "亮红色", "coverage": 0.6}],
-    "visible_text": [{"text": "OMO奥妙除菌除螨", "position": "瓶身正面"}],
-    "products": [
-        {
-            "product_name": "OMO奥妙除菌除螨洗衣液",
-            "brand": "OMO奥妙",
-            "category": "洗护",
-            "package_type": "瓶装",
-            "package_color": "亮红色瓶身",
-            "cap_type": "透明翻盖式按压瓶口",
-            "body_shape": "带侧面握持把手的竖款瓶身",
-            "label_design": "瓶身印十字盾牌图案",
-            "product_features": ["亮红色瓶装", "按压式瓶口", "十字盾牌标签"],
-            "key_selling_points": ["天然除菌除螨"],
-            "position": "main",
-        }
-    ],
-    "has_person": False,
-}
-
-
-def test_assemble_v4_product_fields() -> None:
-    r = assembler.assemble_result(0, V4_PRODUCT, ["OMO奥妙"])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "OMO奥妙除菌除螨洗衣液"
-    assert r["brand"] == "OMO奥妙"
-    assert r["category"] == "洗护"
-    assert "瓶装" in r["packaging"]
-    assert isinstance(r["key_features"], list) and r["key_features"]
-    assert any("除菌" in str(t) for t in r["text_on_package"])
-    assert len(r["portrait_prompt"]) >= 10
-    assert r["_source"] == "v2_fast_json_v4"
-
-
-def test_assemble_v4_product_multi_selects_main() -> None:
+def test_assemble_v8_store_passthrough() -> None:
+    md = "###店铺主体\n这是一家名为“御众堂”的线下门店内部，整体暖木色调……"
     fj = {
-        "type": "product",
-        "products": [
-            {"product_name": "次要商品", "brand": "B"},
-            {"product_name": "主商品", "brand": "A", "position": "main"},
-        ],
-    }
-    r = assembler.assemble_result(1, fj, [])
-    assert r["name"] == "主商品"
-
-
-# ---------- v4 person ----------
-
-V4_PERSON: dict[str, Any] = {
-    "type": "person",
-    "scene": "户外街拍",
-    "mood": "自信",
-    "style": "街拍",
-    "colors": [],
-    "visible_text": [],
-    "has_person": True,
-    "gender": "女",
-    "age_range": "青年",
-    "upper_wear": "白色V领短袖T恤",
-    "upper_color": "白色",
-    "lower_wear": "黑色高腰阔腿裤",
-    "lower_color": "黑色",
-    "dress_color": None,
-    "accessories": ["银色项链"],
-    "hairstyle": "黑色长直发",
-    "expression": "自信",
-    "pose": "侧身站立",
-    "outfit_style": "休闲日常",
-    "portrait_prompt": (
-        "一位年轻女性，身穿白色V领短袖T恤、黑色高腰阔腿裤，佩戴银色项链，"
-        "黑色长直发，神情自信，侧身站立，休闲日常风格，城市街拍场景"
-    ),
-    "products": [],
-}
-
-
-def test_assemble_v4_person() -> None:
-    r = assembler.assemble_result(0, V4_PERSON, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["category"] == "人物穿搭"
-    assert r["_source"] == "v2_fast_json_v5"
-    assert "T恤" in r["name"]
-    assert "年轻女性" in r["portrait_prompt"]
-    assert "项链" in r["portrait_prompt"]
-    assert isinstance(r["key_features"], list) and len(r["key_features"]) <= 8
-
-
-def test_assemble_v4_person_people_nested() -> None:
-    fj = {"type": "person", "people": {**V4_PERSON, "has_person": True}}
-    r = assembler.assemble_result(0, fj, [])
-    assert r["category"] == "人物穿搭"
-    assert "年轻女性" in r["portrait_prompt"]
-
-
-# ---------- v4 store ----------
-
-
-def test_assemble_v4_store() -> None:
-    fj = {
-        "type": "store",
-        "scene": "便利店内部",
-        "mood": "日常便民",
-        "style": "门店实拍",
-        "store_type": "社区便利店",
-        "store_layout": "纵深货架布局",
-        "brand_signage": "全家FamilyMart",
-        "visual_elements": ["红白主色调", "促销海报"],
-        "product_categories_visible": ["饮料", "零食"],
-        "promotion_elements": ["第二件半价海报"],
-        "atmosphere": "亲民生活化",
-        "has_person": False,
+        "images": [
+            {
+                "type": "store",
+                "name": "御众堂门店",
+                "brand": "御众堂",
+                "has_person": False,
+                "summary_markdown": md,
+            }
+        ]
     }
     r = assembler.assemble_result(0, fj, [])
     assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "社区便利店"
-    assert r["brand"] == "全家FamilyMart"
-    assert r["category"] == "门店场景"
-    assert any("饮料" in str(f) for f in r["key_features"])
-    assert "门店实拍" in r["portrait_prompt"]
+    assert r["type"] == "store"
+    assert r["name"] == "御众堂门店"
+    assert r["brand"] == "御众堂"
+    assert r["has_person"] is False
+    assert r["summary_markdown"] == md
+    assert "_source" not in r
 
 
-# ---------- v4 other ----------
-
-
-def test_assemble_v4_other() -> None:
-    fj = {"type": "other", "description": "海边日落风景", "scene": "海边", "mood": "宁静"}
-    r = assembler.assemble_result(0, fj, [])
-    assert REQUIRED_KEYS <= set(r.keys())
-    assert r["name"] == "海边日落风景"
-    assert r["category"] == "非产品图"
-
-
-# ---------- 旧扁平 schema 兼容 ----------
-
-
-def test_assemble_old_flat_person() -> None:
+def test_assemble_v8_product() -> None:
+    md = "这是一瓶洗衣液，亮红色瓶身配白色按压泵头，瓶身正面印着品牌标识……"
     fj = {
-        "has_person": True,
-        "gender": "男",
-        "age_range": "中年",
-        "upper_wear": "西装",
-        "upper_color": "深灰色",
-        "lower_wear": "西裤",
-        "lower_color": "黑色",
-        "accessories": ["手表"],
-        "hairstyle": "短发",
-        "expression": "严肃",
-        "scene": "办公室",
-        "style": "商务",
-        "mood": "专业",
+        "images": [{"type": "product", "name": "洗衣液", "brand": "OMO", "has_person": False, "summary_markdown": md}]
+    }
+    r = assembler.assemble_result(0, fj, ["OMO"])
+    assert r["type"] == "product"
+    assert r["summary_markdown"] == md
+
+
+def test_assemble_v8_person() -> None:
+    md = "画面里是一位年轻女性，穿白色T恤、黑色阔腿裤，神情自信……"
+    fj = {"images": [{"type": "person", "name": "年轻女性", "brand": "", "has_person": True, "summary_markdown": md}]}
+    r = assembler.assemble_result(0, fj, [])
+    assert r["type"] == "person"
+    assert r["has_person"] is True
+    assert r["summary_markdown"] == md
+
+
+def test_assemble_v8_scene() -> None:
+    fj = {
+        "images": [
+            {"type": "scene", "name": "海边日落", "brand": "", "has_person": False, "summary_markdown": "海边……"}
+        ]
     }
     r = assembler.assemble_result(0, fj, [])
+    assert r["type"] == "scene"
+
+
+# ---------- 顶层 products 老键兼容（assembler 层）----------
+
+
+def test_assemble_top_level_products_key() -> None:
+    fj = {"products": [{"type": "store", "name": "门店", "brand": "御众堂", "summary_markdown": "门店……"}]}
+    r = assembler.assemble_result(0, fj, [])
+    assert r["brand"] == "御众堂"
+    assert r["type"] == "store"
+
+
+# ---------- 字段缺失的异常兜底 ----------
+
+
+def test_assemble_missing_summary_uses_basic_fallback() -> None:
+    fj = {"images": [{"type": "store", "name": "御众堂门店", "brand": "御众堂", "has_person": False}]}
+    r = assembler.assemble_result(0, fj, [])
     assert REQUIRED_KEYS <= set(r.keys())
-    assert "中年男性" in r["portrait_prompt"]
-    assert r["_source"] == "v2_fast_json"
+    assert r["summary_markdown"]
+    assert "御众堂" in r["summary_markdown"]
+    assert r.get("_source") == "summary_missing"
 
 
-def test_assemble_old_flat_product() -> None:
-    fj = {
-        "has_person": False,
-        "product_name": "口红",
-        "brand": "Dior",
-        "category": "美妆",
-        "colors": ["红色"],
-        "scene": "通用",
-        "style": "商业",
-        "mood": "高级",
-    }
-    r = assembler.assemble_result(0, fj, ["Dior"])
-    assert r["name"] == "口红"
-    assert r["brand"] == "Dior"
-    assert r["text_on_package"] == ["Dior"]
+def test_assemble_invalid_type_defaults_scene() -> None:
+    fj = {"images": [{"type": "weird", "name": "x", "summary_markdown": ""}]}
+    r = assembler.assemble_result(0, fj, [])
+    assert r["type"] == "scene"
+    assert r["summary_markdown"]  # basic fallback
+
+
+def test_assemble_empty_fast_json_uses_ocr_hint() -> None:
+    r = assembler.assemble_result(0, {}, ["御众堂"])
+    assert REQUIRED_KEYS <= set(r.keys())
+    assert "御众堂" in r["name"]
+    assert r.get("_source") == "empty_fast_json"
 
 
 def test_assemble_none_input() -> None:
     r = assembler.assemble_result(0, None, [])
     assert REQUIRED_KEYS <= set(r.keys())
+    assert r["type"] == "scene"
+
+
+# ---------- 布尔归一化 ----------
+
+
+def test_coerce_bool() -> None:
+    assert assembler._coerce_bool(True) is True
+    assert assembler._coerce_bool(1) is True
+    assert assembler._coerce_bool("true") is True
+    assert assembler._coerce_bool(False) is False
+    assert assembler._coerce_bool(0) is False
+    assert assembler._coerce_bool("否") is False
 
 
 # ---------- _prompt 解析 ----------
@@ -247,45 +144,48 @@ def _clear_prompt_cache() -> Any:
     _prompt.invalidate_cache()
 
 
-def _fake_tpl(system_prompt: str = "v4 system prompt 只返回JSON") -> Any:
+def _fake_tpl(system_prompt: str = "DB_V8_PROMPT_XYZ") -> Any:
     return types.SimpleNamespace(
         system_prompt=system_prompt,
-        user_prompt_template="分析 {image_count} 张图",
-        version=4,
+        user_prompt_template="地址：{image_url}，OCR：{ocr_text}",
+        version=8,
     )
 
 
-def test_resolve_uses_db_prompt_without_append(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl("DB_V4_PROMPT_XYZ"))
-    sys_prompt, user_prompt = _prompt.resolve_fast_prompt()
-    assert sys_prompt == "DB_V4_PROMPT_XYZ"
-    assert "DB_V4_PROMPT_XYZ" not in _prompt._FAST_JSON_APPEND  # sanity: 旧append是另一段文本
-    assert "分析 1 张图" in user_prompt
+@requires_packages
+def test_resolve_uses_db_prompt(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl())
+    sys_prompt, user_prompt = _prompt.resolve_fast_prompt("http://img", "御众堂")
+    assert sys_prompt == "DB_V8_PROMPT_XYZ"
+    assert "http://img" in user_prompt
+    assert "御众堂" in user_prompt
 
 
-def test_resolve_pro_uses_db_prompt_without_append(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl("DB_V4_PRO_PROMPT"))
+@requires_packages
+def test_resolve_pro_uses_db_prompt(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(_prompt, "_load_db_template", lambda: _fake_tpl("DB_PRO_PROMPT"))
     sys_prompt, _ = _prompt.resolve_pro_prompt()
-    assert sys_prompt == "DB_V4_PRO_PROMPT"
-    assert "【输出格式要求】" not in sys_prompt
+    assert sys_prompt == "DB_PRO_PROMPT"
 
 
-def test_resolve_falls_back_when_no_db(monkeypatch: pytest.MonkeyPatch) -> None:
+@requires_packages
+def test_resolve_falls_back_to_default(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(_prompt, "_load_db_template", lambda: None)
-    sys_prompt, user_prompt = _prompt.resolve_fast_prompt()
-    assert sys_prompt == _prompt._FAST_JSON_SCHEMA
-    assert user_prompt == _prompt.DEFAULT_FAST_USER
+    default = _prompt._default_template()
+    sys_prompt, _ = _prompt.resolve_fast_prompt()
+    assert sys_prompt == default["system_prompt"]
 
 
+@requires_packages
 def test_resolve_caches(monkeypatch: pytest.MonkeyPatch) -> None:
     calls = {"n": 0}
 
     def _load() -> Any:
         calls["n"] += 1
-        return _fake_tpl("CACHED_PROMPT")
+        return _fake_tpl("CACHED")
 
     monkeypatch.setattr(_prompt, "_load_db_template", _load)
     s1, _ = _prompt.resolve_fast_prompt()
     s2, _ = _prompt.resolve_fast_prompt()
-    assert s1 == s2 == "CACHED_PROMPT"
+    assert s1 == s2 == "CACHED"
     assert calls["n"] == 1

tests/unit/test_xml_parser.py

@@ -0,0 +1,49 @@
+"""xml_parser CDATA 剥离单元测试。"""
+
+from packages.application.viral_video.xml_parser import find_all, text_of
+
+XML = """<script>
+<copy_display_markdown><![CDATA[# 标题
+
+这是第一段，含**加粗**和[链接](https://a.com)。
+
+第二行，保留换行。]]></copy_display_markdown>
+<voiceover>口播不带 CDATA，保持原样。</voiceover>
+<visual><![CDATA[画面：产品特写，光线柔和]]></visual>
+<action_details><![CDATA[未闭合标签里的 CDATA 也要剥离]]></action_details>
+</script>"""
+
+
+def test_text_of_strips_cdata_with_markdown_newlines():
+    text = text_of(XML, "copy_display_markdown")
+    assert not text.startswith("<![CDATA[")
+    assert not text.endswith("]]>")
+    assert "# 标题" in text
+    assert "**加粗**" in text
+    assert "[链接](https://a.com)" in text
+    # markdown 换行被保留
+    assert "\n\n第二行" in text
+
+
+def test_plain_text_unchanged():
+    assert text_of(XML, "voiceover") == "口播不带 CDATA，保持原样。"
+
+
+def test_other_cdata_fields_stripped():
+    assert text_of(XML, "visual") == "画面：产品特写，光线柔和"
+
+
+def test_unclosed_tag_cdata_stripped():
+    # action_details 没有闭合标签，走未闭合兜底分支
+    node = find_all(XML, "action_details")[0]
+    assert node["text"] == "未闭合标签里的 CDATA 也要剥离"
+
+
+def test_no_cdata_returns_original():
+    xml = "<copy_display_markdown>普通内容]]> 残留结尾</copy_display_markdown>"
+    # 非完整 CDATA 包裹不应被误剥离
+    assert text_of(xml, "copy_display_markdown") == "普通内容]]> 残留结尾"
+
+
+def test_missing_tag_default():
+    assert text_of(XML, "nope", default="缺省") == "缺省"