From cd618c3f2958f7c7332706adf753cf07c1e9ed75 Mon Sep 17 00:00:00 2001 From: Xiaoxia Agent Date: Tue, 6 Oct 2026 00:53:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(vision):=20image=5Fanalysis=20prompt=20XML/?= =?UTF-8?q?JSON=E5=86=B2=E7=AA=81=E4=BF=AE=E5=A4=8D+v4=20JSON=20prompt=20s?= =?UTF-8?q?eed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _prompt: DB有active image_analysis prompt时原样使用，不再追加硬编码schema， 让DB prompt完全控制输出格式（修复DB写XML、调用强制json_object导致解析失败、 商品图全部走fallback的问题） - assembler: 新增v4嵌套schema路由（product/store/person/other四类）， 保留旧扁平schema兼容，下游零改动 - vlm_fallback: pro兜底结果统一走assembler.assemble_result，与fast路径输出一致 - migration 097: 固化staging已验证的v4 JSON prompt并置为active（幂等） --- .../097_viral_video_image_analysis_v4.py | 81 +++ .../worker/worker_app/tasks/vision/_prompt.py | 11 +- .../worker_app/tasks/vision/assembler.py | 539 ++++++++++++++---- .../worker_app/tasks/vision/vlm_fallback.py | 119 +--- tests/unit/test_vision_v4_prompt.py | 292 ++++++++++ 5 files changed, 828 insertions(+), 214 deletions(-) create mode 100644 alembic/versions/097_viral_video_image_analysis_v4.py create mode 100755 tests/unit/test_vision_v4_prompt.py diff --git a/alembic/versions/097_viral_video_image_analysis_v4.py b/alembic/versions/097_viral_video_image_analysis_v4.py new file mode 100644 index 000000000..7f7750b1a --- /dev/null +++ b/alembic/versions/097_viral_video_image_analysis_v4.py @@ -0,0 +1,81 @@ +# -*- coding: utf-8 -*- +"""viral video image_analysis v4 JSON prompt seed + +背景：image_analysis DB prompt 历史上是 XML 输出格式（...）， +但 V2 视觉模块 VLM 调用强制 response_format=json_object，XML 被 JSON 解析失败， +导致所有商品图走 fallback 兜底。本迁移固化已在 staging 验证通过的 v4 JSON prompt +（type/products/people/store_info 多类型嵌套 schema），并将其置为 active； +同时修复解析侧（DB prompt 完全控制输出格式，不再追加硬编码 schema）。 + +幂等：v4 记录已存在则更新为已验证内容；不存在则插入。其他 image_analysis 版本 +统一置为 inactive。 +""" + +from alembic import op + +revision = "097_viral_video_image_analysis_v4" +down_revision = "096_feature_billing_fields" +branch_labels = None +depends_on = None + + +V4_NAME = "图片分析JSON v4（多类型详细描述+代码不追加schema）" +V4_VERSION = 4 +V4_SYSTEM_PROMPT = "你是一位专业的视觉内容分析师。先判断图片主体类型，再按对应维度详细分析，严格输出一个合法的JSON对象。\n\n【第一步：判断主体类型 type】\n- \"product\"：单个或多个商品/产品包装/商品陈列\n- \"store\"：店铺内景、门店门头、货架陈列、经营环境、活动现场\n- \"person\"：人物形象、穿搭造型、肖像照片\n- \"other\"：风景/动物/美食/截图/抽象画面等\n\n【第二步：通用字段（所有类型必填）】\n{\n \"type\": \"product/store/person/other\",\n \"scene\": \"具体场景描述，如'白色背景产品图''超市货架''户外街拍''便利店内部''家居环境'\",\n \"mood\": \"整体氛围，如'清新专业''高级干净''热闹促销''温暖明亮''生活化'\",\n \"style\": \"视觉风格，如'商业产品摄影''街拍''门店实拍''人像写真''证件照''活动现场照'\",\n \"colors\": [\n {\"hex\":\"#E60012\",\"name\":\"亮红色\",\"coverage\":0.6},\n {\"hex\":\"#FFFFFF\",\"name\":\"白色\",\"coverage\":0.3}\n ],\n \"visible_text\": [\n {\"text\":\"识别到的文字内容\",\"position\":\"位置描述，如'瓶身正面'\"}\n ],\n \"lighting\": \"光线描述，如'明亮柔光''自然光''室内灯光''逆光'\",\n \"composition\": \"构图描述，如'居中特写''中景''平视角度''俯拍'\"\n}\n\n【商品类 type=product 额外字段】\n在通用字段基础上增加：\n{\n \"subject_type\": \"product\",\n \"products\": [\n {\n \"product_name\": \"商品名称，尽量具体，如'OMO奥妙除菌除螨洗衣液'\",\n \"brand\": \"品牌名，如'OMO奥妙''CHANTE CLAIR大公鸡头'，无法识别填null\",\n \"category\": \"类目：服饰/鞋包/美妆/数码/食品/家居清洁/洗护/家居/配饰/母婴/其他\",\n \"product_count\": 1,\n \"package_type\": \"包装类型，如'瓶装''袋装''盒装''罐装''喷雾瓶''多瓶装'\",\n \"package_color\": \"包装主色，具体如'亮红色瓶身''透明瓶身装白色液体''蓝色金属罐'\",\n \"cap_type\": \"瓶盖/瓶口类型，如'透明翻盖式按压瓶口''红白按压喷雾头''黑色旋盖'\",\n \"body_shape\": \"瓶身/包装形状特征，如'带侧面握持把手的竖款瓶身''圆润矮胖造型'\",\n \"label_design\": \"标签设计描述，如'瓶身印十字盾牌图案''贴有公鸡图案标签''大面积品牌logo'\",\n \"key_text_on_package\": [\"瓶身可见的主要文字，如'99除菌''除菌除螨''马赛香''强效去污'\"],\n \"background_elements\": [\"背景元素，如'绿叶''金色时钟造型''气泡''艾草类草本植物''红粉拼接背景'\"],\n \"product_features\": [\n \"包装特征短语3-6个，具体细致，颜色要准确（写'亮红色'不写'红色'）\",\n \"包含包装颜色、瓶盖类型、握持结构、标签图案、标志性图形等\"\n ],\n \"key_selling_points\": [\"核心卖点1-3个，如'天然除菌除螨''万能去污一喷净''多瓶家庭装'\"]\n }\n ],\n \"has_person\": false\n}\n多件商品逐个填入products数组，每个元素独立描述。\n\n【门店类 type=store 额外字段】\n在通用字段基础上增加：\n{\n \"subject_type\": \"store\",\n \"store_type\": \"店铺类型，如'社区便利店''连锁超市''品牌专卖店''母婴店''餐饮店'\",\n \"store_layout\": \"空间布局，如'纵深货架布局''L型收银台+两侧货架''开放式陈列'\",\n \"brand_signage\": \"品牌招牌/门头文字，如'全家FamilyMart''XX便利店'\",\n \"visual_elements\": [\n \"视觉元素，如'主色调红白''顶部照明灯管''手写价签''促销海报''地堆陈列''收银台商品'\"\n ],\n \"product_categories_visible\": [\"可见的商品类目，如'饮料''零食''日用品''生鲜'\"],\n \"promotion_elements\": [\"促销活动元素，如'第二件半价海报''红色折扣价签''满减吊旗'，无则空数组\"],\n \"customer_flow\": \"客流描述，如'整洁空旷''零星顾客''排队结账'，无顾客填'店内无顾客'\",\n \"cleanliness\": \"整洁度，如'干净整洁''略显杂乱''货架整齐'\",\n \"atmosphere\": \"经营氛围，如'亲民生活化''高端精致''热闹促销''日常便民'\",\n \"has_person\": false\n}\n\n【人物类 type=person 额外字段】\n在通用字段基础上增加：\n{\n \"subject_type\": \"person\",\n \"has_person\": true,\n \"person_count\": 1,\n \"gender\": \"男/女/无法判断\",\n \"age_range\": \"儿童/青少年/青年/中年/老年/无法判断\",\n \"upper_wear\": \"上装款式，如'白色V领短袖T恤''格纹衬衫''黑色卫衣'\",\n \"upper_color\": \"上装主色\",\n \"lower_wear\": \"下装款式，如'黑色高腰阔腿裤''牛仔短裤''百褶半身裙'，穿连衣裙填null\",\n \"lower_color\": \"下装主色\",\n \"dress_color\": \"连衣裙颜色，穿连衣裙时填，否则null\",\n \"accessories\": [\"配饰，如'银色项链''黑色棒球帽''金属框眼镜''白色运动鞋'\"],\n \"hairstyle\": \"发型发色，如'黑色长直发''棕色波浪卷发''高马尾''短发寸头'\",\n \"expression\": \"表情，如'微笑''严肃''酷感''自信''自然放松'\",\n \"pose\": \"姿势，如'侧身站立看向镜头''双手插兜''坐姿''行走中'\",\n \"outfit_style\": \"穿搭风格，如'休闲日常''商务正装''运动风''复古潮流''甜美淑女''酷飒街头'\",\n \"portrait_prompt\": \"60-100字人物肖像描述，用于AI生图，如'一位25岁左右年轻女性，身穿白色V领短袖T恤、黑色高腰阔腿裤，搭配银色项链，长发披肩，表情自信，街拍风格，阳光明媚的城市街头'\",\n \"products\": []\n}\n\n【其他类 type=other】\n{\n \"subject_type\": \"other\",\n \"description\": \"描述这是什么内容\",\n \"has_person\": false,\n \"products\": []\n}\n\n【输出要求】\n1. 严格输出一个合法JSON对象，不要任何解释文字、不要markdown代码块、不要XML标签\n2. 商品描述要具体细致：颜色写具体色调（亮红色/米白色/深蓝色），不写笼统词汇\n3. 瓶身/包装上的文字尽量识别（品牌名、功能词、卖点词）\n4. 多个商品逐个描述，不要合并\n5. 无法判断的字段填null或空数组，布尔值填true/false\n6. 只输出JSON，不要JSON以外的任何内容" +V4_USER_PROMPT_TEMPLATE = "分析以下图片：\n图片数量：{image_count}张\n行业：{industry}\n{image_urls}\n\n请按system prompt约定的JSON schema输出分析结果，只返回JSON对象。" +V4_EXAMPLE_OUTPUT = None + + +def upgrade() -> None: + conn = op.get_bind() + + existing = conn.execute( + "SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = :v", + {"v": V4_VERSION}, + ).fetchone() + + if existing is None: + conn.execute( + "INSERT INTO viral_video_prompt_templates " + "(name, prompt_type, version, system_prompt, user_prompt_template, " + "example_output, is_active) " + "VALUES (:name, 'image_analysis', :version, :sys, :user, :example, TRUE)", + { + "name": V4_NAME, + "version": V4_VERSION, + "sys": V4_SYSTEM_PROMPT, + "user": V4_USER_PROMPT_TEMPLATE, + "example": V4_EXAMPLE_OUTPUT, + }, + ) + else: + conn.execute( + "UPDATE viral_video_prompt_templates SET name = :name, " + "system_prompt = :sys, user_prompt_template = :user, " + "example_output = :example, is_active = TRUE WHERE id = :id", + { + "name": V4_NAME, + "sys": V4_SYSTEM_PROMPT, + "user": V4_USER_PROMPT_TEMPLATE, + "example": V4_EXAMPLE_OUTPUT, + "id": existing[0], + }, + ) + + conn.execute( + "UPDATE viral_video_prompt_templates SET is_active = FALSE " + "WHERE prompt_type = 'image_analysis' AND version <> :v", + {"v": V4_VERSION}, + ) + + +def downgrade() -> None: + conn = op.get_bind() + conn.execute( + "DELETE FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = :v", + {"v": V4_VERSION}, + ) + conn.execute( + "UPDATE viral_video_prompt_templates SET is_active = TRUE " + "WHERE prompt_type = 'image_analysis' AND version = 1" + ) diff --git a/apps/worker/worker_app/tasks/vision/_prompt.py b/apps/worker/worker_app/tasks/vision/_prompt.py index 0808d603f..ed80bbf76 100644 --- a/apps/worker/worker_app/tasks/vision/_prompt.py +++ b/apps/worker/worker_app/tasks/vision/_prompt.py @@ -3,8 +3,9 @@ 且 is_active=true），30s TTL 热加载；DB 无有效记录/异常时，fallback 到纯硬编码 JSON schema prompt。 规则（简单直接，不做字符串匹配判断）： -- DB 有 is_active=true 的 image_analysis 记录（含种子默认XML和用户修改后的版本）： - * system = DB.system_prompt + JSON_SCHEMA_APPEND（追加完整JSON字段schema，覆盖XML等其他输出格式要求） +- 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 """ @@ -73,7 +74,8 @@ _PRO_JSON_SCHEMA = ( ) DEFAULT_PRO_USER = "分析这张图片，返回符合schema的JSON。" -# DB 配置存在时，追加在用户 system_prompt 末尾的JSON schema约束 +# 保留旧 JSON schema 追加文本作为常量（DB prompt 完全控制输出格式后不再使用， +# 保留以便排查历史行为）。 _FAST_JSON_APPEND = ( "\n\n【输出格式要求】无论上文如何要求，最终你必须只返回一个合法的JSON对象，" "严格包含以下字段（字段值不确定时填null或空数组）：\n" @@ -173,7 +175,6 @@ def _resolve(kind: str) -> tuple[str, str]: default_sys = _FAST_JSON_SCHEMA if kind == "fast" else _PRO_JSON_SCHEMA default_user = DEFAULT_FAST_USER if kind == "fast" else DEFAULT_PRO_USER - append = _FAST_JSON_APPEND if kind == "fast" else _PRO_JSON_APPEND sys_prompt = default_sys usr_prompt = default_user @@ -182,7 +183,7 @@ def _resolve(kind: str) -> tuple[str, str]: if tpl is not None: db_sys = (getattr(tpl, "system_prompt", "") or "").strip() if db_sys: - sys_prompt = db_sys + append + 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)", diff --git a/apps/worker/worker_app/tasks/vision/assembler.py b/apps/worker/worker_app/tasks/vision/assembler.py index b45a4d60e..1b91b04fa 100644 --- a/apps/worker/worker_app/tasks/vision/assembler.py +++ b/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,275 @@ 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: + # 取第一个人物信息（v4 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) + name = person_info.get("upper_wear") or "人物穿搭" + if "连衣裙" in name: + pass + else: + lower = person_info.get("lower_wear") or "" + if lower: + name = f"{name}+{lower}" + brand = "无法判断" + category = "服饰" + outfit_parts = [] + for k in ("upper_wear", "lower_wear", "dress_color", "upper_color", "lower_color", "outfit_style"): + v = person_info.get(k) + if v and v not in ("null", None): + outfit_parts.append(str(v)) + appearance = "、".join(outfit_parts) if outfit_parts else "人像穿搭整体造型" + # key_features: 穿搭特征+配饰 + kf = [] + for k in ( + "upper_wear", + "lower_wear", + "upper_color", + "lower_color", + "hairstyle", + "expression", + "pose", + "outfit_style", + ): + 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): + kf.extend(str(a) for a in acc if a) + if text_on_package: + kf.append(f"画面文字: {'/'.join(text_on_package[:3])}") + kf = kf[:6] or ["无法判断"] + summary = (person_info.get("outfit_style") or "") + (person_info.get("upper_wear") or "穿搭") + if not summary or summary == "穿搭": + summary = "人物穿搭" + return { + "name": name[:30], + "brand": brand, + "category": category, + "appearance": appearance, + "packaging": "人物形象无包装", + "text_on_package": text_on_package, + "key_features": kf, + "scene": scene, + "mood": mood, + "portrait_prompt": portrait_prompt, + "summary": summary[:40], + "_source": "v2_fast_json_v4", + } + + # 商品类 + 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 +644,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 diff --git a/apps/worker/worker_app/tasks/vision/vlm_fallback.py b/apps/worker/worker_app/tasks/vision/vlm_fallback.py index fa38a2115..0967a2842 100644 --- a/apps/worker/worker_app/tasks/vision/vlm_fallback.py +++ b/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) diff --git a/tests/unit/test_vision_v4_prompt.py b/tests/unit/test_vision_v4_prompt.py new file mode 100755 index 000000000..ea2225c3e --- /dev/null +++ b/tests/unit/test_vision_v4_prompt.py @@ -0,0 +1,292 @@ +# -*- 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 "T恤" in r["name"] + assert "阔腿裤" in r["name"] + assert "年轻女性" in r["portrait_prompt"] + assert "项链" in r["portrait_prompt"] + assert isinstance(r["key_features"], list) and len(r["key_features"]) <= 6 + + +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