games-creater/providers.py

"""图像大模型调用层（仅用标准库 urllib，无第三方依赖）。

内置 OpenAI 图像接口（gpt-image-1，支持透明背景）。
接其他厂商：实现同签名函数加进 PROVIDERS 即可（兼容 OpenAI 协议的中转 Base URL 可直接用）。
"""

import base64
import http.client
import io
import json
import os
import ssl
import urllib.request
from urllib.parse import urlparse
from PIL import Image


def _ssl_context():
    """构造可信的 SSL 上下文。
    优先用 certifi 的 CA 证书（解决 macOS python.org 找不到根证书的
    CERTIFICATE_VERIFY_FAILED 问题）；都没有时退回系统默认。
    设环境变量 ANIM_STUDIO_INSECURE=1 可临时关闭校验（不推荐，仅排错用）。
    """
    if os.environ.get("ANIM_STUDIO_INSECURE") == "1":
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        return ctx
    try:
        import certifi
        return ssl.create_default_context(cafile=certifi.where())
    except Exception:
        return ssl.create_default_context()


_SSL = _ssl_context()


def _http_json(url, payload, headers, timeout):
    data = json.dumps(payload).encode("utf-8")
    parsed = urlparse(url)
    path = parsed.path or "/"
    if parsed.query:
        path += "?" + parsed.query
    conn_cls = http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection
    kwargs = {"timeout": timeout}
    if parsed.scheme == "https":
        kwargs["context"] = _SSL
    conn = conn_cls(parsed.hostname, parsed.port, **kwargs)
    try:
        conn.request("POST", path, body=data, headers=headers)
        res = conn.getresponse()
        body = res.read()
    finally:
        conn.close()
    if res.status < 200 or res.status >= 300:
        raise urllib.error.HTTPError(url, res.status, res.reason, res.headers, io.BytesIO(body))
    return json.loads(body.decode("utf-8"))


def _http_bytes(url, timeout):
    parsed = urlparse(url)
    path = parsed.path or "/"
    if parsed.query:
        path += "?" + parsed.query
    conn_cls = http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection
    kwargs = {"timeout": timeout}
    if parsed.scheme == "https":
        kwargs["context"] = _SSL
    conn = conn_cls(parsed.hostname, parsed.port, **kwargs)
    try:
        conn.request("GET", path)
        res = conn.getresponse()
        body = res.read()
    finally:
        conn.close()
    if res.status < 200 or res.status >= 300:
        raise RuntimeError(f"下载生成图失败：{url} 返回 {res.status}")
    return body


def _url_image_to_data_url(url, timeout, max_bytes=8 * 1024 * 1024):
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    with urllib.request.urlopen(req, timeout=timeout, context=_SSL) as res:
        ctype = (res.headers.get("content-type") or "image/png").split(";")[0].strip()
        raw = res.read(max_bytes + 1)
    if len(raw) > max_bytes:
        raise RuntimeError("image is larger than 8MB")
    if not ctype.startswith("image/"):
        ctype = "image/png"
    return f"data:{ctype};base64," + base64.b64encode(raw).decode("ascii")


def _short_json(data, limit=600):
    try:
        return json.dumps(data, ensure_ascii=False)[:limit]
    except Exception:
        return str(data)[:limit]


def _read_http_error(error):
    body = error.read().decode("utf-8", "ignore")[:1200]
    return body


def _should_retry_without_url_fields(status, body):
    if status not in (400, 422):
        return False
    text = body.lower()
    retry_markers = [
        "response_format",
        "store",
        "background",
        "output_format",
        "unknown parameter",
        "unsupported parameter",
        "unrecognized",
        "extra inputs",
        "not permitted",
    ]
    return any(marker in text for marker in retry_markers)


def _extract_image_item(data):
    if isinstance(data, dict):
        if data.get("data"):
            items = data["data"]
        elif data.get("images"):
            items = data["images"]
        else:
            items = [data]
    elif isinstance(data, list):
        items = data
    else:
        items = []

    if not items:
        return {}
    item = items[0]
    if isinstance(item, str):
        if item.startswith("http://") or item.startswith("https://"):
            return {"url": item}
        return {"b64_json": item}
    if not isinstance(item, dict):
        return {}

    image_url = item.get("image_url")
    if isinstance(image_url, dict) and image_url.get("url"):
        return {"url": image_url["url"]}
    return item


def gen_image_openai(prompt, api_key, base_url="https://api.openai.com/v1",
                     model="gpt-image-2", size="1024x1024", timeout=180):
    """调用 OpenAI 兼容图像接口（含第三方中转），返回 PIL.Image (RGBA)。

    兼容性处理：
    - 仅对 gpt-image* 模型发送 background/output_format 等专属参数；其他模型
      （dall-e-3 / flux / sora_image / gpt-4o-image 等中转常见模型）不发，避免被拒。
    - 返回兼容 b64_json 或 url 两种格式。
    """
    url = base_url.rstrip("/") + "/images/generations"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    model_name = model.lower()
    payload = {"model": model, "prompt": prompt, "size": size, "n": 1}
    wants_transparent = "transparent background" in prompt.lower() or "no background" in prompt.lower()
    if model_name == "gpt-image-2":
        # x.long.bid exposes gpt-image-2, but its upstream only accepts the
        # minimal image payload. Adding background/output_format causes 502.
        pass
    elif "gpt-image" in model_name:
        # gpt-image-1 official-compatible endpoints can use these fields.
        # If a gateway rejects them with 400/422, the retry path strips them.
        if wants_transparent:
            payload["background"] = "transparent"
        payload["output_format"] = "png"
    else:
        # Same OpenAI-compatible image flow used by the content platform:
        # ask compatible gateways for a downloadable URL, then still accept b64.
        payload["response_format"] = "url"
        payload["store"] = False
        if wants_transparent:
            payload["background"] = "transparent"
            payload["output_format"] = "png"
    try:
        data = _http_json(url, payload, headers, timeout)
    except urllib.error.HTTPError as e:
        body = _read_http_error(e)
        if _should_retry_without_url_fields(e.code, body):
            fallback = dict(payload)
            fallback.pop("response_format", None)
            fallback.pop("store", None)
            fallback.pop("background", None)
            fallback.pop("output_format", None)
            try:
                data = _http_json(url, fallback, headers, timeout)
            except urllib.error.HTTPError as e2:
                body2 = _read_http_error(e2)
                raise RuntimeError(f"图像接口 HTTP {e2.code}：{body2[:600]}\n"
                                   f"（model={model}, url={url}；已尝试移除 response_format/store 重试）")
        else:
            raise RuntimeError(f"图像接口 HTTP {e.code}：{body[:600]}\n"
                               f"（model={model}, url={url}）")
    except urllib.error.URLError as e:
        raise RuntimeError(f"连不上图像接口：{e}\n（检查 Base URL / 网络 / 证书；url={url}）")

    item = _extract_image_item(data)
    source_url = None
    if item.get("b64_json"):
        raw = base64.b64decode(item["b64_json"])
    elif item.get("base64"):
        raw = base64.b64decode(item["base64"])
    elif item.get("image"):
        raw = base64.b64decode(item["image"])
    elif item.get("url"):
        source_url = item["url"]
        raw = _http_bytes(source_url, timeout)
    else:
        raise RuntimeError(f"返回里既无 b64_json/base64/image 也无 url：{_short_json(data)}")
    img = Image.open(io.BytesIO(raw)).convert("RGBA")
    img = _auto_declutter(img)
    img.info["source_url"] = source_url or ""
    return img


def _auto_declutter(img):
    """统一出口：若模型把"透明背景"画成了灰白棋盘格（真像素、alpha 全 255），
    自动抠成真透明。对本就透明 / 无棋盘格的图（角色、整张背景）是无操作，
    所以可以无差别地套在每张生成图上；失败时绝不影响主流程。"""
    try:
        import dechecker
        cleaned, changed = dechecker.declutter_img(img)
        return cleaned if changed else img
    except Exception:
        return img


PROVIDERS = {"OpenAI 兼容接口": gen_image_openai}


def generate(provider, prompt, api_key, base_url, model, size):
    fn = PROVIDERS.get(provider)
    if fn is None:
        raise ValueError(f"未知 provider: {provider}")
    return fn(prompt, api_key=api_key, base_url=base_url, model=model, size=size)


TEXT_MODEL_FALLBACKS = ["gpt-5.4-mini", "gpt-5.4", "gpt-5.5"]


def _chat_json_once(url, headers, messages, model, timeout):
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.75,
        "response_format": {"type": "json_object"},
    }
    try:
        data = _http_json(url, payload, headers, timeout)
    except urllib.error.HTTPError as e:
        fallback = dict(payload)
        fallback.pop("response_format", None)
        try:
            data = _http_json(url, fallback, headers, timeout)
        except urllib.error.HTTPError as e2:
            raise e2
    content = (((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
    if content.startswith("```"):
        content = content.strip("`")
        if content.lower().startswith("json"):
            content = content[4:].strip()
    try:
        obj = json.loads(content)
    except Exception as e:
        raise RuntimeError(f"文字模型没有返回合法 JSON：{e}；内容={content[:800]}")
    if isinstance(obj, dict):
        obj.setdefault("_model_used", model)
    return obj


def chat_json_openai(messages, api_key, base_url="https://api.openai.com/v1",
                     model="gpt-5.4-mini", timeout=120):
    url = base_url.rstrip("/") + "/chat/completions"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    candidates = [model] + [x for x in TEXT_MODEL_FALLBACKS if x != model]
    errors = []
    for candidate in candidates:
        try:
            return _chat_json_once(url, headers, messages, candidate, timeout)
        except urllib.error.HTTPError as e:
            body = _read_http_error(e)
            errors.append(f"{candidate}: HTTP {e.code} {body[:180]}")
            if e.code not in (429, 500, 502, 503, 504):
                break
    raise RuntimeError(f"文字模型不可用，已尝试 {', '.join(candidates)}："
                       f"{' | '.join(errors)}\n（url={url}；已移除 response_format 重试）")


def analyze_reference_images(reference_urls=None, image_data_urls=None, api_key="",
                             base_url="https://api.openai.com/v1", model="gpt-5.4-mini",
                             timeout=120):
    """Use a vision-capable OpenAI-compatible chat model to summarize style refs.

    `reference_urls` can include direct image URLs. `image_data_urls` are browser
    FileReader data URLs from uploaded screenshots/images.
    """
    reference_urls = [x for x in (reference_urls or []) if x][:4]
    image_data_urls = [x for x in (image_data_urls or []) if x][:4]
    if not reference_urls and not image_data_urls:
        return {}
    content = [{
        "type": "text",
        "text": (
            "分析这些参考图/截图的视觉方向，用于原创移动老虎机游戏美术生成。"
            "只提炼风格，不要复刻具体 IP、logo、角色或版式。"
            "返回 JSON：art_style, palette, materials, ui_shape_language, character_direction, "
            "background_direction, avoid, image_prompt_style。"
        ),
    }]
    for url in reference_urls:
        try:
            content.append({"type": "image_url", "image_url": {"url": _url_image_to_data_url(url, timeout=30)}})
        except Exception as e:
            content.append({
                "type": "text",
                "text": f"参考图 URL 本地下载失败，不能作为视觉输入，只能作为文字线索：{url}；错误：{str(e)[:160]}",
            })
    for data_url in image_data_urls:
        content.append({"type": "image_url", "image_url": {"url": data_url}})
    return chat_json_openai([
        {"role": "system", "content": "你是游戏美术总监，擅长把参考图转成原创美术风格规范。只输出 JSON。"},
        {"role": "user", "content": content},
    ], api_key=api_key, base_url=base_url, model=model, timeout=timeout)