games-creater/pipeline.py

"""可被网站/命令行复用的生成管线。
读 manifest -> 生成各类资产 -> 写 library.json（供网站可视化列出与预览）。
"""

import json
import os

import providers
import spine_builder
import particle_builder
import tween_builder
import baidu_segment
import asset_quality

HERE = os.path.dirname(os.path.abspath(__file__))


def run(manifest, out_root, creds=None, log=print, merge_existing=False):
    """manifest: dict; out_root: 输出根目录; creds: {provider,api_key,base_url,model,size}
    返回 (library_dict, base_out)。"""
    creds = creds or {}
    game = manifest.get("game", "game")
    base_out = os.path.join(out_root, game)
    chars_out = os.path.join(base_out, "characters")
    vfx_out = os.path.join(base_out, "vfx")
    ui_out = os.path.join(base_out, "ui")

    style = manifest.get("style", "")
    library = {
        "game": game,
        "slot_config": manifest.get("slot_config", {}),
        "characters": [],
        "vfx": [],
        "ui": [],
        "ui_art": [],
    }
    if merge_existing:
        library_path = os.path.join(base_out, "library.json")
        if os.path.isfile(library_path):
            with open(library_path, encoding="utf-8") as f:
                existing = json.load(f)
            for key in ("characters", "vfx", "ui", "ui_art"):
                library[key] = existing.get(key, [])
            library["slot_config"] = manifest.get("slot_config") or existing.get("slot_config", {})

    def upsert(section, item):
        item_id = item.get("id")
        if not item_id:
            library[section].append(item)
            return
        rows = library.setdefault(section, [])
        for idx, old in enumerate(rows):
            if old.get("id") == item_id:
                rows[idx] = item
                return
        rows.append(item)
    total_steps = len(manifest.get("characters", [])) + len(manifest.get("ui_art", [])) + len(manifest.get("vfx", [])) + (1 if manifest.get("ui", []) else 0)
    done_steps = 0

    def progress(label):
        nonlocal done_steps
        done_steps += 1
        log(f"进度 {done_steps}/{max(1, total_steps)} · {label}")

    def transparent_prompt(extra):
        return ", ".join([
            extra,
            "生成纯透明背景 PNG，真实 Alpha 通道，不要棋盘格，不要白底，不要阴影。",
        ])

    def alpha_report(img):
        img = img.convert("RGBA")
        alpha = img.getchannel("A")
        mn, mx = alpha.getextrema()
        transparent = sum(1 for v in alpha.getdata() if v == 0)
        ratio = transparent / max(1, img.width * img.height)
        return mn, mx, ratio

    def has_alpha(img):
        return alpha_report(img)[0] == 0

    def log_alpha(label, img, required):
        if not required:
            return
        mn, mx, ratio = alpha_report(img)
        if mn == 0:
            log(f"✅ [{label}] Alpha 透明通道有效：透明像素 {ratio:.1%}")
        else:
            log(f"⚠️ [{label}] 模型返回 PNG 但没有透明 Alpha：alpha={mn}-{mx}，请重新生成或换支持透明输出的图像模型")

    def generate_checked(label, prompt, size, require_alpha):
        img = providers.generate(creds["provider"], prompt, creds["api_key"],
                                 creds.get("base_url", "https://api.openai.com/v1"),
                                 creds.get("model", "gpt-image-2"),
                                 size)
        log_alpha(label, img, require_alpha)
        if not require_alpha or has_alpha(img):
            return img
        log(f"🧠 [{label}] 模型没有真实 Alpha，直接改用百度智能抠图兜底…")
        try:
            fixed = baidu_segment.remove_background(img, label=label, log=log)
        except Exception as e:
            raise RuntimeError(f"模型没有返回真实 Alpha，百度智能抠图也失败：{e}")
        log_alpha(label, fixed, True)
        if has_alpha(fixed):
            return fixed
        raise RuntimeError("百度智能抠图返回结果仍没有真实 Alpha 透明通道")

    def generate_boss_preview_checked(cid, base_prompt, size):
        correction = ""
        last_reason = ""
        for attempt in range(1, 4):
            prompt = base_prompt if not correction else ", ".join([base_prompt, correction])
            if attempt > 1:
                log(f"🔁 [{cid}/preview] 关主预览不合格，重新生成完整角色（第 {attempt}/3 次）…")
            img = generate_checked(f"{cid}/preview", prompt, size, True)
            ok, reason, report = asset_quality.boss_preview_quality(img)
            if ok:
                log(f"✅ [{cid}/preview] 关主完整性检查通过：最大主体 {report['largestShare']:.0%}")
                return img
            last_reason = reason
            log(f"⚠️ [{cid}/preview] 关主完整性检查失败：{reason}")
            correction = (
                "这不是完整关主角色。请重新生成一个单体完整角色：所有身体、头、手、脚、武器必须连接在同一个主体轮廓内，"
                "不要拆件图、不要 sprite sheet、不要分离的靴子/武器/金币袋/碎片，只有一个居中站立的完整关主。"
            )
        raise RuntimeError(f"关主预览不合格：{last_reason}")

    def boss_config():
        boss = manifest.get("slot_config", {}).get("boss", {})
        return boss if boss.get("enabled", False) else {}

    def required_boss_id():
        boss = boss_config()
        if not boss:
            return ""
        return boss.get("id") or "boss_demon_lord"

    def is_required_boss(c):
        boss_id = required_boss_id()
        return bool(boss_id and (c.get("role") == "boss" or c.get("id") == boss_id))

    boss_required_in_this_run = any(is_required_boss(c) for c in manifest.get("characters", []))

    required_failures = []

    def part_sheet_prompt(c, parts, cols, rows):
        ordered = ", ".join(f"{idx + 1}. {p['id']} ({p.get('prompt', '')})" for idx, p in enumerate(parts))
        return ", ".join(x for x in [
            c.get("prompt", ""),
            style,
            (
                f"create one transparent PNG boss rigging parts sprite sheet, exactly {cols} columns and {rows} rows, "
                "one isolated part per cell, no labels, no numbers, no grid lines, no text, no shadows, no background, "
                "parts must not overlap cell borders, all parts from the same character, same lighting and style, "
                "center each part inside its own cell, leave unused cells fully transparent"
            ),
            "cell order left to right, top to bottom: " + ordered,
            transparent_prompt("sprite sheet of separated rigging parts only")
        ] if x)

    def split_part_sheet(sheet_img, parts, cols, rows):
        sheet = sheet_img.convert("RGBA")
        cell_w = max(1, sheet.width // cols)
        cell_h = max(1, sheet.height // rows)
        part_images = {}
        major_ids = {
            "torso", "head", "left_arm", "right_arm", "greatsword",
            "left_leg", "right_leg", "cape",
        }
        for idx, part in enumerate(parts):
            col = idx % cols
            row = idx // cols
            box = (col * cell_w, row * cell_h, (col + 1) * cell_w, (row + 1) * cell_h)
            cell = sheet.crop(box)
            bbox = cell.getchannel("A").point(lambda v: 255 if v > 16 else 0).getbbox()
            if not bbox:
                raise RuntimeError(f"拆件表第 {idx + 1} 格 {part['id']} 是空的")
            bw, bh = bbox[2] - bbox[0], bbox[3] - bbox[1]
            if part["id"] in major_ids and (bw < cell_w * 0.16 or bh < cell_h * 0.16):
                raise RuntimeError(
                    f"拆件表第 {idx + 1} 格 {part['id']} 太小：{bw}x{bh}，需要填满单元格"
                )
            part_images[part["id"]] = cell
        return part_images

    # ---- A. 角色（Spine）----
    for i, c in enumerate(manifest.get("characters", [])):
        cid = c.get("id", f"char_{i}")
        anims = c.get("animations", ["idle"])
        if not creds.get("api_key"):
            log(f"⚠️  未填 key，跳过角色 {cid}")
            continue
        try:
            part_files = []
            if c.get("type") == "spine_parts":
                parts = c.get("parts") or []
                part_images = {}
                preview_img = None
                try:
                    preview_prompt = ", ".join(x for x in [
                        c.get("prompt", ""),
                        style,
                        transparent_prompt(
                            "single complete assembled boss character preview, full body, centered, readable silhouette, no separated parts, no sprite sheet, no atlas"
                        )
                    ] if x)
                    log(f"🖼  [{cid}] 生成完整 Boss 预览图…")
                    preview_img = generate_boss_preview_checked(
                        cid,
                        preview_prompt,
                        c.get("size", creds.get("size", "1024x1024")),
                    )
                except Exception as e:
                    raise RuntimeError(f"完整 Boss 预览图生成失败：{e}")
                sheet_cfg = c.get("spriteSheet") or {}
                use_sheet = c.get("partGeneration") == "sprite_sheet" or sheet_cfg.get("enabled")
                if use_sheet:
                    cols = int(sheet_cfg.get("cols") or 4)
                    rows = int(sheet_cfg.get("rows") or 4)
                    if len(parts) > cols * rows:
                        raise RuntimeError(f"Boss 拆件数 {len(parts)} 超过 sprite sheet 容量 {cols}x{rows}")
                    try:
                        log(f"🎨 [{cid}] 生成 Boss 拆件表 {cols}×{rows}…")
                        sheet = generate_checked(f"{cid}/parts_sheet",
                                                 part_sheet_prompt(c, parts, cols, rows),
                                                 sheet_cfg.get("size", c.get("size", "1024x1024")),
                                                 True)
                        part_images = split_part_sheet(sheet, parts, cols, rows)
                        log(f"✂️ [{cid}] 已按固定网格切出 {len(parts)} 个 Boss 拆件")
                    except Exception as e:
                        log(f"⚠️ [{cid}] 拆件表生成/切图失败，回退逐部件生成：{e}")
                        part_images = {}
                if not part_images:
                    for part in parts:
                        part_id = part["id"]
                        part_prompt = ", ".join(x for x in [
                            part.get("prompt", ""),
                            style,
                            transparent_prompt(
                                f"only the {part_id} rigging part from the boss character, isolated single part, "
                                "not a full character, no complete body, no other body parts, centered"
                            )
                        ] if x)
                        log(f"🎨 [{cid}/{part_id}] 生成 Boss 拆件…")
                        pimg = generate_checked(f"{cid}/{part_id}", part_prompt,
                                                part.get("size", c.get("size", creds.get("size", "1024x1024"))),
                                                True)
                        part_images[part_id] = pimg
                spine_builder.build_parts_character(cid, part_images, chars_out, anims, parts,
                                                    write_preview=preview_img is None)
                part_files = spine_builder.write_part_pngs_from_files(cid, chars_out)
                bad_parts = []
                for part in part_files:
                    ok, reason, detail = asset_quality.boss_part_quality(
                        part["id"],
                        os.path.join(base_out, part["file"]),
                    )
                    if not ok:
                        bad_parts.append(
                            f"{part['id']}：{reason} 最大主体 {detail.get('largestShare', 0):.0%}，第二主体 {detail.get('secondShare', 0):.0%}"
                        )
                if bad_parts:
                    raise RuntimeError("Boss 拆件质量不合格：" + "；".join(bad_parts[:4]))
                if preview_img is not None:
                    os.makedirs(chars_out, exist_ok=True)
                    spine_builder.trim_to_content(preview_img, pad=16).save(
                        os.path.join(chars_out, f"{cid}_preview.png")
                    )
                w, h = 1000, 1000
                files = [f"characters/{cid}.json", f"characters/{cid}.atlas", f"characters/{cid}.png",
                         f"characters/{cid}_preview.png"]
                files.extend([p["file"] for p in part_files])
                preview = f"characters/{cid}_preview.png"
            else:
                full_prompt = ", ".join(x for x in [
                    c.get("prompt", ""), style,
                    transparent_prompt("single game icon character or slot symbol, centered, full body in frame, no text, not a boss, not a demon lord, not dark armor"),
                ] if x)
                log(f"🎨 [{cid}] 生成角色图…")
                img = generate_checked(cid, full_prompt, c.get("size", creds.get("size", "1024x1024")), True)
                spine_builder.build_character(cid, img, chars_out, anims)
                w, h = spine_builder.trim_to_content(img).size
                files = [f"characters/{cid}.json", f"characters/{cid}.atlas", f"characters/{cid}.png"]
                preview = ""
            upsert("characters", {
                "id": cid,
                "png": f"characters/{cid}.png",
                "w": w, "h": h,
                "type": c.get("type", "spine"),
                "role": c.get("role", ""),
                "preview": preview,
                "animations": spine_builder.anim_data(anims),
                "files": files,
                "parts": part_files,
            })
            log(f"✅ [{cid}] 完成 ({anims})")
            progress(f"{cid}")
        except Exception as e:
            log(f"❌ [{cid}] 失败: {e}")
            if is_required_boss(c):
                required_failures.append(f"{cid}: {e}")
            progress(f"{cid}")

    boss_id = required_boss_id()
    boss_missing_after_chars = (
        boss_id and boss_required_in_this_run
        and not any(c.get("id") == boss_id for c in library["characters"])
    )

    # ---- A2. UI 美术（背景 / Logo / 卷轴框 / 按钮 等整图）----
    ui_art_out = os.path.join(base_out, "ui_art")
    for a in manifest.get("ui_art", []):
        aid = a.get("id", "art")
        if not creds.get("api_key"):
            log(f"⚠️  未填 key，跳过 UI 美术 {aid}")
            continue
        try:
            transparent = a.get("transparent", True)
            extra = (transparent_prompt("single clean UI element, no text")
                     if transparent
                     else "full-bleed illustration, no text, no UI elements")
            full_prompt = ", ".join(x for x in [a.get("prompt", ""), style if a.get("use_style") else "", extra] if x)
            log(f"🖼  [{aid}] 生成 UI 美术…")
            img = generate_checked(aid, full_prompt, a.get("size", creds.get("size", "1024x1024")), transparent)
            os.makedirs(ui_art_out, exist_ok=True)
            img.save(os.path.join(ui_art_out, f"{aid}.png"))
            upsert("ui_art", {"id": aid, "file": f"ui_art/{aid}.png",
                              "w": img.width, "h": img.height,
                              "transparent": transparent})
            log(f"✅ [{aid}] UI 美术完成")
            progress(f"{aid}")
        except Exception as e:
            log(f"❌ [{aid}] UI 美术失败: {e}")
            progress(f"{aid}")

    # ---- B. 粒子 VFX ----
    for v in manifest.get("vfx", []):
        vid = v.get("id", "vfx")
        try:
            path = particle_builder.build_particle(
                vid, v.get("template", "burst"), v.get("color", [255, 255, 255]), vfx_out)
            cfg = json.load(open(path, encoding="utf-8"))
            upsert("vfx", {"id": vid, "template": v.get("template"),
                           "file": f"vfx/{vid}.particle.json", "config": cfg})
            log(f"✨ [{vid}] 粒子配置完成")
            progress(f"{vid}")
        except Exception as e:
            log(f"❌ [{vid}] 粒子失败: {e}")
            progress(f"{vid}")

    # ---- C. UI Tween ----
    ui = manifest.get("ui", [])
    if ui:
        used = [u.get("preset") for u in ui if u.get("preset")]
        try:
            tween_builder.build_tweens(used, ui_out)
            for u in ui:
                upsert("ui", {"id": u.get("id"), "preset": u.get("preset"),
                              "params": u.get("params", {})})
            log(f"🎛  TweenPresets.ts 完成 ({used})")
            progress("TweenPresets")
        except Exception as e:
            log(f"❌ Tween 失败: {e}")
            progress("TweenPresets")

    os.makedirs(base_out, exist_ok=True)
    with open(os.path.join(base_out, "library.json"), "w", encoding="utf-8") as f:
        json.dump(library, f, ensure_ascii=False, indent=2)
    log("—— 完成 ——")
    if boss_missing_after_chars:
        detail = "；".join(required_failures) if required_failures else "生成结果中没有关主资源"
        raise RuntimeError(
            f"关主大魔王资源缺失：{boss_id}。已保存其他成功资源；请在任务卡片里继续补生成 boss。原因：{detail}"
        )
    return library, base_out