我的 API：异步生图接口

from __future__ import annotations

import base64
import json
import time
import urllib.error
import urllib.request

BASE_URL = "https://img.2capi.com"
API_KEY = "c2a_xxx"  # 替换成你的固定 API Key / Replace with your API key
MODEL = "gpt-image-2-5.5"
PROMPT = "请生成一张图片：一只橘猫坐在月光窗台上，水彩风格，只输出图片。"
TASK_ID = f"api_demo_{int(time.time())}"

if not API_KEY or API_KEY == "c2a_xxx":
    raise SystemExit("请先把 API_KEY 替换成你的 c2a_xxx 固定 Key")


def request_json(method: str, path: str, body: dict | None = None) -> dict:
    data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
    headers = {"Accept": "application/json", "Authorization": f"Bearer {API_KEY}"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(BASE_URL + path, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(request, timeout=60) as response:
            return json.loads(response.read().decode("utf-8") or "{}")
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        if exc.code == 429 and "image_task_limited" in raw:
            raise SystemExit("当前固定 Key 所属用户已有图片任务处理中，请等待上一个任务完成后重试。") from exc
        raise RuntimeError(f"HTTP {exc.code}: {raw}") from exc


submit = request_json(
    "POST",
    "/api/image-tasks/generations",
    {
        "client_task_id": TASK_ID,
        "prompt": PROMPT,
        "model": MODEL,
        "n": 1,
    },
)
poll_url = submit.get("poll_url") or f"/api/image-tasks?ids={TASK_ID}"
print("submitted:", json.dumps(submit, ensure_ascii=False))

deadline = time.time() + 360
task = {}
while time.time() < deadline:
    result = request_json("GET", poll_url)
    items = result.get("items") or []
    task = items[0] if items else {}
    status = task.get("status")
    print("status:", status)
    if status in {"succeeded", "failed", "cancelled"}:
        break
    time.sleep(6)

if task.get("status") != "succeeded":
    raise SystemExit(f"任务失败：{json.dumps(task, ensure_ascii=False)}")

image = ((task.get("result") or {}).get("data") or [{}])[0]
with open("async_image_result.png", "wb") as file:
    file.write(base64.b64decode(image.get("b64_json") or ""))

print("saved:", "async_image_result.png")
gallery_task = task.get("gallery_task") or {}
print("gallery_image_url:", gallery_task.get("image_url_public"))
print("说明：优先读 image_url_public，公开图片地址可重复访问，不暴露 API Key。")