#!/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()
