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

- _run_render_pipeline: copy_result为空时直接ValueError报错，不再补跑意图+编导
- review自动重写失败不再从意图重跑，直接报错终止
- generate-copy: 移除意图解析后单独_save_job，等mark_copy_generated后一次性原子写入
- confirm-copy API: 加copy_result非空校验，拦截脏数据入队
- 加日志: render入口记录copy_result来源和shots数量
- 审核异常处理区分ValueError(业务错误)/其他异常(LLM调用失败可继续)

apps/api/app/api/routes/viral_video.py

@@ -341,6 +341,9 @@ def confirm_copy(
         raise HTTPException(status_code=403, detail="无权操作此任务")
     if job.status != ViralVideoStatus.COPY_GENERATED:
         raise HTTPException(status_code=409, detail=f"任务当前状态 {job.status} 不能确认文案（需 copy_generated）")
+    # #2218: 额外校验 copy_result 完整性，防止孤儿/脏数据进入渲染
+    if not isinstance(job.copy_result, dict) or not job.copy_result:
+        raise HTTPException(status_code=409, detail="文案数据缺失，请先点击「生成文案」")
 
     # 积分预扣（已扣过/重试任务跳过）
     from app.config import settings as _settings

apps/worker/worker_app/tasks/viral_video.py

@@ -421,10 +421,14 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
             render_system_prompt,
             render_user_prompt,
         )
-        from packages.shared.ai_service import call_llm
+        from packages.shared.ai_client import get_doubao_client
     except ImportError:
         return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""}
 
+    _llm_client = get_doubao_client()
+    if not _llm_client.is_available:
+        return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""}
+
     products_summary = ""
     products = (image_analysis or {}).get("products", []) or []
     for p in products:
@@ -479,13 +483,13 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
     for _m, _lbl in [(_fast, "fast"), (_pro, "pro-fallback")]:
         try:
             logger.info("[爆款视频] 意图解析 model=%s label=%s", _m, _lbl)
-            raw = call_llm(
+            raw = _llm_client.chat_completion(
                 [{"role": "system", "content": system}, {"role": "user", "content": user}],
                 temperature=0.4,
                 max_tokens=1024,
                 model=_m,
                 timeout=60,
-            )  # #2180: 意图解析 LLM 实测需更长响应，原25s太紧
+            )  # #2180/#2215: 直接用 client.chat_completion 传 messages list，不再走 call_llm 字符串包装
             if not raw:
                 continue
             parsed = _parse(raw)
@@ -825,10 +829,14 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
             GLOBAL_CONSTRAINTS,
             NEGATIVE_RULES,
         )
-        from packages.shared.ai_service import call_llm
+        from packages.shared.ai_client import get_doubao_client
     except ImportError:
         return _fallback_script(job)
 
+    _llm_client2 = get_doubao_client()
+    if not _llm_client2.is_available:
+        return _fallback_script(job)
+
     products_summary = _build_products_summary(image_analysis)
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
 
@@ -864,7 +872,7 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
 
     def _try_gen(model: str, temp: float, max_tok: int, label: str, tmo: int = 25):
         logger.info("[爆款视频] 编导脚本生成 model=%s label=%s timeout=%d", model, label, tmo)
-        raw = call_llm(
+        raw = _llm_client2.chat_completion(
             [{"role": "system", "content": system_tpl}, {"role": "user", "content": user}],
             temperature=temp,
             max_tokens=max_tok,
@@ -902,18 +910,19 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
     _s = get_shared_settings()
     _fast = _s.doubao_fast_model
     _pro = getattr(_s, "doubao_model", None) or _fast
+    _script_fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "150"))
+    _script_pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "150"))
     try:
-        # 第一次：快模型 25s
-        normalized = _try_gen(_fast, 0.8, 2500, "fast-first", tmo=90)
+        # #2217: doubao-seed-2-1-pro生成长编导脚本高峰期>90s，上调到150s，支持ENV覆盖
+        normalized = _try_gen(_fast, 0.8, 2500, "fast-first", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
-        # #2183: 实测pro 1500tok输出需75.8s，单次timeout提到90s
-        normalized = _try_gen(_fast, 0.6, 3200, "fast-retry", tmo=90)
+        normalized = _try_gen(_fast, 0.6, 3200, "fast-retry", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
-        # 第三次：用主力模型兜底，给 120s
+        # 第三次：用主力模型兜底
         if _pro and _pro != _fast:
-            normalized = _try_gen(_pro, 0.7, 3500, "pro-fallback", tmo=120)
+            normalized = _try_gen(_pro, 0.7, 3500, "pro-fallback", tmo=_script_pro_tmo)
             if normalized is not None:
                 return normalized
         logger.warning("[爆款视频] 编导脚本三次都未生成合格结果，使用兜底脚本")
@@ -1670,7 +1679,7 @@ def run_viral_video_generate_copy(self: Task, job_id: str) -> dict:
         image_analysis = job.image_analysis or {"products": []}
         intent_result = _step_intent_parsing(job, image_analysis)
         job.intent_result = intent_result
-        _save_job(repo, job, session)
+        # #2218: 不在意图解析后单独落库，等 copy_result 生成后与 mark_copy_generated 一起原子写入
         _emit_progress(job_id, ViralVideoStage.INTENT_PARSING, 35.0, "意图解析完成")
 
         # 阶段：编导脚本生成（核心耗时环节，已用快模型）
@@ -1728,7 +1737,8 @@ def run_viral_video_generate_copy(self: Task, job_id: str) -> dict:
     except Retry:
         raise
     except Exception as e:
-        logger.error("[爆款视频][阶段2] 异常: %s", e, exc_info=True)
+        logger.error("[爆款视频][阶段2] 异常 job_id=%s: %s", job_id, e, exc_info=True)
+        # #2218: 阶段2任何异常都标记为 failed（由 _mark_failed_and_notify 处理），前端提示重试
         _mark_failed_and_notify(job_id, session, None, None, str(e), ViralVideoStage.SCRIPT_GENERATION)
         return {"ok": False, "job_id": job_id, "error": str(e)}
     finally:
@@ -1887,16 +1897,26 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
     阶段2 generate-copy 已把 LLM 深度审核后置，这里在 TTS 前做最终审核（不通过则自动重写1次）。
     所有阶段通过 _set_stage 持久化 current_stage/phase_message。
     """
-    image_analysis = job.image_analysis or {"products": []}
 
-    # 如果没有 copy_result（旧数据/失败重试），现场补生成（意图+脚本，不走 LLM 审核，出片前会统一做）
+    # #2218: render 流程严禁补生成意图+编导脚本。copy_result 必须由 generate-copy 提前准备好；
+    # 若缺失说明 generate-copy 未完成或数据丢失，直接报错让用户重新点「生成文案」。
     copy_result = job.copy_result
+    _copy_src = "db"
     if not isinstance(copy_result, dict) or not copy_result:
-        _set_stage(job, repo, session, ViralVideoStage.SCRIPT_GENERATION, "正在补生成编导脚本...")
-        intent = job.intent_result or _step_intent_parsing(job, image_analysis)
-        copy_result = _step_script_generation(job, intent, image_analysis)
-        job.mark_copy_generated(copy_result)
-        _save_job(repo, job, session)
+        logger.error(
+            "[爆款视频][阶段3] copy_result 为空或无效，无法进入渲染流程。job_id=%s status=%s intent_len=%d，请重新触发「生成文案」",
+            job_id,
+            job.status,
+            len((job.intent_result or {}) if isinstance(job.intent_result, dict) else {}),
+        )
+        raise ValueError("文案数据缺失，请先点击「生成文案」完成文案生成后再生成视频")
+    logger.info(
+        "[爆款视频][阶段3] 进入渲染流程 job_id=%s copy_result_shots=%d copy_result_len=%d source=%s",
+        job_id,
+        len((copy_result.get("shots") or [])),
+        len(str(copy_result)),
+        _copy_src,
+    )
 
     # 出片前 LLM 深度合规审核（#2134 问题7：审核从阶段2后置到这里，不阻塞前端预览脚本）
     _set_stage(job, repo, session, ViralVideoStage.REVIEW, "正在进行出片前合规审核...")
@@ -1909,12 +1929,18 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
             if isinstance(rewritten, dict) and rewritten:
                 copy_result = rewritten
             else:
-                intent = job.intent_result or _step_intent_parsing(job, image_analysis)
-                copy_result = _step_script_generation(job, intent, image_analysis)
-                _step_review(job, copy_result)
+                # #2218: 审核重写失败不再从意图解析重跑，直接报错让用户重新生成文案
+                logger.error(
+                    "[爆款视频][阶段3] 合规审核未通过且自动重写失败 job_id=%s，终止渲染",
+                    job_id,
+                )
+                raise ValueError("文案合规审核未通过，请修改文案后重试或重新生成文案")
             job.copy_result = copy_result
             job.generated_copy_text = copy_result.get("voiceover_script", "") or ""
             _save_job(repo, job, session)
+    except ValueError:
+        # #2218: 审核未通过/文案缺失的业务异常，不继续出片，向上抛出
+        raise
     except Exception as e:
         logger.warning("[爆款视频][阶段3] 合规审核异常，继续出片: %s", e)
     _emit_progress(job_id, ViralVideoStage.REVIEW, 70.0, "合规审核完成")

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

@@ -1,6 +1,13 @@
 # -*- coding: utf-8 -*-
-"""V2 prompt 解析：优先读后台 viral_video_prompt_templates 配置，30s TTL 热加载；
-DB 不可用/读到默认XML模板时，fallback 到硬编码 JSON schema prompt。
+"""V2 prompt 解析：优先读后台 viral_video_prompt_templates 表（prompt_type='image_analysis'
+且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到纯硬编码 JSON schema prompt。
+
+规则（简单直接，不做字符串匹配判断）：
+- 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
 """
 
 from __future__ import annotations
@@ -12,9 +19,9 @@ from typing import Any
 
 logger = logging.getLogger(__name__)
 
-# ---- 默认硬编码 prompt（DB 不可用或读到默认XML模板时使用） ----
+# ---- 纯硬编码 JSON schema（DB 无有效配置时全量使用） ----
 
-DEFAULT_FAST_SYSTEM = (
+_FAST_JSON_SCHEMA = (
     "你是图片结构化识别器。严格按下方 JSON schema 返回一个对象，不要任何解释、"
     "不要markdown、不要代码块、不要前后缀文字。字段值不确定时填 null 或空数组。\n"
     "{\n"
@@ -40,12 +47,13 @@ DEFAULT_FAST_SYSTEM = (
     '  "pattern": "图案，如纯色/条纹/波点/格子/印花/碎花/Logo等",\n'
     '  "colors": ["主色数组"],\n'
     '  "mood": "整体氛围/情绪，如清新/活力/高级/温暖/冷峻/甜美/复古等"\n'
-    "}"
+    "}\n\n"
+    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
 )
 DEFAULT_FAST_USER = "识别这张图片的人物穿搭与主体信息，只返回JSON对象。"
 
-DEFAULT_PRO_SYSTEM = (
-    "你是图片分析专家。严格按下方 JSON schema 返回一个对象，不要解释、不要markdown、不要代码块。\n"
+_PRO_JSON_SCHEMA = (
+    "你是图片分析专家。严格按下方 JSON schema 返回一个对象，不要解释、不要markdown、不要代码块、不要XML标签。\n"
     "{\n"
     '  "has_person": true/false,\n'
     '  "gender": "男"/"女"/null,\n'
@@ -61,52 +69,99 @@ DEFAULT_PRO_SYSTEM = (
     '  "product_name": "产品名，非产品图填null",\n'
     '  "brand": "品牌，无则null",\n'
     '  "key_features": ["核心特征数组，3-6个短语"]\n'
-    "}"
+    "}\n\n"
+    "你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释、XML标签或markdown。"
 )
 DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。"
 
-# 用户自定义 prompt 末尾追加的硬约束
-_JSON_TAIL_FAST = "\n\n你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释或markdown。"
-_JSON_TAIL_PRO = "\n\n你必须只返回一个合法的JSON对象，不要输出任何其他文字、解释或markdown。"
+# 保留旧 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。"
+)
 
-# 默认模板特征头（用于识别是否是内置XML模板）
-_DEFAULT_XML_MARKER = "你是电商商品视觉分析师"
+_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_TTL = 30.0
 
 
-def _load_template() -> Any | None:
-    """从DB读image_analysis模板，失败返回None。"""
+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无记录或异常。"""
     try:
-        from packages.application.viral_video.prompt_loader import get_template
+        from packages.application.viral_video.prompt_loader import _load_from_db
 
-        return get_template("image_analysis")
+        return _load_from_db("image_analysis")
     except Exception as e:
-        logger.warning("[vision.v2] 读取后台prompt配置失败: %s", e)
+        logger.warning("[vision.v2] 查询DB prompt配置失败: %s", e)
         return None
 
 
-def _is_custom(template: Any) -> bool:
-    """判断读到的模板是不是用户自定义的（不是内置默认XML长prompt）。"""
-    if not template:
-        return False
-    sp = getattr(template, "system_prompt", "") or ""
-    # 默认模板开头是"你是电商商品视觉分析师..."，XML格式，不适合qwen+JSON
-    if _DEFAULT_XML_MARKER in sp[:30]:
-        return False
-    # 其他有内容的system_prompt视为用户自定义
-    return bool(sp.strip())
+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 resolve_fast_prompt() -> tuple[str, str]:
-    """返回 (system_prompt, user_prompt) 给 qwen3.8-flash fast 路径。"""
     return _resolve("fast")
 
 
 def resolve_pro_prompt() -> tuple[str, str]:
-    """返回 (system_prompt, user_prompt) 给 qwen3.7-plus fallback 路径。"""
     return _resolve("pro")
 
 
@@ -118,32 +173,30 @@ def _resolve(kind: str) -> tuple[str, str]:
         if hit and now - hit[0] < _CACHE_TTL:
             return hit[1]
 
-    default_sys = DEFAULT_FAST_SYSTEM if kind == "fast" else DEFAULT_PRO_SYSTEM
+    default_sys = _FAST_JSON_SCHEMA if kind == "fast" else _PRO_JSON_SCHEMA
     default_user = DEFAULT_FAST_USER if kind == "fast" else DEFAULT_PRO_USER
-    tail = _JSON_TAIL_FAST if kind == "fast" else _JSON_TAIL_PRO
 
     sys_prompt = default_sys
     usr_prompt = default_user
     try:
-        tpl = _load_template()
-        if _is_custom(tpl):
-            custom_sys = (getattr(tpl, "system_prompt", "") or "").strip()
-            custom_user_tpl = getattr(tpl, "user_prompt_template", "") or ""
-            if custom_sys:
-                sys_prompt = custom_sys + tail
-            if custom_user_tpl:
-                # V2 是单图调用，简单替换几个常用占位符；缺键原样保留
-                usr_prompt = (
-                    custom_user_tpl.replace("{image_count}", "1")
-                    .replace("{industry}", "通用")
-                    .replace("{image_urls}", "")
-                    .strip()
+        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),
                 )
-                if not usr_prompt:
-                    usr_prompt = default_user
-            logger.info("[vision.v2] 使用后台自定义prompt (kind=%s version=%s)", kind, getattr(tpl, "version", "?"))
+            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] 解析后台prompt失败，使用默认: %s", e)
+        logger.warning("[vision.v2] 解析DB prompt异常，使用默认: %s", e)
 
     with _cache_lock:
         _cache[cache_key] = (now, (sys_prompt, usr_prompt))

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

@@ -1,5 +1,8 @@
 # -*- coding: utf-8 -*-
-"""把 fast_json VLM 输出 + OCR 文本组装为与旧 _normalize() 完全一致的 dict。
+"""把 fast_json VLM 输出 + OCR 文本组装为下游兼容的 product dict。
+
+v4 schema: DB prompt完全控制输出格式，可能是v4嵌套schema(type/products/people/store_info)
+或旧扁平schema(has_person/upper_wear/product_name/brand等)。assembler兼容两种格式。
 
 目标：下游（信任链t2i/intent_parsing/script_generation）零改动。
 必出字段：name, brand, category, appearance, packaging, text_on_package,
@@ -10,29 +13,16 @@ from __future__ import annotations
 
 from typing import Any
 
-# ---------- portrait_prompt 模板 ----------
-# 目标：60-100 字的人物穿搭描述，用于 Seedream 纯文生图。要求具体、风格化、视觉细节丰富。
-# 旧 VLM 输出格式参考："一位25岁左右的亚洲女性，身穿白色V领短袖T恤，黑色高腰阔腿裤，
-# 搭配银色项链，长发披肩，表情自信，街拍风格，阳光明媚的城市街头"
-
 
 def _join_parts(*parts: str | None) -> str:
     return "".join(p for p in parts if p)
 
 
-_AGE_PREFIX = {
-    "青年": "年轻",
-    "中年": "中年",
-    "老年": "老年",
-}
-# gender 后缀
+_AGE_PREFIX = {"青年": "年轻", "中年": "中年", "老年": "老年"}
 _GENDER_WORD = {"男": "男性", "女": "女性"}
 
 
-def _person_subject(fj: dict[str, Any]) -> str:
-    """人物主语：年轻女性 / 中年男性 / 少女 / 小男孩 / 人物 等。"""
-    gender = fj.get("gender") or ""
-    age = fj.get("age_range") or ""
+def _person_subject(gender: str, age: str) -> str:
     gw = _GENDER_WORD.get(gender, "")
     if age == "儿童":
         if gender == "女":
@@ -52,8 +42,147 @@ def _person_subject(fj: dict[str, Any]) -> str:
     return f"{prefix}人物" if prefix else "人物"
 
 
-def _build_wear_sentence(fj: dict[str, Any]) -> str:
-    """穿搭段：上装+下装/连衣裙，带颜色+材质+图案。"""
+def _build_wear_from_v4(p: dict) -> str:
+    """v4 person schema: upper_wear/upper_color/lower_wear/lower_color/dress_color"""
+    upper = p.get("upper_wear") or ""
+    upper_color = p.get("upper_color") or ""
+    lower = p.get("lower_wear") or ""
+    lower_color = p.get("lower_color") or ""
+    dress_color = p.get("dress_color") or ""
+    is_dress = ("连衣裙" in upper) or ("裙" in upper and not lower)
+    if is_dress:
+        c = dress_color or upper_color
+        return f"身穿{c}{upper}" if c else f"身穿{upper}"
+    parts = []
+    if upper:
+        up = f"{upper_color}{upper}" if upper_color else upper
+        parts.append(f"上身{up}")
+    if lower:
+        lo = f"{lower_color}{lower}" if lower_color else lower
+        parts.append(f"下身{lo}")
+    return "，".join(parts)
+
+
+def _build_portrait_prompt_from_v4(p: dict) -> str:
+    """v4 person: 直接用portrait_prompt字段；没有就拼"""
+    direct = p.get("portrait_prompt")
+    if direct and len(direct) >= 10:
+        return direct
+    subject = _person_subject(p.get("gender", ""), p.get("age_range", ""))
+    wear = _build_wear_from_v4(p)
+    acc = p.get("accessories") or []
+    if isinstance(acc, str):
+        acc = [acc]
+    acc_str = "，佩戴" + "、".join(str(a) for a in acc if a) if acc else ""
+    hair = p.get("hairstyle") or ""
+    expr = p.get("expression") or ""
+    pose = p.get("pose") or ""
+    style = p.get("outfit_style") or p.get("style") or ""
+    scene = p.get("scene") or ""
+    mood = p.get("mood") or ""
+    details = []
+    if hair:
+        details.append(hair)
+    if expr and expr not in ("自然", "平静"):
+        details.append(f"神情{expr}")
+    if pose and pose not in ("站立",):
+        details.append(pose)
+    style_parts = []
+    if style:
+        style_parts.append(style)
+    if mood:
+        style_parts.append(mood)
+    if scene and scene not in ("通用",):
+        style_parts.append(scene)
+    pieces = [f"一位{subject}"]
+    if wear:
+        pieces.append(wear)
+    if acc_str:
+        pieces.append(acc_str.lstrip("，"))
+    if details:
+        pieces.append("，".join(details))
+    pieces.append(("".join(style_parts) + "风格") if style_parts else "人像写真")
+    full = "，".join(p for p in pieces if p)
+    if len(full) < 40:
+        full += "，自然光线下人像特写，画面清晰"
+    if len(full) > 120:
+        full = full[:120].rstrip("，") + "。"
+    return full
+
+
+def _build_product_prompt_from_v4(prod: dict, top: dict) -> str:
+    """v4 product: 拼商品视觉描述prompt（用于AI生图参考）"""
+    name = prod.get("product_name") or "商品"
+    brand = prod.get("brand") or ""
+    lead = f"{brand} {name}" if brand and brand not in name else name
+    pkg_color = prod.get("package_color") or ""
+    pkg_type = prod.get("package_type") or ""
+    cap = prod.get("cap_type") or ""
+    body = prod.get("body_shape") or ""
+    features = prod.get("product_features") or []
+    sell = prod.get("key_selling_points") or []
+    colors = top.get("colors") or []
+    style = top.get("style") or ""
+    scene = top.get("scene") or ""
+    mood = top.get("mood") or ""
+
+    parts = [lead]
+    desc = []
+    if pkg_color:
+        desc.append(pkg_color)
+    if pkg_type:
+        desc.append(pkg_type)
+    if cap and len(desc) < 3:
+        desc.append(f"配{cap}")
+    if body and len(desc) < 3:
+        desc.append(body)
+    if desc:
+        parts.append("，".join(desc))
+    if features:
+        core = [str(f) for f in features[:3] if f and len(str(f)) <= 25]
+        if core:
+            parts.append("；".join(core))
+    if sell:
+        s = [str(x) for x in sell[:2] if x]
+        if s:
+            parts.append("突出" + "、".join(s))
+    cnames = []
+    for cc in colors:
+        if isinstance(cc, dict) and cc.get("name"):
+            cnames.append(cc["name"])
+        elif isinstance(cc, str):
+            cnames.append(cc)
+    cnames = cnames[:3]
+    if cnames:
+        parts.append("、".join(cnames) + "主色")
+    if style:
+        parts.append(style)
+    if mood:
+        parts.append(mood)
+    if scene and not any(k in scene for k in ("白色背景", "纯色", "通用")):
+        parts.append(scene)
+    parts.append("产品特写，画面清晰")
+    prompt = "，".join(p for p in parts if p)
+    return prompt if len(prompt) >= 10 else "产品展示图，特写镜头"
+
+
+def _is_v4_schema(fj: dict) -> bool:
+    """判断是v4嵌套schema还是旧扁平schema"""
+    return (
+        isinstance(fj.get("products"), list)
+        or fj.get("type") in ("product", "store", "person", "other")
+        or isinstance(fj.get("people"), dict)
+    )
+
+
+# ---------- 旧扁平schema兼容（保留原逻辑） ----------
+
+
+def _person_subject_old(fj: dict) -> str:
+    return _person_subject(fj.get("gender", ""), fj.get("age_range", ""))
+
+
+def _build_wear_sentence_old(fj: dict) -> str:
     upper = fj.get("upper_wear") or ""
     upper_color = fj.get("upper_color") or ""
     lower = fj.get("lower_wear") or ""
@@ -61,7 +190,6 @@ def _build_wear_sentence(fj: dict[str, Any]) -> str:
     dress_color = fj.get("dress_color") or ""
     material = fj.get("material") or ""
     pattern = fj.get("pattern") or ""
-
     is_dress = ("连衣裙" in upper) or ("裙" in upper and not lower)
     if is_dress:
         c = dress_color or upper_color
@@ -71,25 +199,22 @@ def _build_wear_sentence(fj: dict[str, Any]) -> str:
         if pattern and pattern not in wear and pattern != "纯色":
             wear += f"，{pattern}图案"
         return f"身穿{wear}"
-
-    parts: list[str] = []
+    parts = []
     if upper:
         up = f"{upper_color}{upper}" if upper_color else upper
         if material and material not in up:
             up = f"{material}{up}"
         if pattern and pattern != "纯色" and pattern not in up:
             up += f"（{pattern}）"
-        parts.append(f"上身{up}" if up else "")
+        parts.append(f"上身{up}")
     if lower:
         lo = f"{lower_color}{lower}" if lower_color else lower
-        parts.append(f"下身{lo}" if lo else "")
+        parts.append(f"下身{lo}")
     return "，".join(p for p in parts if p)
 
 
-def _build_portrait_prompt(fj: dict[str, Any]) -> str:
-    """组装最终 portrait_prompt（目标 60-100 字，用于 Seedream 纯文生图）。"""
+def _build_portrait_prompt_old(fj: dict) -> str:
     if not fj.get("has_person"):
-        # 非人像：用商品+场景+mood 拼一段
         name = fj.get("product_name") or "商品"
         brand = fj.get("brand") or ""
         colors = fj.get("colors") or []
@@ -101,7 +226,15 @@ def _build_portrait_prompt(fj: dict[str, Any]) -> str:
             pieces.append(brand)
         pieces.append(name)
         if colors:
-            pieces.append("、".join(colors[:3]) + "配色")
+            cnames = []
+            for c in colors:
+                if isinstance(c, dict):
+                    cnames.append(c.get("name", ""))
+                elif isinstance(c, str):
+                    cnames.append(c)
+            cnames = [c for c in cnames if c][:3]
+            if cnames:
+                pieces.append("、".join(cnames) + "配色")
         if style:
             pieces.append(style + "风格")
         if mood:
@@ -111,40 +244,32 @@ def _build_portrait_prompt(fj: dict[str, Any]) -> str:
         pieces.append("产品特写")
         prompt = "，".join(p for p in pieces if p)
         return prompt if len(prompt) >= 10 else "产品展示图，特写镜头"
-
-    subject = _person_subject(fj)
-    wear = _build_wear_sentence(fj)
-
+    subject = _person_subject_old(fj)
+    wear = _build_wear_sentence_old(fj)
     accessories = fj.get("accessories") or []
     if isinstance(accessories, str):
         accessories = [accessories]
-    acc_str = ""
-    if accessories:
-        acc_str = "，佩戴" + "、".join(str(a) for a in accessories if a)
-
+    acc_str = "，佩戴" + "、".join(str(a) for a in accessories if a) if accessories else ""
     hairstyle = fj.get("hairstyle") or ""
     expression = fj.get("expression") or ""
     pose = fj.get("pose") or ""
     style = fj.get("style") or ""
     scene = fj.get("scene") or ""
     mood = fj.get("mood") or ""
-
-    detail_parts: list[str] = []
+    detail_parts = []
     if hairstyle:
         detail_parts.append(hairstyle)
     if expression and expression not in ("自然", "平静"):
         detail_parts.append(f"神情{expression}")
     if pose and pose not in ("站立",):
         detail_parts.append(pose)
-
-    style_parts: list[str] = []
+    style_parts = []
     if style:
         style_parts.append(style)
     if mood:
         style_parts.append(mood)
     if scene and scene not in ("通用",):
         style_parts.append(scene)
-
     pieces = [f"一位{subject}"]
     if wear:
         pieces.append(wear)
@@ -152,53 +277,40 @@ def _build_portrait_prompt(fj: dict[str, Any]) -> str:
         pieces.append(acc_str.lstrip("，"))
     if detail_parts:
         pieces.append("，".join(detail_parts))
-    if style_parts:
-        # 风格词之间不用逗号，用空格紧凑
-        pieces.append("".join(style_parts) + "风格")
-    else:
-        pieces.append("人像写真")
-
+    pieces.append("".join(style_parts) + "风格" if style_parts else "人像写真")
     full = "，".join(p for p in pieces if p)
-    # 过短补充镜头词
     if len(full) < 40:
         full += "，自然光线下人像特写，画面清晰"
-    # 过长截断
     if len(full) > 120:
         full = full[:120].rstrip("，") + "。"
     return full
 
 
-# ---------- 商品字段 ----------
-
-
-def _infer_name(fj: dict[str, Any], ocr_texts: list[str]) -> str:
+def _infer_name_old(fj: dict, ocr_texts: list[str]) -> str:
     pname = fj.get("product_name")
     if pname and pname != "未识别":
         return str(pname)
-    # 人物图 → name 用穿搭主件
     if fj.get("has_person"):
         up = fj.get("upper_wear") or ""
         if "连衣裙" in up:
             return up
         return up or "人物穿搭"
     if ocr_texts:
-        # 商品名可能是 OCR 最长的一行（品牌/产品名）
         return max(ocr_texts, key=len)
     return "未识别"
 
 
-def _infer_brand(fj: dict[str, Any], ocr_texts: list[str]) -> str:
+def _infer_brand_old(fj: dict, ocr_texts: list[str]) -> str:
     brand = fj.get("brand")
     if brand:
         return str(brand)
-    # OCR 里短的、纯字母/汉字短串可能是 brand
     for t in ocr_texts:
         if 1 < len(t) <= 12:
             return t
     return "无法判断"
 
 
-def _infer_category(fj: dict[str, Any]) -> str:
+def _infer_category_old(fj: dict) -> str:
     cat = fj.get("category")
     if cat:
         return str(cat)
@@ -207,27 +319,19 @@ def _infer_category(fj: dict[str, Any]) -> str:
     return "非产品图"
 
 
-def _build_appearance(fj: dict[str, Any]) -> str:
-    """外观描述：颜色+款式+材质+图案 拼成一段。"""
-    parts: list[str] = []
-    for key, _label in [
-        ("upper_color", "主色"),
-        ("upper_wear", "款式"),
-        ("material", "材质"),
-        ("pattern", "图案"),
-    ]:
+def _build_appearance_old(fj: dict) -> str:
+    parts = []
+    for key in ("upper_color", "upper_wear", "material", "pattern"):
         v = fj.get(key)
         if v and v not in ("无法判断", "未知", "纯色"):
             parts.append(str(v))
     if not parts:
-        if fj.get("has_person"):
-            return "人像穿搭整体造型"
-        return "无法判断"
+        return "人像穿搭整体造型" if fj.get("has_person") else "无法判断"
     return "、".join(parts)
 
 
-def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]:
-    feats: list[str] = []
+def _build_key_features_old(fj: dict, ocr_texts: list[str]) -> list[str]:
+    feats = []
     for key in (
         "upper_wear",
         "lower_wear",
@@ -248,9 +352,7 @@ def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]:
             feats.append(v)
     if ocr_texts:
         feats.append(f"画面文字: {'/'.join(ocr_texts[:3])}")
-    # 去重
-    out: list[str] = []
-    seen: set[str] = set()
+    out, seen = [], set()
     for f in feats:
         f = f.strip()
         if f and f not in seen and len(f) <= 30:
@@ -259,27 +361,292 @@ def _build_key_features(fj: dict[str, Any], ocr_texts: list[str]) -> list[str]:
     return out[:6] if out else ["无法判断"]
 
 
-def assemble_result(
-    idx: int,
-    fast_json: dict[str, Any] | None,
-    ocr_texts: list[str],
-) -> dict[str, Any]:
-    """把 fast_json 结果 + OCR 文本组装成下游兼容的 product dict。"""
+def _flatten_colors(c) -> list[str]:
+    """colors可能是字符串数组或[{hex,name,coverage}]，统一返回名字数组"""
+    if not c:
+        return []
+    out = []
+    for item in c:
+        if isinstance(item, dict):
+            n = item.get("name")
+            if n:
+                out.append(n)
+        elif isinstance(item, str):
+            out.append(item)
+    return out
+
+
+def assemble_result(idx: int, fast_json: dict | None, ocr_texts: list[str]) -> dict[str, Any]:
     fj = fast_json or {}
     ocr_texts = ocr_texts or []
 
-    portrait_prompt = _build_portrait_prompt(fj)
-    name = _infer_name(fj, ocr_texts)
-    brand = _infer_brand(fj, ocr_texts)
-    category = _infer_category(fj)
-    appearance = _build_appearance(fj)
-    key_features = _build_key_features(fj, ocr_texts)
+    if _is_v4_schema(fj):
+        return _assemble_v4(idx, fj, ocr_texts)
+    else:
+        return _assemble_old(idx, fj, ocr_texts)
+
+
+def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
+    """v4嵌套schema → 下游product dict"""
+    vtype = fj.get("type") or "other"
+    products = fj.get("products") or []
     scene = fj.get("scene") or "通用"
     mood = fj.get("mood") or ""
-    packaging = "无法判断"  # 包装细节专用API无，保留占位
-    text_on_package = ocr_texts[:8]
-    summary = _build_summary(fj, name, brand, category)
+    colors = fj.get("colors") or []
+    visible_text = fj.get("visible_text") or []
+    color_names = _flatten_colors(colors)
 
+    # 合并OCR文字和visible_text
+    pkg_texts = []
+    for vt in visible_text:
+        if isinstance(vt, dict):
+            t = vt.get("text")
+            if t:
+                pkg_texts.append(str(t))
+        elif isinstance(vt, str):
+            pkg_texts.append(vt)
+    pkg_texts.extend(ocr_texts[:5])
+    # 去重
+    seen_t = set()
+    text_on_package = []
+    for t in pkg_texts:
+        t = str(t).strip()
+        if t and t not in seen_t and len(t) <= 50:
+            seen_t.add(t)
+            text_on_package.append(t)
+    text_on_package = text_on_package[:8]
+
+    has_person = fj.get("has_person", False)
+
+    # 人物类
+    if vtype == "person" or has_person:
+        # 取第一个人物信息（v5 schema人物信息在顶层）
+        person_info = fj
+        # 兼容people嵌套
+        ppl = fj.get("people")
+        if isinstance(ppl, dict) and ppl.get("has_person"):
+            person_info = {**fj, **ppl}
+            has_person = True
+
+        portrait_prompt = _build_portrait_prompt_from_v4(person_info)
+        outfit_style = person_info.get("outfit_style") or ""
+        upper = person_info.get("upper_wear") or ""
+        lower = person_info.get("lower_wear") or ""
+        dress = person_info.get("dress_wear") or ""
+        outer = person_info.get("outerwear") or ""
+        if dress:
+            name = str(dress)[:25]
+        elif outer and upper:
+            name = f"{outer}+{upper}"[:30]
+        elif upper:
+            name = (str(upper) + (f"+{lower}" if lower else ""))[:30]
+        else:
+            name = "人物穿搭"
+        brand = "无法判断"
+        category = "人物穿搭"
+        # appearance: 外套+上衣+下装/裙+鞋+包+发型+妆容
+        app_parts = []
+        for k in ("outerwear", "upper_wear", "lower_wear", "dress_wear", "shoes", "bag", "hairstyle", "makeup"):
+            v = person_info.get(k)
+            if v and v not in ("null", None, "无明显妆容"):
+                app_parts.append(str(v))
+        appearance = "；".join(app_parts) if app_parts else "人像穿搭整体造型"
+        # key_features: 服装+配饰+拍摄信息
+        kf = []
+        for k in (
+            "outfit_style",
+            "upper_wear",
+            "lower_wear",
+            "dress_wear",
+            "outerwear",
+            "shoes",
+            "bag",
+            "hairstyle",
+            "expression",
+            "pose",
+        ):
+            v = person_info.get(k)
+            if v and v not in ("null", None, "无法判断"):
+                kf.append(str(v))
+        acc = person_info.get("accessories") or []
+        if isinstance(acc, list):
+            for a in acc:
+                if a and str(a) not in kf:
+                    kf.append(str(a))
+        elif isinstance(acc, str) and acc:
+            kf.append(acc)
+        for k in ("shot_type", "camera_angle", "lighting", "atmosphere"):
+            v = person_info.get(k)
+            if v and v not in ("null", None):
+                kf.append(str(v))
+        if text_on_package:
+            kf.append(f"文字:{'/'.join(text_on_package[:3])}")
+        kf = kf[:8] or ["无法判断"]
+        summary = (outfit_style + " " if outfit_style and outfit_style not in name else "") + name[:25]
+        if not summary.strip():
+            summary = "人物穿搭"
+        return {
+            "name": name[:30],
+            "brand": brand,
+            "category": category,
+            "appearance": appearance[:400],
+            "packaging": "人物形象无包装",
+            "text_on_package": text_on_package,
+            "key_features": kf,
+            "scene": scene,
+            "mood": mood,
+            "portrait_prompt": portrait_prompt[:300],
+            "summary": summary[:50],
+            "_source": "v2_fast_json_v5",
+        }
+
+    # 商品类
+    if vtype == "product" and products:
+        # 主商品（第一个position=main或第一个）
+        main = products[0]
+        for p in products:
+            if p.get("position") == "main":
+                main = p
+                break
+        name = main.get("product_name") or "未识别"
+        brand = main.get("brand") or "无法判断"
+        category = main.get("category") or "非产品图"
+        # appearance: 包装外观
+        app_parts = []
+        for k in ("package_color", "package_type", "cap_type", "body_shape", "label_design"):
+            v = main.get(k)
+            if v and v not in ("null", None):
+                app_parts.append(str(v))
+        appearance = "；".join(app_parts) if app_parts else "无法判断"
+        # packaging: 包装信息（直接用package_type+package_color）
+        pkg_parts = []
+        if main.get("package_type"):
+            pkg_parts.append(str(main["package_type"]))
+        if main.get("package_color"):
+            pkg_parts.append(str(main["package_color"]))
+        if main.get("cap_type"):
+            pkg_parts.append(f"配{main['cap_type']}")
+        packaging = "，".join(pkg_parts) if pkg_parts else "无法判断"
+        # key_features: product_features字段
+        feats = main.get("product_features") or []
+        if not isinstance(feats, list):
+            feats = [str(feats)]
+        kf = [str(f) for f in feats if f and len(str(f)) <= 40][:6]
+        # 补充卖点
+        sell = main.get("key_selling_points") or []
+        if isinstance(sell, list):
+            for s in sell[:2]:
+                if s and len(str(s)) <= 30 and str(s) not in kf:
+                    kf.append(f"卖点:{s}")
+        if text_on_package:
+            kf.append(f"文字: {'/'.join(text_on_package[:3])}")
+        kf = kf[:6] or ["无法判断"]
+        portrait_prompt = _build_product_prompt_from_v4(main, fj)
+        if brand != "无法判断" and brand not in name:
+            summary = f"{brand} {name}"
+        else:
+            summary = name
+        return {
+            "name": str(name)[:50],
+            "brand": str(brand)[:30],
+            "category": str(category)[:20],
+            "appearance": appearance[:200],
+            "packaging": packaging[:100],
+            "text_on_package": text_on_package,
+            "key_features": kf,
+            "scene": scene,
+            "mood": mood,
+            "portrait_prompt": portrait_prompt[:200],
+            "summary": str(summary)[:60],
+            "_source": "v2_fast_json_v4",
+        }
+
+    # 门店类或其他
+    if vtype == "store":
+        store_type = fj.get("store_type") or "店铺"
+        name = store_type
+        brand = fj.get("brand_signage") or "无法判断"
+        category = "门店场景"
+        visual = fj.get("visual_elements") or []
+        if isinstance(visual, str):
+            visual = [visual]
+        atmosphere = fj.get("atmosphere") or mood
+        appearance_parts = []
+        if fj.get("store_layout"):
+            appearance_parts.append(str(fj["store_layout"]))
+        if visual:
+            appearance_parts.append("、".join(str(v) for v in visual[:3]))
+        if fj.get("cleanliness"):
+            appearance_parts.append(str(fj["cleanliness"]))
+        appearance = "；".join(appearance_parts) if appearance_parts else "门店环境"
+        kf = []
+        if isinstance(visual, list):
+            kf.extend(str(v) for v in visual if v and len(str(v)) <= 30)
+        prods_vis = fj.get("product_categories_visible") or []
+        if isinstance(prods_vis, list):
+            kf.extend(str(c) for c in prods_vis[:3] if c)
+        promo = fj.get("promotion_elements") or []
+        if isinstance(promo, list) and promo:
+            kf.append("促销活动:" + "、".join(str(p) for p in promo[:2]))
+        if text_on_package:
+            kf.append(f"文字: {'/'.join(text_on_package[:3])}")
+        kf = kf[:6] or ["门店场景"]
+        portrait_prompt = f"{brand if brand!='无法判断' else ''}{store_type}，{atmosphere}，{scene}场景，{('、'.join(color_names[:3])+'配色，') if color_names else ''}产品陈列丰富，门店实拍"
+        portrait_prompt = portrait_prompt.strip("，")
+        summary = f"{store_type}场景"
+        return {
+            "name": name[:30],
+            "brand": str(brand)[:30],
+            "category": category,
+            "appearance": appearance[:200],
+            "packaging": "门店场景无包装",
+            "text_on_package": text_on_package,
+            "key_features": kf,
+            "scene": scene,
+            "mood": atmosphere or mood,
+            "portrait_prompt": portrait_prompt[:200],
+            "summary": summary[:40],
+            "_source": "v2_fast_json_v4",
+        }
+
+    # other 兜底
+    desc = fj.get("description") or "未识别"
+    return {
+        "name": desc[:30],
+        "brand": "无法判断",
+        "category": "非产品图",
+        "appearance": desc[:200],
+        "packaging": "无法判断",
+        "text_on_package": text_on_package,
+        "key_features": [desc[:30]] if desc != "未识别" else ["无法判断"],
+        "scene": scene,
+        "mood": mood,
+        "portrait_prompt": f"{scene}，{mood}氛围，{desc}"[:200],
+        "summary": desc[:40],
+        "_source": "v2_fast_json_v4_other",
+    }
+
+
+def _assemble_old(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
+    """旧扁平schema（兼容存量prompt或pro兜底输出）"""
+    portrait_prompt = _build_portrait_prompt_old(fj)
+    name = _infer_name_old(fj, ocr_texts)
+    brand = _infer_brand_old(fj, ocr_texts)
+    category = _infer_category_old(fj)
+    appearance = _build_appearance_old(fj)
+    key_features = _build_key_features_old(fj, ocr_texts)
+    scene = fj.get("scene") or "通用"
+    mood = fj.get("mood") or ""
+    packaging = "无法判断"
+    text_on_package = ocr_texts[:8]
+    if fj.get("has_person"):
+        up = fj.get("upper_wear") or "穿搭"
+        style = fj.get("style") or ""
+        summary = f"{style}{up}" if style and style not in up else up
+    elif brand != "无法判断" and name != brand:
+        summary = f"{brand} {name}"
+    else:
+        summary = name
     return {
         "name": name,
         "brand": brand,
@@ -294,14 +661,3 @@ def assemble_result(
         "summary": summary,
         "_source": "v2_fast_json",
     }
-
-
-def _build_summary(fj: dict, name: str, brand: str, category: str) -> str:
-    if fj.get("has_person"):
-        up = fj.get("upper_wear") or "穿搭"
-        style = fj.get("style") or ""
-        base = f"{style}{up}" if style and style not in up else up
-        return base
-    if brand != "无法判断" and name != brand:
-        return f"{brand} {name}"
-    return name

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

@@ -8,7 +8,7 @@ fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
 - system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
 - timeout=25s
 - API Key 从环境变量 DASHSCOPE_API_KEY 读取
-- 返回 dict 字段与旧 _normalize() 兼容，下游零改动
+- 返回 dict 统一走 assembler.assemble_result 组装，与 fast 路径输出格式完全一致
 """
 
 from __future__ import annotations
@@ -19,7 +19,7 @@ import os
 import time
 from typing import Any
 
-from . import _prompt
+from . import _prompt, assembler
 
 logger = logging.getLogger(__name__)
 
@@ -33,84 +33,6 @@ def _api_key() -> str | None:
     return os.environ.get("DASHSCOPE_API_KEY")
 
 
-def _assemble_pp(obj: dict[str, Any]) -> str:
-    """从 JSON 字段组装 portrait_prompt（60-100字人物穿搭描述，给 Seedream t2i 用）。"""
-    if not obj.get("has_person"):
-        name = obj.get("product_name") or "商品"
-        brand = obj.get("brand") or ""
-        kf = obj.get("key_features") or []
-        scene = obj.get("scene") or ""
-        mood = obj.get("mood") or ""
-        outfit = obj.get("outfit") or ""
-        if outfit:
-            return outfit
-        pieces = []
-        if brand:
-            pieces.append(brand)
-        pieces.append(str(name))
-        if isinstance(kf, list):
-            pieces.extend(str(x) for x in kf[:2] if x)
-        if mood:
-            pieces.append(str(mood) + "氛围")
-        if scene:
-            pieces.append(str(scene) + "场景")
-        pieces.append("产品特写")
-        p = "，".join(x for x in pieces if x)
-        return p if len(p) >= 10 else "产品展示图，特写镜头"
-
-    parts: list[str] = []
-    gender = obj.get("gender") or ""
-    age = obj.get("age_range") or ""
-    subj = ""
-    if age == "儿童":
-        subj = "小女孩" if gender == "女" else ("小男孩" if gender == "男" else "儿童")
-    elif age == "青少年":
-        subj = "少女" if gender == "女" else ("少年" if gender == "男" else "青少年")
-    else:
-        prefix_map = {"青年": "年轻", "中年": "中年", "老年": "老年"}
-        gw = {"男": "男性", "女": "女性"}.get(gender, "")
-        prefix = prefix_map.get(age, "")
-        subj = (prefix + gw) if (prefix or gw) else "人物"
-    parts.append(f"一位{subj}")
-
-    outfit = obj.get("outfit") or ""
-    if outfit:
-        parts.append(f"身着{outfit}")
-
-    hair = obj.get("hair") or ""
-    if hair:
-        parts.append(str(hair))
-
-    pose = obj.get("pose") or ""
-    expr = obj.get("expression") or ""
-    det = []
-    if expr and expr not in ("自然", "平静"):
-        det.append(f"神情{expr}")
-    if pose and pose not in ("站立",):
-        det.append(str(pose))
-    if det:
-        parts.append("，".join(det))
-
-    style_parts = []
-    mood = obj.get("mood") or ""
-    scene = obj.get("scene") or ""
-    if mood:
-        style_parts.append(str(mood))
-    if scene and scene != "通用":
-        style_parts.append(str(scene))
-    if style_parts:
-        parts.append("".join(style_parts) + "风格")
-    else:
-        parts.append("人像写真")
-
-    full = "，".join(p for p in parts if p)
-    if len(full) < 40:
-        full += "，自然光线下人像特写，画面清晰"
-    if len(full) > 120:
-        full = full[:120].rstrip("，") + "。"
-    return full
-
-
 def call_pro_vlm(
     img_url: str,
     idx: int,
@@ -182,9 +104,9 @@ def call_pro_vlm(
             if lines and lines[-1].strip().startswith("```"):
                 lines = lines[:-1]
             s = "\n".join(lines).strip()
-        l, rr = s.find("{"), s.rfind("}")
-        if l >= 0 and rr > l:
-            s = s[l : rr + 1]
+        lpos, rr = s.find("{"), s.rfind("}")
+        if lpos >= 0 and rr > lpos:
+            s = s[lpos : rr + 1]
         try:
             obj = json.loads(s)
         except json.JSONDecodeError:
@@ -193,32 +115,11 @@ def call_pro_vlm(
         if not isinstance(obj, dict):
             return None
 
-        pp = _assemble_pp(obj)
-        kf = obj.get("key_features")
-        if not isinstance(kf, list):
-            kf = [str(kf)] if kf else ["无法判断"]
-        else:
-            kf = [str(x) for x in kf if x] or ["无法判断"]
-
-        name = obj.get("product_name") or "未识别"
-        if obj.get("has_person") and (not name or name == "未识别"):
-            name = obj.get("outfit") or "人物穿搭"
-        brand = obj.get("brand") or "无法判断"
-        category = obj.get("category") or ("服饰" if obj.get("has_person") else "非产品图")
-        return {
-            "name": str(name),
-            "brand": str(brand),
-            "category": str(category),
-            "appearance": str(obj.get("outfit") or "无法判断"),
-            "packaging": "无法判断",
-            "text_on_package": [],
-            "key_features": kf[:6],
-            "scene": str(obj.get("scene") or "通用"),
-            "mood": str(obj.get("mood") or ""),
-            "portrait_prompt": pp,
-            "summary": str(name),
-            "_source": "vlm_pro",
-        }
+        # 通过assembler统一组装，兼容v4嵌套schema和旧扁平schema
+        result = assembler.assemble_result(idx, obj, [])
+        result["_source"] = "vlm_pro"
+        result["_fallback_used"] = True
+        return result
     except Exception as e:
         elapsed = time.time() - t0
         logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)

packages/shared/ai_client.py

@@ -291,6 +291,7 @@ class DoubaoClient:
                     data.get("usage", {}).get("completion_tokens", 0),
                     _elapsed,
                     attempt + 1,
+                    _req_timeout,
                 )
                 return content.strip()
             except Exception as e:

tests/unit/test_vision_v4_prompt.py

@@ -0,0 +1,291 @@
+# -*- coding: utf-8 -*-
+"""vision v4 prompt / assembler 单元测试：
+
+- assembler 正确识别 v4 嵌套 schema 与旧扁平 schema
+- v4 product/person/store/other 四类输出组装出下游必出字段
+- 旧扁平 schema 行为不变
+- _prompt._resolve：DB 有 active prompt 时原样使用（不追加硬编码 schema）；
+  DB 无记录时回落到硬编码 JSON schema
+"""
+
+from __future__ import annotations
+
+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",
+}
+
+
+# ---------- schema 识别 ----------
+
+
+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:
+    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,
+    }
+    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"]
+
+
+# ---------- 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:
+    fj = {
+        "has_person": True,
+        "gender": "男",
+        "age_range": "中年",
+        "upper_wear": "西装",
+        "upper_color": "深灰色",
+        "lower_wear": "西裤",
+        "lower_color": "黑色",
+        "accessories": ["手表"],
+        "hairstyle": "短发",
+        "expression": "严肃",
+        "scene": "办公室",
+        "style": "商务",
+        "mood": "专业",
+    }
+    r = assembler.assemble_result(0, fj, [])
+    assert REQUIRED_KEYS <= set(r.keys())
+    assert "中年男性" in r["portrait_prompt"]
+    assert r["_source"] == "v2_fast_json"
+
+
+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_none_input() -> None:
+    r = assembler.assemble_result(0, None, [])
+    assert REQUIRED_KEYS <= set(r.keys())
+
+
+# ---------- _prompt 解析 ----------
+
+
+@pytest.fixture(autouse=True)
+def _clear_prompt_cache() -> Any:
+    _prompt.invalidate_cache()
+    yield
+    _prompt.invalidate_cache()
+
+
+def _fake_tpl(system_prompt: str = "v4 system prompt 只返回JSON") -> Any:
+    return types.SimpleNamespace(
+        system_prompt=system_prompt,
+        user_prompt_template="分析 {image_count} 张图",
+        version=4,
+    )
+
+
+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
+
+
+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"))
+    sys_prompt, _ = _prompt.resolve_pro_prompt()
+    assert sys_prompt == "DB_V4_PRO_PROMPT"
+    assert "【输出格式要求】" not in sys_prompt
+
+
+def test_resolve_falls_back_when_no_db(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
+
+
+def test_resolve_caches(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = {"n": 0}
+
+    def _load() -> Any:
+        calls["n"] += 1
+        return _fake_tpl("CACHED_PROMPT")
+
+    monkeypatch.setattr(_prompt, "_load_db_template", _load)
+    s1, _ = _prompt.resolve_fast_prompt()
+    s2, _ = _prompt.resolve_fast_prompt()
+    assert s1 == s2 == "CACHED_PROMPT"
+    assert calls["n"] == 1