#!/usr/bin/env python3
"""通过 API易 调用 gpt-image-2 系列（gpt-image-2 官转 / gpt-image-2-all / gpt-image-2-vip 官逆）生成 / 编辑图片。
都走 OpenAI Images API（/v1/images/generations + /v1/images/edits），用 --model 切换。需要：pip install openai"""
import argparse
import base64
import os
import sys
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI
# 一次调用最多并发出几张图（服务端 n 只出 1 张，这里用客户端并发模拟多张）
MAX_COUNT = 5
# 各模型能力门控：是否接受这些参数（不接受的一律不传，避免报错或超扣费）
MODEL_CAPS = {
"gpt-image-2": {"size": True, "quality": True, "output_format": True, "mask": True}, # 官转
"gpt-image-2-all": {"size": False, "quality": False, "output_format": False, "mask": False}, # 官逆·ChatGPT
"gpt-image-2-vip": {"size": True, "quality": False, "output_format": False, "mask": False}, # 官逆·Codex
}
def caps_of(model):
# 未知模型回落到官转能力集
return MODEL_CAPS.get(model, MODEL_CAPS["gpt-image-2"])
def load_api_key():
"""优先读环境变量；否则在脚本所在目录及其父目录找 .env。"""
key = os.environ.get("APIYI_API_KEY")
if key:
return key
here = os.path.dirname(os.path.abspath(__file__))
for d in (here, os.path.dirname(here)):
env_path = os.path.join(d, ".env")
if os.path.exists(env_path):
with open(env_path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if line.startswith("APIYI_API_KEY") and "=" in line:
return line.split("=", 1)[1].strip().strip('"').strip("'")
return None
def project_root():
"""从脚本位置向上找包含 .git 或 .claude 的目录，作为项目根目录；找不到则用当前工作目录。"""
d = os.path.dirname(os.path.abspath(__file__))
while True:
if os.path.isdir(os.path.join(d, ".git")) or os.path.isdir(os.path.join(d, ".claude")):
return d
parent = os.path.dirname(d)
if parent == d:
return os.getcwd()
d = parent
def resolve_paths(out, count):
"""决定输出路径列表。纯文件名 → 存到 <项目根>/gpt-image-output/；带目录则按给定路径。"""
if os.path.dirname(out):
base_path = os.path.abspath(out)
else:
out_dir = os.path.join(project_root(), "gpt-image-output")
os.makedirs(out_dir, exist_ok=True)
base_path = os.path.join(out_dir, out)
if count == 1:
return [base_path]
base, ext = os.path.splitext(base_path)
return [f"{base}-{i}{ext}" for i in range(1, count + 1)]
def decode_image(item):
"""统一取图字节：b64_json 可能是纯 base64 或带 data:image 前缀的 data URL（官逆模型）；也可能只给 url。"""
raw = getattr(item, "b64_json", None)
if raw:
if raw.startswith("data:"):
raw = raw.split(",", 1)[1] # 剥掉 data:image/png;base64, 前缀
return base64.b64decode(raw)
url = getattr(item, "url", None)
if url:
with urllib.request.urlopen(url, timeout=360) as r:
return r.read()
raise RuntimeError("响应里既没有 b64_json 也没有 url")
def one_image(client, model, args):
"""发一次请求，返回图片字节；失败抛异常（由 _safe 兜住）。一律不传 n（默认 1 张，多张靠客户端并发）。"""
cap = caps_of(model)
if args.image:
# 编辑 / 多图融合：每次重新 open 文件，避免线程间共享句柄
files = [open(p, "rb") for p in args.image]
try:
kwargs = dict(model=model, image=files if len(files) > 1 else files[0], prompt=args.prompt)
if cap["size"] and args.size:
kwargs["size"] = args.size
if cap["quality"] and args.quality:
kwargs["quality"] = args.quality
if cap["mask"] and args.mask:
kwargs["mask"] = open(args.mask, "rb")
resp = client.images.edit(**kwargs)
finally:
for fh in files:
fh.close()
else:
# 文生图
kwargs = dict(model=model, prompt=args.prompt)
if cap["size"] and args.size:
kwargs["size"] = args.size
if cap["quality"] and args.quality:
kwargs["quality"] = args.quality
if cap["output_format"] and args.format:
kwargs["output_format"] = args.format
resp = client.images.generate(**kwargs)
return decode_image(resp.data[0])
def main():
api_key = load_api_key()
if not api_key:
sys.exit("未找到 API Key：请在技能目录的 .env 写一行 APIYI_API_KEY=sk-xxx")
default_model = os.environ.get("APIYI_IMAGE_MODEL", "gpt-image-2")
# 同步阻塞调用，图片生成慢，超时给足 360s
client = OpenAI(api_key=api_key, base_url="https://api.apiyi.com/v1", timeout=360)
parser = argparse.ArgumentParser(description="gpt-image-2 系列出图")
parser.add_argument("prompt", help="提示词 / 编辑指令")
parser.add_argument("--model", default=default_model,
help="gpt-image-2(官转) / gpt-image-2-all(最快) / gpt-image-2-vip(可锁尺寸)")
parser.add_argument("-i", "--image", action="append", default=[],
help="输入图片路径（可重复，最多 16 张；传入即为编辑/融合模式）")
parser.add_argument("-o", "--out", default="output.png", help="输出文件名")
parser.add_argument("-n", "--count", type=int, default=1,
help=f"一次出几张，默认 1，最多 {MAX_COUNT}（客户端并发）")
parser.add_argument("--size", default="auto",
help="尺寸，如 1024x1024 / 2048x1152 / auto（gpt-image-2-all 不支持，写进 prompt）")
parser.add_argument("--quality", default="high",
help="画质 low / medium / high / auto（仅 gpt-image-2 官转生效）")
parser.add_argument("--format", default="png", help="输出格式 png / jpeg / webp（仅官转生效）")
parser.add_argument("--mask", help="掩码图（仅官转编辑，PNG 带 alpha，对第一张图生效）")
args = parser.parse_args()
count = args.count
if count < 1:
count = 1
if count > MAX_COUNT:
print(f"提示：一次最多 {MAX_COUNT} 张，已将 {args.count} 限制为 {MAX_COUNT}。", file=sys.stderr)
count = MAX_COUNT
paths = resolve_paths(args.out, count)
def task(path):
data = one_image(client, args.model, args)
with open(path, "wb") as f:
f.write(data)
return os.path.abspath(path)
failures = 0
with ThreadPoolExecutor(max_workers=count) as pool:
for path, result in zip(paths, pool.map(lambda p: _safe(task, p), paths)):
ok, value = result
if ok:
print(f"图片已保存至 {value}")
else:
failures += 1
print(f"第 {os.path.basename(path)} 张生成失败：{value}", file=sys.stderr)
if failures == count:
sys.exit("全部生成失败。")
def _safe(fn, arg):
try:
return True, fn(arg)
except Exception as e: # noqa: BLE001 — 单张失败不影响其它并发任务
return False, str(e)
if __name__ == "__main__":
main()
