xiaoxia-saas/apps/worker/worker_app/tasks/vision/assembler.py

# -*- 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:
        # 取第一个人物信息（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,
        "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",
    }