Files
xiaoxia-saas/apps/worker/worker_app/tasks/vision/assembler.py
T
Xiaoxia Agent b8c6091a11 feat(vision): 人物图片分析质量提升 v5
- assembler 新增 v5 人物 schema 支持:扩充 shoes/bag/outerwear/makeup/shot_type/lighting/atmosphere 等字段消费,key_features 上限从6扩到8,appearance 更丰富
- 保留旧扁平 schema 兼容(_assemble_old)
- migration 098:固化 v5 JSON prompt(4583字符,27个人物字段)为 seed,幂等写入
- 新增 test_vision_v4_prompt.py 17个测试覆盖 v4/v5 schema 路由
2026-10-06 10:43:35 +08:00

664 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""把 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,
key_features, scene, mood, portrait_prompt, summary, _source
"""
from __future__ import annotations
from typing import Any
def _join_parts(*parts: str | None) -> str:
return "".join(p for p in parts if p)
_AGE_PREFIX = {"青年": "年轻", "中年": "中年", "老年": "老年"}
_GENDER_WORD = {"男": "男性", "女": "女性"}
def _person_subject(gender: str, age: str) -> str:
gw = _GENDER_WORD.get(gender, "")
if age == "儿童":
if gender == "女":
return "小女孩"
if gender == "男":
return "小男孩"
return "儿童"
if age == "青少年":
if gender == "女":
return "少女"
if gender == "男":
return "少年"
return "青少年"
prefix = _AGE_PREFIX.get(age, "")
if gw:
return f"{prefix}{gw}" if prefix else gw
return f"{prefix}人物" if prefix else "人物"
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 ""
lower_color = fj.get("lower_color") or ""
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
wear = f"{c}{upper}" if c else upper
if material and material not in wear:
wear = f"{material}{wear}"
if pattern and pattern not in wear and pattern != "纯色":
wear += f",{pattern}图案"
return f"身穿{wear}"
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 lower:
lo = f"{lower_color}{lower}" if lower_color else lower
parts.append(f"下身{lo}")
return ",".join(p for p in parts if p)
def _build_portrait_prompt_old(fj: dict) -> str:
if not fj.get("has_person"):
name = fj.get("product_name") or "商品"
brand = fj.get("brand") or ""
colors = fj.get("colors") or []
style = fj.get("style") or ""
scene = fj.get("scene") or ""
mood = fj.get("mood") or ""
pieces = []
if brand:
pieces.append(brand)
pieces.append(name)
if colors:
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:
pieces.append(mood + "氛围")
if scene and scene not in ("通用",):
pieces.append(scene + "场景")
pieces.append("产品特写")
prompt = ",".join(p for p in pieces if p)
return prompt if len(prompt) >= 10 else "产品展示图,特写镜头"
subject = _person_subject_old(fj)
wear = _build_wear_sentence_old(fj)
accessories = fj.get("accessories") or []
if isinstance(accessories, str):
accessories = [accessories]
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 = []
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 = []
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 detail_parts:
pieces.append(",".join(detail_parts))
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_old(fj: dict, ocr_texts: list[str]) -> str:
pname = fj.get("product_name")
if pname and pname != "未识别":
return str(pname)
if fj.get("has_person"):
up = fj.get("upper_wear") or ""
if "连衣裙" in up:
return up
return up or "人物穿搭"
if ocr_texts:
return max(ocr_texts, key=len)
return "未识别"
def _infer_brand_old(fj: dict, ocr_texts: list[str]) -> str:
brand = fj.get("brand")
if brand:
return str(brand)
for t in ocr_texts:
if 1 < len(t) <= 12:
return t
return "无法判断"
def _infer_category_old(fj: dict) -> str:
cat = fj.get("category")
if cat:
return str(cat)
if fj.get("has_person"):
return "服饰"
return "非产品图"
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:
return "人像穿搭整体造型" if fj.get("has_person") else "无法判断"
return "、".join(parts)
def _build_key_features_old(fj: dict, ocr_texts: list[str]) -> list[str]:
feats = []
for key in (
"upper_wear",
"lower_wear",
"upper_color",
"lower_color",
"dress_color",
"material",
"pattern",
"style",
"accessories",
):
v = fj.get(key)
if not v:
continue
if isinstance(v, list):
feats.extend(str(x) for x in v if x)
elif isinstance(v, str) and v not in ("无法判断", "未知", "纯色"):
feats.append(v)
if ocr_texts:
feats.append(f"画面文字: {'/'.join(ocr_texts[:3])}")
out, seen = [], set()
for f in feats:
f = f.strip()
if f and f not in seen and len(f) <= 30:
seen.add(f)
out.append(f)
return out[:6] if out else ["无法判断"]
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 []
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 ""
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,
"category": category,
"appearance": appearance,
"packaging": packaging,
"text_on_package": text_on_package,
"key_features": key_features,
"scene": scene,
"mood": mood,
"portrait_prompt": portrait_prompt,
"summary": summary,
"_source": "v2_fast_json",
}