cang-ying/engines/image_api_adapter.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
image_api_adapter.py · Python端图片API适配器
D144 · 铸渊 ICE-GL-ZY001

桥接到 Node.js image-api-bridge.js，为 char-hero-design-packer.py 提供
generate_image() 函数。

本地密钥文件: ~/Documents/guanghulab-local-secrets/video-ai-system.env
不提交仓库，不硬编码密钥。
"""

import json
import os
import subprocess
from pathlib import Path

ENGINES_DIR = Path(__file__).parent
BRIDGE_SCRIPT = ENGINES_DIR / "image-api-bridge.js"
NODE_BIN = "/Users/bingshuolingdianyuanhe/.workbuddy/binaries/node/versions/22.22.2/bin/node"


def _call_bridge(action: str, **kwargs) -> dict:
    """
    调用 Node.js 桥接脚本
    :param action: 操作名 (generate_character_ref | generate_image)
    :param kwargs: 传给 JS 的参数
    :return: {ok: bool, data: {...} | error: str}
    """
    if not BRIDGE_SCRIPT.exists():
        return {"ok": False, "error": f"桥接脚本不存在: {BRIDGE_SCRIPT}"}

    payload = {"action": action, **kwargs}

    try:
        result = subprocess.run(
            [NODE_BIN, str(BRIDGE_SCRIPT)],
            input=json.dumps(payload, ensure_ascii=False),
            capture_output=True,
            text=True,
            timeout=120,
            cwd=str(ENGINES_DIR),
        )
    except FileNotFoundError:
        return {"ok": False, "error": f"Node 未找到: {NODE_BIN}"}
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": "桥接调用超时（120秒）"}

    if result.returncode != 0:
        stderr = result.stderr.strip()[:500] if result.stderr else "无输出"
        return {"ok": False, "error": f"桥接退出码 {result.returncode}: {stderr}"}

    try:
        return json.loads(result.stdout.strip())
    except json.JSONDecodeError:
        return {"ok": False, "error": f"桥接输出非JSON: {result.stdout[:200]}"}


def generate_image(prompt: str, reference_image: str = None,
                   output_dir: str = None, output_name: str = None) -> str:
    """
    生成图片（为 char-hero-design-packer.py 提供接口）

    :param prompt: 提示词
    :param reference_image: 参考图路径（当前桥接为文生图，参考图暂不支持）
    :param output_dir: 输出目录
    :param output_name: 输出文件名
    :return: 生成的图片路径，失败返回 None
    """
    kwargs = {"prompt": prompt, "size": "2048x2048"}

    if output_dir and output_name:
        kwargs["outputPath"] = os.path.join(output_dir, output_name)
    elif output_name:
        kwargs["filename"] = output_name

    result = _call_bridge("generate_image", **kwargs)

    if not result.get("ok"):
        print(f"  ⚠️  图片生成失败: {result.get('error', '未知错误')}")
        return None

    data = result.get("data", {})
    image_path = data.get("imagePath", "")
    if image_path and os.path.isfile(image_path):
        return image_path
    return None


def generate_character_ref(char_id: str, l0_descriptor: str = None,
                           l1_clothing: str = None) -> str:
    """
    生成角色参考图

    :param char_id: 角色编号，如 CHAR-003
    :param l0_descriptor: L0底层编码描述
    :param l1_clothing: L1服装配置
    :return: 生成的图片路径，失败返回 None
    """
    kwargs = {"charId": char_id}
    if l0_descriptor:
        kwargs["l0"] = l0_descriptor
    if l1_clothing:
        kwargs["l1"] = l1_clothing

    result = _call_bridge("generate_character_ref", **kwargs)

    if not result.get("ok"):
        print(f"  ⚠️  角色参考图生成失败: {result.get('error', '未知错误')}")
        return None

    data = result.get("data", {})
    return data.get("imagePath", "")