feat(image): add gpt-image-2 single and multi-image workflows

.env.example

@@ -0,0 +1,12 @@
+OPENAI_API_KEY=your_api_key_here
+OPENAI_BASE_URL=https://api.aoixx.com
+IMAGE_REQUEST_STYLE=apicodex
+IMAGE_MODEL=gpt-image-2
+IMAGE_SIZE=1024x1024
+IMAGE_PROMPT="Describe the image you want here"
+IMAGE_N=1
+IMAGE_OUTPUT_STEM=gpt-image-2-multi
+IMAGE_RESPONSE_DIR=outputs/image-multi-runs
+REFERENCE_IMAGE=./input.png
+# Multi-image alternative:
+# REFERENCE_IMAGES=["./a.png","./b.png"]

.gitignore

@@ -0,0 +1,20 @@
+.env
+.venv/
+venv/
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.ipynb_checkpoints/
+.vscode/
+.idea/
+*.log
+*.swp
+*.swo
+build/
+dist/
+tmp/
+outputs/
+references/
+.DS_Store

README.md

@@ -0,0 +1,46 @@
+# image-generate
+
+这是一个基于 `gpt-image-2` 的图片生成小项目，支持 OpenAI-compatible / apicodex-compatible 接口。
+
+## 安装
+
+```bash
+python3 -m pip install -r requirements.txt
+```
+
+## 入口
+
+- `gpt_image2_generate.py`：单参考图 / 文生图 / 图生图入口
+- `gpt_image2_generate_multi.py`：多参考图、多输出入口
+- `scripts/*.sh`：常用启动脚本
+
+## 快速开始
+
+单参考图：
+
+```bash
+bash scripts/run_openai_gpt_image2_edit.sh
+```
+
+多参考图、多输出：
+
+```bash
+bash scripts/run_openai_gpt_image2_multi.sh
+```
+
+## 配置
+
+复制 `.env.example` 为 `.env`，按需填写：
+
+- `OPENAI_API_KEY`
+- `OPENAI_BASE_URL`
+- `IMAGE_PROMPT`
+- `IMAGE_SIZE`
+- `REFERENCE_IMAGE` 或 `REFERENCE_IMAGES`
+
+详细说明见 [docs/usage.md](docs/usage.md)。
+
+## 仓库约定
+
+- `outputs/` 是运行产物目录，默认不纳入版本控制。
+- `references/` 是本地参考图目录，默认不纳入版本控制。

docs/usage.md

@@ -0,0 +1,316 @@
+# 图片生成接口指南
+
+## 1. 环境准备
+
+Windows PowerShell：
+
+```powershell
+py -m pip install requests pillow
+```
+
+Mac / Linux Terminal：
+
+```bash
+python3 -m pip install requests pillow
+```
+
+## 2. 接口地址
+
+`POST https://api.toskaxy.xyz/v1/images/generations`
+
+## 3. 鉴权方式
+
+```http
+Authorization: Bearer pk_live_xxx
+Content-Type: application/json
+```
+
+## 4. 支持的图片模型
+
+- `gpt-image-1`
+- `gpt-image-1.5`
+- `gpt-image-2`
+
+## 5. 文生图最小字段
+
+- `model`：图片模型名称，例如 `gpt-image-2`
+- `prompt`：图片提示词
+- `size`：输出尺寸。推荐直接传像素尺寸，例如 `4096x4096`、`4096x2304`、`2304x4096`
+- `resolution`：兼容字段。使用比例写法时传 `1K`、`2K`、`4K`，网关会转换成实际像素尺寸
+
+## 6. `gpt-image-2` 的 4K 尺寸规则
+
+推荐写法：直接传像素。
+
+```json
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "4096x4096" }
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "4096x2304" }
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "2304x4096" }
+```
+
+兼容写法：比例 + `resolution`。
+
+```json
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "1:1", "resolution": "4K" }
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "16:9", "resolution": "4K" }
+{ "model": "gpt-image-2", "prompt": "draw a clean product poster", "size": "9:16", "resolution": "4K" }
+```
+
+## 7. 比例转换示例
+
+- `1:1` + `4K` -> `4096x4096`
+- `16:9` + `4K` -> `4096x2304`
+- `9:16` + `4K` -> `2304x4096`
+- `3:4` + `4K` -> `3072x4096`
+- `4:3` + `4K` -> `4096x3072`
+
+## 8. 旧图片模型的 4K 降级规则
+
+`gpt-image-1` 和 `gpt-image-1.5` 不支持真正 4K。如果用户传 `resolution: "4K"`，网关会自动降级成上游支持的最大尺寸。
+
+- 横图比例：`1536x1024`
+- 竖图比例：`1024x1536`
+- 方图比例：`1024x1024`
+
+## 9. 支持的比例
+
+`1:1`、`16:9`、`9:16`、`4:3`、`3:4`、`3:2`、`2:3`、`5:4`、`4:5`、`2:1`、`1:2`、`21:9`、`9:21`
+
+建议只通过 `size` 传比例，不要在 `prompt` 里重复写比例。
+
+## 10. 4K 文生图完整 Python 示例
+
+把 `API_KEY` 改成自己的密钥，保存为 `gpt_image2_generate.py` 后运行。
+
+```python
+import base64
+import requests
+
+API_KEY = "pk_live_xxx"
+BASE_URL = "https://api.toskaxy.xyz"
+
+payload = {
+    "model": "gpt-image-2",
+    "prompt": "高端香水产品海报，玻璃瓶，干净背景，商业摄影，细节清晰",
+    "n": 1,
+    "size": "4096x4096",
+    "quality": "high",
+    "response_format": "b64_json",
+}
+
+headers = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+}
+
+resp = requests.post(
+    f"{BASE_URL}/v1/images/generations",
+    json=payload,
+    headers=headers,
+    timeout=240,
+)
+resp.raise_for_status()
+data = resp.json()
+
+img_b64 = data["data"][0]["b64_json"]
+with open("output.png", "wb") as f:
+    f.write(base64.b64decode(img_b64))
+
+print("已保存 output.png")
+```
+
+## 11. 图生图：参考图为公开 URL
+
+```python
+import base64
+import requests
+
+API_KEY = "pk_live_xxx"
+BASE_URL = "https://api.toskaxy.xyz"
+
+payload = {
+    "model": "gpt-image-2",
+    "prompt": "把参考图变成水彩画风格，保留主体结构",
+    "n": 1,
+    "size": "2048x2048",
+    "quality": "high",
+    "response_format": "b64_json",
+    "image_urls": [
+        "https://example.com/reference.jpg"
+    ],
+}
+
+headers = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+}
+
+resp = requests.post(
+    f"{BASE_URL}/v1/images/generations",
+    json=payload,
+    headers=headers,
+    timeout=240,
+)
+resp.raise_for_status()
+data = resp.json()
+
+img_b64 = data["data"][0]["b64_json"]
+with open("output.png", "wb") as f:
+    f.write(base64.b64decode(img_b64))
+
+print("已保存 output.png")
+```
+
+## 12. 图生图：参考图为本地文件 base64
+
+把 `reference.png` 放到脚本同一个文件夹，脚本会把它转成 data URL 再提交。
+
+```python
+import base64
+import requests
+
+API_KEY = "pk_live_xxx"
+BASE_URL = "https://api.toskaxy.xyz"
+REFERENCE_FILE = "reference.png"
+
+with open(REFERENCE_FILE, "rb") as f:
+    reference_b64 = base64.b64encode(f.read()).decode("utf-8")
+
+payload = {
+    "model": "gpt-image-2",
+    "prompt": "基于参考图生成一张高端产品海报，干净构图，中文界面可用",
+    "n": 1,
+    "size": "4096x2304",
+    "quality": "high",
+    "response_format": "b64_json",
+    "image_urls": [
+        f"data:image/png;base64,{reference_b64}"
+    ],
+}
+
+headers = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+}
+
+resp = requests.post(
+    f"{BASE_URL}/v1/images/generations",
+    json=payload,
+    headers=headers,
+    timeout=240,
+)
+resp.raise_for_status()
+data = resp.json()
+
+img_b64 = data["data"][0]["b64_json"]
+with open("output.png", "wb") as f:
+    f.write(base64.b64decode(img_b64))
+
+print("已保存 output.png")
+```
+
+## 13. 下载后校验是否为 4K
+
+```python
+from io import BytesIO
+from PIL import Image
+import requests
+
+image_url = "<接口返回的图片URL>"
+img_bytes = requests.get(image_url, timeout=120).content
+img = Image.open(BytesIO(img_bytes))
+print(img.size)
+
+width, height = img.size
+assert max(width, height) >= 3840, f"not 4K enough: {img.size}"
+```
+
+## 14. 同步返回与异步轮询兼容写法
+
+```python
+import base64
+import time
+import requests
+
+API_KEY = "pk_live_xxx"
+BASE_URL = "https://api.toskaxy.xyz"
+
+headers = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+}
+
+payload = {
+    "model": "gpt-image-2",
+    "prompt": "a corgi astronaut on the moon, cinematic, 8k",
+    "size": "16:9",
+    "resolution": "4K",
+    "n": 1,
+    "response_format": "b64_json",
+}
+
+resp = requests.post(
+    f"{BASE_URL}/v1/images/generations",
+    json=payload,
+    headers=headers,
+    timeout=240,
+)
+resp.raise_for_status()
+data = resp.json()
+
+# 情况 1：同步返回 base64
+if isinstance(data, dict) and data.get("data") and data["data"][0].get("b64_json"):
+    with open("output.png", "wb") as f:
+        f.write(base64.b64decode(data["data"][0]["b64_json"]))
+    print("已保存 output.png")
+
+# 情况 2：异步任务返回 task_id
+elif isinstance(data, dict) and isinstance(data.get("data"), dict) and data["data"].get("id"):
+    task_id = data["data"]["id"]
+    task_url = f"{BASE_URL}/v1/tasks/{task_id}"
+
+    time.sleep(10)
+    while True:
+        task_resp = requests.get(task_url, headers=headers, timeout=60)
+        task_resp.raise_for_status()
+        task = task_resp.json()
+
+        status = task["data"]["status"]
+        if status == "completed":
+            print(task["data"]["result"]["images"][0]["url"][0])
+            break
+        if status == "failed":
+            raise RuntimeError(task["data"]["error"]["message"])
+
+        time.sleep(5)
+else:
+    raise RuntimeError(f"unexpected response: {data}")
+```
+
+## 15. 图片接口常见问题
+
+- 只传 `size: "16:9"` 不能保证一定是 4K；要 4K，推荐直接传 `4096x2304` 等像素尺寸。
+- `gpt-image-1` 和 `gpt-image-1.5` 不支持真正 4K，网关会自动降级。
+- `gpt-image-2` 的图生图不要把参考图塞进 `/v1/images/generations` 的 JSON。应改走 `/v1/images/edits`，并用 `multipart/form-data` 上传 `image` 或 `image[]`。
+- 单张图可能耗时较长，客户端 `timeout` 建议至少 120 秒；4K 建议 240 秒。
+- 如果返回 `b64_json`，说明当前是同步模式，不需要再轮询 `task_id`。
+
+## 16. 直连测试命令
+
+- 运行 `scripts/run_figma_direct_test.sh` 会先读取当前 `.env`，再临时清空本地代理并直连 `vip.auto-code.net`。
+- 它只强制 `IMAGE_REQUEST_STYLE=apicodex`、`IMAGE_SIZE=3840x2160` 和长超时，其余参数继续沿用 `.env` 和 `gpt_image2_generate.py` 的默认值。
+- 每次请求会落到 `outputs/image-demo-runs/<时间戳>/`。
+- 直接运行：`bash scripts/run_figma_direct_test.sh`
+- 只测 `https://api.aoixx.com/v1/images/generations` 时，运行 `bash scripts/run_aoixx_direct_test.sh`。
+- 如果要走 OpenAI-compatible 的 `gpt-image-2` 图生图，运行 `bash scripts/run_openai_gpt_image2_edit.sh`。
+- 如果要走多图参考生成多图，运行 `bash scripts/run_openai_gpt_image2_multi.sh`。
+- 图生图可通过环境变量传参考图：
+  `REFERENCE_IMAGE=./input.png`，或 `REFERENCE_IMAGE_URL=https://example.com/input.png`。
+- 如需 mask，可额外传：
+  `MASK_IMAGE=./mask.png`，或 `MASK_IMAGE_URL=https://example.com/mask.png`。
+- 多图参考推荐两种写法：
+  `REFERENCE_IMAGE_1=./a.png`、`REFERENCE_IMAGE_2=./b.png`，或
+  `REFERENCE_IMAGES='["./a.png","./b.png","./c.png"]'`。
+- 多图输出用 `IMAGE_N=3` 控制，未设置时默认等于参考图数量。
+- 示例：
+  `REFERENCE_IMAGE=./input.png IMAGE_PROMPT="改成极简白底产品海报" bash scripts/run_openai_gpt_image2_edit.sh`

gpt_image2_generate.py

@@ -0,0 +1,573 @@
+from __future__ import annotations
+
+import base64
+import json
+import mimetypes
+import os
+import traceback
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from io import BytesIO
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+import requests
+from PIL import Image, ImageFile
+
+
+DEFAULT_PROMPT = "高端电子产品海报，干净背景，商业摄影，细节清晰"
+DEFAULT_MODEL = "gpt-image-2"
+DEFAULT_PIXEL_SIZE = "4096x4096"
+DEFAULT_QUALITY = "high"
+DEFAULT_OUTPUT_FILE = "output.png"
+DEFAULT_RESPONSE_DIR = "outputs/image-demo-runs"
+DEFAULT_CONNECT_TIMEOUT = 10
+DEFAULT_READ_TIMEOUT = 900
+DEFAULT_IMAGE_FETCH_TIMEOUT = 120
+DEFAULT_POLL_CONNECT_TIMEOUT = 10
+DEFAULT_POLL_READ_TIMEOUT = 60
+DEFAULT_POLL_INTERVAL = 5
+DEFAULT_INITIAL_POLL_DELAY = 10
+DEFAULT_MAX_POLL_WAIT = 1800
+DIRECT_HOSTS = {"vip.auto-code.net","api.toskaxy.xyz","api.aoixx.com"}
+SUPPORTED_ASPECT_RATIOS = {
+    "1:1",
+    "16:9",
+    "9:16",
+    "4:3",
+    "3:4",
+    "3:2",
+    "2:3",
+    "5:4",
+    "4:5",
+    "2:1",
+    "1:2",
+    "21:9",
+    "9:21",
+}
+OPENAI_RATIO_SIZE_MAP = {
+    "1:1": "1024x1024",
+    "16:9": "1536x864",
+    "9:16": "864x1536",
+    "4:3": "1536x1152",
+    "3:4": "1152x1536",
+    "3:2": "1536x1024",
+    "2:3": "1024x1536",
+    "5:4": "1280x1024",
+    "4:5": "1024x1280",
+    "2:1": "1536x768",
+    "1:2": "768x1536",
+    "21:9": "1680x720",
+    "9:21": "720x1680",
+}
+
+
+@dataclass(frozen=True)
+class ImageInput:
+    filename: str
+    mime_type: str
+    data: bytes
+
+
+def load_env_file(path: str = ".env") -> None:
+    env_path = Path(path)
+    if not env_path.exists():
+        return
+
+    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+
+        key, value = line.split("=", 1)
+        key = key.strip()
+        value = value.strip().strip('"').strip("'")
+        if key and key not in os.environ:
+            os.environ[key] = value
+
+
+def is_pixel_size(size: str) -> bool:
+    parts = size.lower().split("x")
+    return len(parts) == 2 and all(part.isdigit() for part in parts)
+
+
+def normalize_size(size: str) -> str:
+    value = size.strip().lower().replace(" ", "")
+    if is_pixel_size(value):
+        width, height = value.split("x", 1)
+        return f"{int(width)}x{int(height)}"
+    return value
+
+
+def is_solid_black_image(img: Image.Image) -> bool:
+    rgb = img.convert("RGB")
+    return all(lo == 0 and hi == 0 for lo, hi in rgb.getextrema())
+
+
+def env_int(name: str, default: int) -> int:
+    raw_value = os.getenv(name, "").strip()
+    if not raw_value:
+        return default
+
+    try:
+        return int(raw_value)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} 不是有效整数: {raw_value}") from exc
+
+
+def make_run_dir() -> Path:
+    response_root = Path(os.getenv("IMAGE_RESPONSE_DIR", DEFAULT_RESPONSE_DIR)).expanduser()
+    run_id = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
+    run_dir = response_root / run_id
+    run_dir.mkdir(parents=True, exist_ok=False)
+    return run_dir
+
+
+def dump_json(path: Path, data: Any) -> None:
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def save_http_response(path: Path, response: requests.Response) -> dict[str, Any]:
+    record: dict[str, Any] = {
+        "url": response.url,
+        "status_code": response.status_code,
+        "headers": dict(response.headers),
+    }
+
+    try:
+        record["json"] = response.json()
+    except ValueError:
+        record["body_text"] = response.text
+
+    dump_json(path, record)
+    return record
+
+
+def save_error_artifact(path: Path, error: BaseException, extra: dict[str, Any] | None = None) -> None:
+    payload: dict[str, Any] = {
+        "error_type": type(error).__name__,
+        "error_message": str(error),
+        "traceback": traceback.format_exc(),
+    }
+    if extra:
+        payload.update(extra)
+    dump_json(path, payload)
+
+
+def resolve_output_path(run_dir: Path, output_file: str) -> Path:
+    output_path = Path(output_file).expanduser()
+    if output_path.is_absolute():
+        return output_path
+    return run_dir / output_path
+
+
+def normalize_base_url(base_url: str) -> str:
+    normalized = base_url.rstrip("/")
+    if normalized.endswith("/v1"):
+        return normalized[:-3]
+    return normalized
+
+
+def build_request_session(base_url: str) -> requests.Session:
+    session = requests.Session()
+    hostname = urlparse(base_url).hostname or ""
+    if hostname in DIRECT_HOSTS:
+        session.trust_env = False
+        print(f"直连 {hostname}，忽略本地代理环境变量")
+    return session
+
+
+def resolve_request_style(base_url: str) -> str:
+    request_style = os.getenv("IMAGE_REQUEST_STYLE", "auto").strip().lower()
+    if request_style in {"openai", "apicodex"}:
+        return request_style
+
+    hostname = (urlparse(base_url).hostname or "").lower()
+    if hostname.endswith("openai.com"):
+        return "openai"
+    return "apicodex"
+
+
+def resolve_api_size(size: str, request_style: str) -> str:
+    if request_style == "openai" and size in OPENAI_RATIO_SIZE_MAP:
+        return OPENAI_RATIO_SIZE_MAP[size]
+    return size
+
+
+def resolve_image_source(prefix: str) -> str | None:
+    candidates = {
+        prefix: os.getenv(prefix, "").strip(),
+        f"{prefix}_FILE": os.getenv(f"{prefix}_FILE", "").strip(),
+        f"{prefix}_URL": os.getenv(f"{prefix}_URL", "").strip(),
+        f"{prefix}_DATA_URL": os.getenv(f"{prefix}_DATA_URL", "").strip(),
+    }
+    provided = [(name, value) for name, value in candidates.items() if value]
+    if len(provided) > 1:
+        raise RuntimeError(
+            f"{prefix} 相关变量只能设置一个，当前同时设置了: "
+            + ", ".join(name for name, _ in provided)
+        )
+    if not provided:
+        return None
+    return provided[0][1]
+
+
+def safe_filename(filename: str, fallback: str, mime_type: str) -> str:
+    candidate = Path(filename).name.strip()
+    if not candidate:
+        candidate = fallback
+    if not Path(candidate).suffix:
+        candidate += mimetypes.guess_extension(mime_type) or ".png"
+    return candidate
+
+
+def data_url_to_image_input(data_url: str, fallback: str) -> ImageInput:
+    if not data_url.startswith("data:image/") or "," not in data_url:
+        raise RuntimeError("data URL 参考图格式无效")
+
+    header, data_b64 = data_url.split(",", 1)
+    if ";base64" not in header.lower():
+        raise RuntimeError("data URL 参考图必须使用 base64 编码")
+
+    mime_type = header[5:].split(";", 1)[0] or "image/png"
+    try:
+        data = base64.b64decode(data_b64, validate=True)
+    except ValueError as exc:
+        raise RuntimeError("data URL 参考图 base64 无法解码") from exc
+
+    return ImageInput(
+        filename=safe_filename("", fallback, mime_type),
+        mime_type=mime_type,
+        data=data,
+    )
+
+
+def local_file_to_image_input(file_path: str, fallback: str) -> ImageInput:
+    path = Path(file_path).expanduser()
+    if not path.is_absolute():
+        path = Path.cwd() / path
+    if not path.exists():
+        raise FileNotFoundError(f"找不到参考图文件: {file_path}")
+
+    mime_type = mimetypes.guess_type(path.name)[0] or "image/png"
+    return ImageInput(
+        filename=safe_filename(path.name, fallback, mime_type),
+        mime_type=mime_type,
+        data=path.read_bytes(),
+    )
+
+
+def fetch_url_to_image_input(image_url: str, fallback: str, session: requests.Session) -> ImageInput:
+    resp = session.get(image_url, timeout=(DEFAULT_CONNECT_TIMEOUT, DEFAULT_IMAGE_FETCH_TIMEOUT))
+    resp.raise_for_status()
+
+    mime_type = (resp.headers.get("Content-Type") or "").split(";", 1)[0].strip()
+    if not mime_type.startswith("image/"):
+        mime_type = mimetypes.guess_type(urlparse(image_url).path)[0] or "image/png"
+
+    filename = safe_filename(Path(urlparse(image_url).path).name, fallback, mime_type)
+    return ImageInput(filename=filename, mime_type=mime_type, data=resp.content)
+
+
+def image_source_to_input(source: str, fallback: str, session: requests.Session) -> ImageInput:
+    value = source.strip()
+    lowered = value.lower()
+    if lowered.startswith("data:image/"):
+        return data_url_to_image_input(value, fallback)
+    if lowered.startswith(("http://", "https://")):
+        return fetch_url_to_image_input(value, fallback, session)
+    return local_file_to_image_input(value, fallback)
+
+
+def build_payload(base_url: str) -> tuple[dict[str, Any], str, str, str, str, str | None, str | None]:
+    model = os.getenv("IMAGE_MODEL", DEFAULT_MODEL).strip()
+    prompt = os.getenv("IMAGE_PROMPT", DEFAULT_PROMPT).strip() or DEFAULT_PROMPT
+    requested_size = normalize_size(os.getenv("IMAGE_SIZE", DEFAULT_PIXEL_SIZE))
+    resolution = os.getenv("IMAGE_RESOLUTION", "").strip()
+    quality = os.getenv("IMAGE_QUALITY", DEFAULT_QUALITY).strip()
+    output_file = os.getenv("OUTPUT_FILE", DEFAULT_OUTPUT_FILE).strip()
+    request_style = resolve_request_style(base_url)
+    if not (is_pixel_size(requested_size) or requested_size in SUPPORTED_ASPECT_RATIOS):
+        raise RuntimeError(f"不支持的 IMAGE_SIZE: {requested_size}")
+    api_size = resolve_api_size(requested_size, request_style)
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "prompt": prompt,
+        "n": int(os.getenv("IMAGE_N", "1")),
+        "size": api_size,
+        "quality": quality,
+    }
+    if not (request_style == "openai" and model.startswith("gpt-image")):
+        payload["response_format"] = "b64_json"
+
+    if not is_pixel_size(requested_size) and request_style != "openai":
+        payload["resolution"] = resolution or "4K"
+
+    reference_source = resolve_image_source("REFERENCE_IMAGE")
+    mask_source = resolve_image_source("MASK_IMAGE")
+    if mask_source and not reference_source:
+        raise RuntimeError("MASK_IMAGE 只能和 REFERENCE_IMAGE 一起使用")
+
+    request_mode = "edit" if reference_source else "generate"
+    return payload, model, requested_size, output_file, request_mode, reference_source, mask_source
+
+
+def payload_to_form_fields(payload: dict[str, Any]) -> dict[str, str]:
+    fields: dict[str, str] = {}
+    for key, value in payload.items():
+        if value is None:
+            continue
+        fields[key] = str(value)
+    return fields
+
+
+def build_edit_files(
+    request_style: str,
+    reference_input: ImageInput,
+    mask_input: ImageInput | None,
+) -> list[tuple[str, tuple[str, bytes, str]]]:
+    image_field_name = "image" if request_style == "openai" else "image[]"
+    files: list[tuple[str, tuple[str, bytes, str]]] = [
+        (
+            image_field_name,
+            (reference_input.filename, reference_input.data, reference_input.mime_type),
+        )
+    ]
+    if mask_input is not None:
+        files.append(("mask", (mask_input.filename, mask_input.data, mask_input.mime_type)))
+    return files
+
+
+def resolve_min_output_edge(requested_size: str, resolution: str | None, request_style: str) -> int | None:
+    if is_pixel_size(requested_size):
+        width, height = requested_size.split("x", 1)
+        min_edge = max(int(width), int(height))
+        return min_edge if min_edge >= 3840 else None
+    if resolution and resolution.strip().upper() == "4K" and request_style != "openai":
+        return 3840
+    return None
+
+
+def save_image_bytes(image_bytes: bytes, output_path: Path, min_output_edge: int | None = None) -> tuple[int, int]:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    previous_truncated_setting = ImageFile.LOAD_TRUNCATED_IMAGES
+    ImageFile.LOAD_TRUNCATED_IMAGES = True
+    try:
+        with Image.open(BytesIO(image_bytes)) as img:
+            img.load()
+            width, height = img.size
+            if is_solid_black_image(img):
+                raise RuntimeError(
+                    "上游返回了疑似纯黑占位图 "
+                    f"({width}x{height})，通常表示请求没有走到正常生成链路。"
+                    "请检查 OPENAI_BASE_URL、IMAGE_REQUEST_STYLE、model 和 size，"
+                    "必要时切换到 vip.auto-code.net。"
+                )
+
+            if min_output_edge is not None and max(width, height) < min_output_edge:
+                raise RuntimeError(f"生成结果未达到预期尺寸下限 {min_output_edge}px：{width}x{height}")
+    finally:
+        ImageFile.LOAD_TRUNCATED_IMAGES = previous_truncated_setting
+
+    output_path.write_bytes(image_bytes)
+    print(f"已保存 {output_path} ({width}x{height})")
+
+    return width, height
+
+
+def extract_image_bytes_from_item(item: dict[str, Any], session: requests.Session) -> bytes:
+    if item.get("b64_json"):
+        return base64.b64decode(item["b64_json"])
+
+    url_value = item.get("url")
+    if isinstance(url_value, list) and url_value:
+        image_url = url_value[0]
+    elif isinstance(url_value, str):
+        image_url = url_value
+    else:
+        image_url = None
+
+    if image_url:
+        resp = session.get(image_url, timeout=(DEFAULT_CONNECT_TIMEOUT, DEFAULT_IMAGE_FETCH_TIMEOUT))
+        resp.raise_for_status()
+        return resp.content
+
+    raise RuntimeError(f"响应里没有可用的图片数据: {item}")
+
+
+def poll_task(
+    task_id: str,
+    base_url: str,
+    headers: dict[str, str],
+    session: requests.Session,
+    run_dir: Path,
+    max_wait_seconds: int,
+    initial_poll_delay: int,
+    poll_interval: int,
+) -> bytes:
+    task_url = f"{base_url}/v1/tasks/{task_id}"
+    deadline = time.monotonic() + max_wait_seconds
+    poll_index = 0
+
+    time.sleep(initial_poll_delay)
+    while True:
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"任务轮询超时: task_id={task_id}")
+
+        poll_index += 1
+        task_resp = session.get(
+            task_url,
+            headers=headers,
+            timeout=(DEFAULT_POLL_CONNECT_TIMEOUT, DEFAULT_POLL_READ_TIMEOUT),
+        )
+        task_record = save_http_response(run_dir / f"poll-{poll_index:03d}.json", task_resp)
+        task_resp.raise_for_status()
+        task = task_record.get("json")
+        if not isinstance(task, dict):
+            raise RuntimeError(f"任务响应不是 JSON 对象: {task_record}")
+
+        data = task.get("data")
+        if not isinstance(data, dict):
+            raise RuntimeError(f"任务响应格式异常: {task}")
+
+        status = data.get("status")
+        print(f"任务轮询 #{poll_index}: status={status}")
+        if status == "completed":
+            result = data.get("result", {})
+            images = result.get("images", [])
+            if not images:
+                raise RuntimeError(f"任务已完成，但没有返回图片: {task}")
+            return extract_image_bytes_from_item(images[0], session)
+
+        if status == "failed":
+            error = data.get("error", {})
+            if isinstance(error, dict):
+                message = error.get("message") or str(error)
+            else:
+                message = str(error)
+            raise RuntimeError(message or "任务失败")
+
+        time.sleep(poll_interval)
+
+
+def main() -> None:
+    load_env_file()
+
+    api_key = os.getenv("OPENAI_API_KEY") or os.getenv("API_KEY")
+    base_url = normalize_base_url(
+        os.getenv("OPENAI_BASE_URL") or os.getenv("BASE_URL") or "https://api.toskaxy.xyz"
+    )
+    if not api_key:
+        raise RuntimeError("未找到 API Key，请先在 .env 中设置 OPENAI_API_KEY")
+
+    payload, model, size, output_file, request_mode, reference_source, mask_source = build_payload(base_url)
+    resolution = payload.get("resolution")
+    run_dir = make_run_dir()
+    output_path = resolve_output_path(run_dir, output_file)
+    session = build_request_session(base_url)
+    request_style = resolve_request_style(base_url)
+    request_timeout = env_int("IMAGE_REQUEST_TIMEOUT", DEFAULT_READ_TIMEOUT)
+    max_poll_wait = env_int("IMAGE_MAX_POLL_WAIT", DEFAULT_MAX_POLL_WAIT)
+    initial_poll_delay = env_int("IMAGE_INITIAL_POLL_DELAY", DEFAULT_INITIAL_POLL_DELAY)
+    poll_interval = env_int("IMAGE_POLL_INTERVAL", DEFAULT_POLL_INTERVAL)
+    endpoint_path = "/v1/images/edits" if request_mode == "edit" else "/v1/images/generations"
+    min_output_edge = resolve_min_output_edge(size, resolution, request_style)
+
+    print(
+        "开始生成图片:",
+        f"mode={request_mode}",
+        f"request_style={request_style}",
+        f"model={model}",
+        f"size={size}",
+        f"resolution={resolution or 'none'}",
+        f"request_timeout={request_timeout}s",
+        f"max_poll_wait={max_poll_wait}s",
+        f"initial_poll_delay={initial_poll_delay}s",
+        f"poll_interval={poll_interval}s",
+        f"endpoint={base_url}{endpoint_path}",
+        f"output={output_path}",
+        f"responses={run_dir}",
+    )
+
+    dump_json(run_dir / "request.json", {
+        "base_url": base_url,
+        "endpoint_path": endpoint_path,
+        "request_mode": request_mode,
+        "model": model,
+        "size": size,
+        "resolution": resolution,
+        "output_file": str(output_path),
+        "request_timeout_seconds": request_timeout,
+        "max_poll_wait_seconds": max_poll_wait,
+        "initial_poll_delay_seconds": initial_poll_delay,
+        "poll_interval_seconds": poll_interval,
+        "request_style": request_style,
+        "min_output_edge": min_output_edge,
+        "reference_image_present": bool(reference_source),
+        "mask_image_present": bool(mask_source),
+        "payload": payload,
+    })
+
+    try:
+        if request_mode == "edit":
+            reference_input = image_source_to_input(reference_source or "", "reference-image", session)
+            mask_input = (
+                image_source_to_input(mask_source, "mask-image", session)
+                if mask_source
+                else None
+            )
+            resp = session.post(
+                f"{base_url}{endpoint_path}",
+                data=payload_to_form_fields(payload),
+                files=build_edit_files(request_style, reference_input, mask_input),
+                headers={"Authorization": f"Bearer {api_key}"},
+                timeout=(DEFAULT_CONNECT_TIMEOUT, request_timeout),
+            )
+        else:
+            resp = session.post(
+                f"{base_url}{endpoint_path}",
+                json=payload,
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json",
+                },
+                timeout=(DEFAULT_CONNECT_TIMEOUT, request_timeout),
+            )
+        post_record = save_http_response(run_dir / "post-response.json", resp)
+        resp.raise_for_status()
+        data = post_record.get("json")
+        if not isinstance(data, dict):
+            raise RuntimeError(f"post response is not a JSON object: {post_record}")
+
+        response_data = data.get("data")
+        if isinstance(response_data, list) and response_data:
+            image_bytes = extract_image_bytes_from_item(response_data[0], session)
+            save_image_bytes(image_bytes, output_path, min_output_edge=min_output_edge)
+            return
+
+        if isinstance(response_data, dict) and response_data.get("id"):
+            image_bytes = poll_task(
+                str(response_data["id"]),
+                base_url,
+                {"Authorization": f"Bearer {api_key}"},
+                session,
+                run_dir,
+                max_poll_wait,
+                initial_poll_delay,
+                poll_interval,
+            )
+            save_image_bytes(image_bytes, output_path, min_output_edge=min_output_edge)
+            return
+
+        raise RuntimeError(f"unexpected response: {data}")
+    except BaseException as exc:
+        save_error_artifact(run_dir / "error.json", exc, {"output_file": str(output_path)})
+        raise
+
+
+if __name__ == "__main__":
+    main()

gpt_image2_generate_multi.py

@@ -0,0 +1,461 @@
+from __future__ import annotations
+
+import json
+import os
+import re
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import requests
+
+from gpt_image2_generate import (
+    DEFAULT_CONNECT_TIMEOUT,
+    DEFAULT_INITIAL_POLL_DELAY,
+    DEFAULT_MAX_POLL_WAIT,
+    DEFAULT_MODEL,
+    DEFAULT_PROMPT,
+    DEFAULT_PIXEL_SIZE,
+    DEFAULT_QUALITY,
+    DEFAULT_READ_TIMEOUT,
+    DEFAULT_POLL_CONNECT_TIMEOUT,
+    DEFAULT_POLL_INTERVAL,
+    DEFAULT_POLL_READ_TIMEOUT,
+    ImageInput,
+    build_request_session,
+    dump_json,
+    env_int,
+    extract_image_bytes_from_item,
+    image_source_to_input,
+    is_pixel_size,
+    load_env_file,
+    make_run_dir,
+    normalize_base_url,
+    normalize_size,
+    payload_to_form_fields,
+    resolve_api_size,
+    resolve_min_output_edge,
+    resolve_request_style,
+    save_error_artifact,
+    save_http_response,
+    save_image_bytes,
+)
+
+
+DEFAULT_OUTPUT_STEM = "gpt-image-2-multi"
+SUPPORTED_ASPECT_RATIOS = {
+    "1:1",
+    "16:9",
+    "9:16",
+    "4:3",
+    "3:4",
+    "3:2",
+    "2:3",
+    "5:4",
+    "4:5",
+    "2:1",
+    "1:2",
+    "21:9",
+    "9:21",
+}
+
+
+def _json_dumps(data: Any) -> str:
+    return json.dumps(data, ensure_ascii=False, indent=2)
+
+
+def split_reference_values(raw_value: str) -> list[str]:
+    value = raw_value.strip()
+    if not value:
+        return []
+
+    if value.startswith("["):
+        parsed = json.loads(value)
+        if not isinstance(parsed, list):
+            raise RuntimeError("REFERENCE_IMAGES 必须是 JSON 数组")
+        return [str(item).strip() for item in parsed if str(item).strip()]
+
+    return [part.strip() for part in value.split("|") if part.strip()]
+
+
+def collect_reference_sources() -> list[str]:
+    raw_list = os.getenv("REFERENCE_IMAGES", "").strip()
+    if raw_list:
+        refs = split_reference_values(raw_list)
+    else:
+        refs = []
+        for index in range(1, 17):
+            value = os.getenv(f"REFERENCE_IMAGE_{index}", "").strip()
+            if value:
+                refs.append(value)
+        if not refs:
+            single_candidates = [
+                os.getenv("REFERENCE_IMAGE", "").strip(),
+                os.getenv("REFERENCE_IMAGE_FILE", "").strip(),
+                os.getenv("REFERENCE_IMAGE_URL", "").strip(),
+                os.getenv("REFERENCE_IMAGE_DATA_URL", "").strip(),
+            ]
+            refs.extend([value for value in single_candidates if value])
+
+    deduped: list[str] = []
+    seen: set[str] = set()
+    for ref in refs:
+        if ref not in seen:
+            deduped.append(ref)
+            seen.add(ref)
+    return deduped
+
+
+def collect_mask_source() -> str | None:
+    candidates = {
+        "MASK_IMAGE": os.getenv("MASK_IMAGE", "").strip(),
+        "MASK_IMAGE_FILE": os.getenv("MASK_IMAGE_FILE", "").strip(),
+        "MASK_IMAGE_URL": os.getenv("MASK_IMAGE_URL", "").strip(),
+        "MASK_IMAGE_DATA_URL": os.getenv("MASK_IMAGE_DATA_URL", "").strip(),
+    }
+    provided = [value for value in candidates.values() if value]
+    if len(provided) > 1:
+        raise RuntimeError("MASK_IMAGE 相关变量只能设置一个")
+    return provided[0] if provided else None
+
+
+def resolve_output_stem() -> str:
+    raw_value = os.getenv("IMAGE_OUTPUT_STEM", DEFAULT_OUTPUT_STEM).strip()
+    candidate = re.sub(r"[^A-Za-z0-9._-]+", "_", Path(raw_value).name).strip("._")
+    return candidate or DEFAULT_OUTPUT_STEM
+
+
+def validate_requested_size(requested_size: str) -> None:
+    if not (is_pixel_size(requested_size) or requested_size in SUPPORTED_ASPECT_RATIOS):
+        raise RuntimeError(f"不支持的 IMAGE_SIZE: {requested_size}")
+
+
+def build_payload(base_url: str, reference_sources: list[str]) -> tuple[dict[str, Any], str, str, str, str, str | None, str | None]:
+    model = os.getenv("IMAGE_MODEL", DEFAULT_MODEL).strip()
+    prompt = os.getenv("IMAGE_PROMPT", DEFAULT_PROMPT).strip() or DEFAULT_PROMPT
+
+    requested_size = normalize_size(os.getenv("IMAGE_SIZE", DEFAULT_PIXEL_SIZE))
+    validate_requested_size(requested_size)
+    request_style = resolve_request_style(base_url)
+    api_size = resolve_api_size(requested_size, request_style)
+    quality = os.getenv("IMAGE_QUALITY", DEFAULT_QUALITY).strip()
+    resolution = os.getenv("IMAGE_RESOLUTION", "").strip()
+    background = os.getenv("IMAGE_BACKGROUND", "").strip() or None
+    input_fidelity = os.getenv("IMAGE_INPUT_FIDELITY", "").strip() or None
+    output_format = os.getenv("IMAGE_OUTPUT_FORMAT", "").strip() or None
+
+    n = env_int("IMAGE_N", max(1, len(reference_sources)))
+    if n < 1:
+        raise RuntimeError("IMAGE_N 必须大于等于 1")
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "prompt": prompt,
+        "n": n,
+        "size": api_size,
+        "quality": quality,
+    }
+    if not (request_style == "openai" and model.startswith("gpt-image")):
+        payload["response_format"] = "b64_json"
+    if not is_pixel_size(requested_size) and request_style != "openai":
+        payload["resolution"] = resolution or "4K"
+    if background:
+        payload["background"] = background
+    if input_fidelity:
+        payload["input_fidelity"] = input_fidelity
+    if output_format:
+        payload["output_format"] = output_format
+
+    mask_source = collect_mask_source()
+    if mask_source and not reference_sources:
+        raise RuntimeError("MASK_IMAGE 只能和参考图一起使用")
+
+    request_mode = "edit" if reference_sources else "generate"
+    return payload, model, requested_size, request_mode, mask_source, background, input_fidelity
+
+
+def build_edit_files(
+    request_style: str,
+    reference_inputs: list[ImageInput],
+    mask_input: ImageInput | None,
+) -> list[tuple[str, tuple[str, bytes, str]]]:
+    field_name = "image" if request_style == "openai" and len(reference_inputs) == 1 else "image[]"
+    files: list[tuple[str, tuple[str, bytes, str]]] = [
+        (field_name, (ref.filename, ref.data, ref.mime_type))
+        for ref in reference_inputs
+    ]
+    if mask_input is not None:
+        files.append(("mask", (mask_input.filename, mask_input.data, mask_input.mime_type)))
+    return files
+
+
+def extract_task_id(data: Any) -> str | None:
+    if not isinstance(data, dict):
+        return None
+    for key in ("task_id", "id"):
+        value = data.get(key)
+        if isinstance(value, str) and value:
+            return value
+    nested = data.get("data")
+    if isinstance(nested, dict):
+        for key in ("task_id", "id"):
+            value = nested.get(key)
+            if isinstance(value, str) and value:
+                return value
+    return None
+
+
+def get_task_error(task: dict[str, Any]) -> str:
+    nested = task.get("data")
+    error = nested.get("error") if isinstance(nested, dict) else task.get("error")
+    if isinstance(error, dict):
+        return str(error.get("message") or error)
+    if error:
+        return str(error)
+    return "task failed"
+
+
+def extract_image_items(data: Any) -> list[dict[str, Any]]:
+    if isinstance(data, list):
+        return [item for item in data if isinstance(item, dict)]
+
+    if isinstance(data, dict):
+        for key in ("images", "data"):
+            nested = data.get(key)
+            if isinstance(nested, list):
+                return [item for item in nested if isinstance(item, dict)]
+
+        result = data.get("result")
+        if isinstance(result, dict):
+            images = result.get("images")
+            if isinstance(images, list):
+                return [item for item in images if isinstance(item, dict)]
+
+    return []
+
+
+def poll_task_images(
+    task_id: str,
+    base_url: str,
+    headers: dict[str, str],
+    session: requests.Session,
+    run_dir: Path,
+    max_wait_seconds: int,
+    initial_poll_delay: int,
+    poll_interval: int,
+) -> list[dict[str, Any]]:
+    task_url = f"{base_url}/v1/tasks/{task_id}"
+    deadline = time.monotonic() + max_wait_seconds
+    poll_index = 0
+
+    time.sleep(initial_poll_delay)
+    while True:
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"任务轮询超时: task_id={task_id}")
+
+        poll_index += 1
+        task_resp = session.get(
+            task_url,
+            headers=headers,
+            timeout=(DEFAULT_POLL_CONNECT_TIMEOUT, DEFAULT_POLL_READ_TIMEOUT),
+        )
+        task_record = save_http_response(run_dir / f"poll-{poll_index:03d}.json", task_resp)
+        task_resp.raise_for_status()
+        task = task_record.get("json")
+        if not isinstance(task, dict):
+            raise RuntimeError(f"任务响应不是 JSON 对象: {task_record}")
+
+        data = task.get("data")
+        if not isinstance(data, dict):
+            raise RuntimeError(f"任务响应格式异常: {task}")
+
+        status = data.get("status")
+        print(f"任务轮询 #{poll_index}: status={status}")
+        if status == "completed":
+            result = data.get("result", {})
+            images = result.get("images", [])
+            if not images:
+                raise RuntimeError(f"任务已完成，但没有返回图片: {task}")
+            return [item for item in images if isinstance(item, dict)]
+
+        if status == "failed":
+            raise RuntimeError(get_task_error(task))
+
+        time.sleep(poll_interval)
+
+
+def save_output_images(
+    items: list[dict[str, Any]],
+    session: requests.Session,
+    run_dir: Path,
+    stem: str,
+    min_output_edge: int | None,
+) -> list[str]:
+    paths: list[str] = []
+    for index, item in enumerate(items, start=1):
+        image_bytes = extract_image_bytes_from_item(item, session)
+        output_path = run_dir / f"{stem}-{index}.png"
+        save_image_bytes(image_bytes, output_path, min_output_edge=min_output_edge)
+        paths.append(str(output_path))
+    return paths
+
+
+def main() -> None:
+    load_env_file()
+
+    api_key = os.getenv("OPENAI_API_KEY") or os.getenv("API_KEY")
+    base_url = normalize_base_url(
+        os.getenv("OPENAI_BASE_URL") or os.getenv("BASE_URL") or "https://api.toskaxy.xyz"
+    )
+    if not api_key:
+        raise RuntimeError("未找到 API Key，请先在 .env 中设置 OPENAI_API_KEY")
+
+    reference_sources = collect_reference_sources()
+    payload, model, requested_size, request_mode, mask_source, background, input_fidelity = build_payload(
+        base_url,
+        reference_sources,
+    )
+    run_dir = make_run_dir()
+    session = build_request_session(base_url)
+    request_style = resolve_request_style(base_url)
+    request_timeout = env_int("IMAGE_REQUEST_TIMEOUT", DEFAULT_READ_TIMEOUT)
+    max_poll_wait = env_int("IMAGE_MAX_POLL_WAIT", DEFAULT_MAX_POLL_WAIT)
+    initial_poll_delay = env_int("IMAGE_INITIAL_POLL_DELAY", DEFAULT_INITIAL_POLL_DELAY)
+    poll_interval = env_int("IMAGE_POLL_INTERVAL", DEFAULT_POLL_INTERVAL)
+    endpoint_path = "/v1/images/edits" if request_mode == "edit" else "/v1/images/generations"
+    min_output_edge = resolve_min_output_edge(requested_size, payload.get("resolution"), request_style)
+    prompt = payload["prompt"]
+    stem = f"{resolve_output_stem()}-{datetime.now().strftime('%Y%m%d-%H%M%S-%f')}"
+
+    print(
+        "开始生成图片:",
+        f"mode={request_mode}",
+        f"request_style={request_style}",
+        f"model={model}",
+        f"refs={len(reference_sources)}",
+        f"n={payload['n']}",
+        f"size={requested_size}",
+        f"resolution={payload.get('resolution') or 'none'}",
+        f"request_timeout={request_timeout}s",
+        f"max_poll_wait={max_poll_wait}s",
+        f"initial_poll_delay={initial_poll_delay}s",
+        f"poll_interval={poll_interval}s",
+        f"endpoint={base_url}{endpoint_path}",
+        f"responses={run_dir}",
+    )
+
+    dump_json(
+        run_dir / "request.json",
+        {
+            "base_url": base_url,
+            "endpoint_path": endpoint_path,
+            "request_mode": request_mode,
+            "request_style": request_style,
+            "model": model,
+            "prompt": prompt,
+            "size": requested_size,
+            "resolution": payload.get("resolution"),
+            "n": payload["n"],
+            "reference_sources": reference_sources,
+            "mask_source": mask_source,
+            "background": background,
+            "input_fidelity": input_fidelity,
+            "min_output_edge": min_output_edge,
+            "output_dir": str(run_dir),
+            "request_timeout_seconds": request_timeout,
+            "max_poll_wait_seconds": max_poll_wait,
+            "initial_poll_delay_seconds": initial_poll_delay,
+            "poll_interval_seconds": poll_interval,
+            "payload": payload,
+        },
+    )
+
+    try:
+        if request_mode == "edit":
+            reference_inputs = [
+                image_source_to_input(source, f"reference-{index}", session)
+                for index, source in enumerate(reference_sources, start=1)
+            ]
+            mask_input = image_source_to_input(mask_source, "mask-image", session) if mask_source else None
+            resp = session.post(
+                f"{base_url}{endpoint_path}",
+                data=payload_to_form_fields(payload),
+                files=build_edit_files(request_style, reference_inputs, mask_input),
+                headers={"Authorization": f"Bearer {api_key}"},
+                timeout=(DEFAULT_CONNECT_TIMEOUT, request_timeout),
+            )
+        else:
+            resp = session.post(
+                f"{base_url}{endpoint_path}",
+                json=payload,
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json",
+                },
+                timeout=(DEFAULT_CONNECT_TIMEOUT, request_timeout),
+            )
+
+        post_record = save_http_response(run_dir / "post-response.json", resp)
+        resp.raise_for_status()
+        data = post_record.get("json")
+        if not isinstance(data, dict):
+            raise RuntimeError(f"post response is not a JSON object: {post_record}")
+
+        response_data = data.get("data")
+        items = extract_image_items(response_data)
+        if items:
+            image_paths = save_output_images(items, session, run_dir, stem, min_output_edge)
+            dump_json(
+                run_dir / "result.json",
+                {
+                    "images": image_paths,
+                    "response_path": str(run_dir / "post-response.json"),
+                    "request_mode": request_mode,
+                    "request_style": request_style,
+                    "model": model,
+                    "size": requested_size,
+                    "n": payload["n"],
+                    "reference_count": len(reference_sources),
+                },
+            )
+            print(_json_dumps({"images": image_paths, "output_dir": str(run_dir)}))
+            return
+
+        task_id = extract_task_id(response_data)
+        if task_id:
+            items = poll_task_images(
+                task_id,
+                base_url,
+                {"Authorization": f"Bearer {api_key}"},
+                session,
+                run_dir,
+                max_poll_wait,
+                initial_poll_delay,
+                poll_interval,
+            )
+            image_paths = save_output_images(items, session, run_dir, stem, min_output_edge)
+            dump_json(
+                run_dir / "result.json",
+                {
+                    "images": image_paths,
+                    "response_path": str(run_dir / "post-response.json"),
+                    "request_mode": request_mode,
+                    "request_style": request_style,
+                    "model": model,
+                    "size": requested_size,
+                    "n": payload["n"],
+                    "reference_count": len(reference_sources),
+                },
+            )
+            print(_json_dumps({"images": image_paths, "output_dir": str(run_dir)}))
+            return
+
+        raise RuntimeError(f"unexpected response: {data}")
+    except BaseException as exc:
+        save_error_artifact(run_dir / "error.json", exc, {"output_dir": str(run_dir)})
+        raise
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,2 @@
+Pillow>=10.0.0
+requests>=2.31.0

scripts/run_aoixx_direct_test.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT_DIR"
+
+if [[ ! -f .env ]]; then
+  echo ".env not found" >&2
+  exit 1
+fi
+
+while IFS= read -r line; do
+  [[ -n "$line" ]] && export "$line"
+done < <(
+  awk '
+    $0 == "# https://api.aoixx.com" { in_block = 1; next }
+    in_block && /^#/ { exit }
+    in_block && /^OPENAI_API_KEY=/ { print; next }
+  ' .env
+)
+
+unset http_proxy https_proxy all_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY
+
+if [[ -z "${OPENAI_API_KEY:-}" ]]; then
+  echo "aoixx OPENAI_API_KEY not found in .env" >&2
+  exit 1
+fi
+
+export OPENAI_BASE_URL="https://api.aoixx.com"
+export IMAGE_REQUEST_STYLE="${IMAGE_REQUEST_STYLE:-apicodex}"
+export IMAGE_SIZE="${IMAGE_SIZE:-3840x2160}"
+export IMAGE_MODEL="${IMAGE_MODEL:-gpt-image-2}"
+export IMAGE_REQUEST_TIMEOUT="${IMAGE_REQUEST_TIMEOUT:-900}"
+export IMAGE_MAX_POLL_WAIT="${IMAGE_MAX_POLL_WAIT:-1800}"
+export IMAGE_INITIAL_POLL_DELAY="${IMAGE_INITIAL_POLL_DELAY:-10}"
+export IMAGE_POLL_INTERVAL="${IMAGE_POLL_INTERVAL:-5}"
+
+python3 gpt_image2_generate.py "$@"

scripts/run_figma_direct_test.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT_DIR"
+
+declare -A PRESERVED_ENV=()
+for name in \
+  OPENAI_API_KEY \
+  OPENAI_BASE_URL \
+  IMAGE_REQUEST_STYLE \
+  IMAGE_SIZE \
+  IMAGE_RESOLUTION \
+  IMAGE_PROMPT \
+  IMAGE_MODEL \
+  IMAGE_QUALITY \
+  IMAGE_N \
+  OUTPUT_FILE \
+  REFERENCE_IMAGE \
+  REFERENCE_IMAGE_FILE \
+  REFERENCE_IMAGE_URL \
+  REFERENCE_IMAGE_DATA_URL \
+  MASK_IMAGE \
+  MASK_IMAGE_FILE \
+  MASK_IMAGE_URL \
+  MASK_IMAGE_DATA_URL \
+  IMAGE_REQUEST_TIMEOUT \
+  IMAGE_MAX_POLL_WAIT \
+  IMAGE_INITIAL_POLL_DELAY \
+  IMAGE_POLL_INTERVAL
+do
+  if [[ -v "$name" ]]; then
+    PRESERVED_ENV["$name"]="${!name}"
+  fi
+done
+
+if [[ -f .env ]]; then
+  set -a
+  source .env
+  set +a
+fi
+
+for name in "${!PRESERVED_ENV[@]}"; do
+  export "$name=${PRESERVED_ENV[$name]}"
+done
+
+unset http_proxy https_proxy all_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY
+
+export IMAGE_REQUEST_STYLE="${IMAGE_REQUEST_STYLE:-apicodex}"
+export IMAGE_SIZE="${IMAGE_SIZE:-3840x2160}"
+export IMAGE_REQUEST_TIMEOUT="${IMAGE_REQUEST_TIMEOUT:-900}"
+export IMAGE_MAX_POLL_WAIT="${IMAGE_MAX_POLL_WAIT:-1800}"
+export IMAGE_INITIAL_POLL_DELAY="${IMAGE_INITIAL_POLL_DELAY:-10}"
+export IMAGE_POLL_INTERVAL="${IMAGE_POLL_INTERVAL:-5}"
+export IMAGE_PROMPT="${IMAGE_PROMPT:-A realistic Figma design board screenshot for a digital transformation system, with a left layers panel, center canvas of nested frames and components, and right properties panel. Enterprise product design, Chinese labels, visible grid, alignment guides, annotations, components, variants, spacing notes, dashboard cards, workflow timeline, and system modules. Clean blue-gray palette, crisp typography, premium product design presentation, highly organized layered structure, realistic Figma workspace aesthetic.}"
+
+python3 gpt_image2_generate.py "$@"

scripts/run_openai_gpt_image2_edit.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT_DIR"
+
+declare -A PRESERVED_ENV=()
+for name in \
+  OPENAI_API_KEY \
+  OPENAI_BASE_URL \
+  IMAGE_REQUEST_STYLE \
+  IMAGE_SIZE \
+  IMAGE_RESOLUTION \
+  IMAGE_PROMPT \
+  IMAGE_MODEL \
+  IMAGE_QUALITY \
+  IMAGE_N \
+  OUTPUT_FILE \
+  REFERENCE_IMAGE \
+  REFERENCE_IMAGE_FILE \
+  REFERENCE_IMAGE_URL \
+  REFERENCE_IMAGE_DATA_URL \
+  MASK_IMAGE \
+  MASK_IMAGE_FILE \
+  MASK_IMAGE_URL \
+  MASK_IMAGE_DATA_URL \
+  IMAGE_REQUEST_TIMEOUT \
+  IMAGE_MAX_POLL_WAIT \
+  IMAGE_INITIAL_POLL_DELAY \
+  IMAGE_POLL_INTERVAL
+do
+  if [[ -v "$name" ]]; then
+    PRESERVED_ENV["$name"]="${!name}"
+  fi
+done
+
+if [[ -f .env ]]; then
+  set -a
+  source .env
+  set +a
+fi
+
+for name in "${!PRESERVED_ENV[@]}"; do
+  export "$name=${PRESERVED_ENV[$name]}"
+done
+
+unset http_proxy https_proxy all_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY
+
+export IMAGE_REQUEST_STYLE="${IMAGE_REQUEST_STYLE:-openai}"
+export IMAGE_MODEL="${IMAGE_MODEL:-gpt-image-2}"
+export IMAGE_SIZE="${IMAGE_SIZE:-1024x1024}"
+export IMAGE_REQUEST_TIMEOUT="${IMAGE_REQUEST_TIMEOUT:-900}"
+export IMAGE_MAX_POLL_WAIT="${IMAGE_MAX_POLL_WAIT:-1800}"
+export IMAGE_INITIAL_POLL_DELAY="${IMAGE_INITIAL_POLL_DELAY:-10}"
+export IMAGE_POLL_INTERVAL="${IMAGE_POLL_INTERVAL:-5}"
+
+python3 gpt_image2_generate.py "$@"

scripts/run_openai_gpt_image2_multi.sh

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT_DIR"
+
+declare -A PRESERVED_ENV=()
+for name in \
+  OPENAI_API_KEY \
+  OPENAI_BASE_URL \
+  IMAGE_REQUEST_STYLE \
+  IMAGE_SIZE \
+  IMAGE_RESOLUTION \
+  IMAGE_PROMPT \
+  IMAGE_MODEL \
+  IMAGE_QUALITY \
+  IMAGE_N \
+  IMAGE_OUTPUT_STEM \
+  IMAGE_RESPONSE_DIR \
+  IMAGE_BACKGROUND \
+  IMAGE_INPUT_FIDELITY \
+  IMAGE_OUTPUT_FORMAT \
+  REFERENCE_IMAGES \
+  REFERENCE_IMAGE_LIST \
+  REFERENCE_IMAGE \
+  REFERENCE_IMAGE_FILE \
+  REFERENCE_IMAGE_URL \
+  REFERENCE_IMAGE_DATA_URL \
+  MASK_IMAGE \
+  MASK_IMAGE_FILE \
+  MASK_IMAGE_URL \
+  MASK_IMAGE_DATA_URL \
+  IMAGE_REQUEST_TIMEOUT \
+  IMAGE_MAX_POLL_WAIT \
+  IMAGE_INITIAL_POLL_DELAY \
+  IMAGE_POLL_INTERVAL
+do
+  if [[ -v "$name" ]]; then
+    PRESERVED_ENV["$name"]="${!name}"
+  fi
+done
+
+for i in $(seq 1 16); do
+  name="REFERENCE_IMAGE_${i}"
+  if [[ -v "$name" ]]; then
+    PRESERVED_ENV["$name"]="${!name}"
+  fi
+done
+
+if [[ -f .env ]]; then
+  set -a
+  source .env
+  set +a
+fi
+
+for name in "${!PRESERVED_ENV[@]}"; do
+  export "$name=${PRESERVED_ENV[$name]}"
+done
+
+unset http_proxy https_proxy all_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY
+
+export IMAGE_REQUEST_STYLE="${IMAGE_REQUEST_STYLE:-openai}"
+export IMAGE_MODEL="${IMAGE_MODEL:-gpt-image-2}"
+export IMAGE_SIZE="${IMAGE_SIZE:-1024x1024}"
+export IMAGE_RESPONSE_DIR="${IMAGE_RESPONSE_DIR:-outputs/image-multi-runs}"
+export IMAGE_REQUEST_TIMEOUT="${IMAGE_REQUEST_TIMEOUT:-900}"
+export IMAGE_MAX_POLL_WAIT="${IMAGE_MAX_POLL_WAIT:-1800}"
+export IMAGE_INITIAL_POLL_DELAY="${IMAGE_INITIAL_POLL_DELAY:-10}"
+export IMAGE_POLL_INTERVAL="${IMAGE_POLL_INTERVAL:-5}"
+
+python3 gpt_image2_generate_multi.py "$@"

tests/test_scripts.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import patch
+
+import gpt_image2_generate as image_demo
+import gpt_image2_generate_multi as image_multi_demo
+
+
+class ImageDemoHelpersTest(unittest.TestCase):
+    def test_openai_ratio_size_mapping(self) -> None:
+        self.assertEqual(image_demo.resolve_api_size("16:9", "openai"), "1536x864")
+        self.assertEqual(image_demo.resolve_api_size("3840x2160", "openai"), "3840x2160")
+
+    def test_min_output_edge(self) -> None:
+        self.assertEqual(image_demo.resolve_min_output_edge("4096x2304", None, "apicodex"), 4096)
+        self.assertIsNone(image_demo.resolve_min_output_edge("1024x1024", None, "openai"))
+
+
+class ImageMultiDemoHelpersTest(unittest.TestCase):
+    def test_collect_reference_sources_from_json_array(self) -> None:
+        with patch.dict(os.environ, {"REFERENCE_IMAGES": '["./a.png", "./b.png", "./a.png"]'}, clear=True):
+            self.assertEqual(image_multi_demo.collect_reference_sources(), ["./a.png", "./b.png"])
+
+    def test_build_edit_files_field_names(self) -> None:
+        ref1 = image_demo.ImageInput("a.png", "image/png", b"a")
+        ref2 = image_demo.ImageInput("b.png", "image/png", b"b")
+
+        openai_single = image_multi_demo.build_edit_files("openai", [ref1], None)
+        openai_multi = image_multi_demo.build_edit_files("openai", [ref1, ref2], None)
+        apicodex_multi = image_multi_demo.build_edit_files("apicodex", [ref1, ref2], None)
+
+        self.assertEqual([name for name, _ in openai_single], ["image"])
+        self.assertEqual([name for name, _ in openai_multi], ["image[]", "image[]"])
+        self.assertEqual([name for name, _ in apicodex_multi], ["image[]", "image[]"])
+
+    def test_output_stem_env_is_sanitized(self) -> None:
+        with patch.dict(os.environ, {"IMAGE_OUTPUT_STEM": "../bad stem!"}, clear=True):
+            self.assertEqual(image_multi_demo.resolve_output_stem(), "bad_stem")
+
+    def test_default_n_matches_reference_count(self) -> None:
+        env = {
+            "IMAGE_PROMPT": "hello",
+            "IMAGE_SIZE": "1:1",
+        }
+        with patch.dict(os.environ, env, clear=True):
+            payload, model, size, request_mode, mask_source, background, input_fidelity = image_multi_demo.build_payload(
+                "http://localhost",
+                ["./a.png", "./b.png"],
+            )
+
+        self.assertEqual(payload["n"], 2)
+        self.assertEqual(request_mode, "edit")
+        self.assertIsNone(mask_source)
+        self.assertIsNone(background)
+        self.assertIsNone(input_fidelity)
+        self.assertEqual(model, image_demo.DEFAULT_MODEL)
+        self.assertEqual(size, "1:1")
+
+
+if __name__ == "__main__":
+    unittest.main()