"""图像大模型调用层（仅用标准库 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.info["source_url"] = source_url or "" 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)