48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
"""GPT Image 2.0 — OpenAI Images API.
|
|
|
|
POST https://api.openai.com/v1/images/generations
|
|
body {model:"gpt-image-1", prompt, size, n:1} → data[0].b64_json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
|
|
import requests
|
|
|
|
from nas_client import webhook_update_task
|
|
from providers._media import save_b64_png
|
|
|
|
logger = logging.getLogger(__name__)
|
|
OPENAI_URL = "https://api.openai.com/v1/images/generations"
|
|
DEFAULT_MODEL = "gpt-image-1"
|
|
|
|
|
|
def run_gpt_image_generation(task_id: str, params: dict) -> None:
|
|
try:
|
|
if not os.getenv("OPENAI_API_KEY"):
|
|
webhook_update_task(task_id, "failed", 0, "", error="OPENAI_API_KEY 미설정 (Windows .env)")
|
|
return
|
|
webhook_update_task(task_id, "processing", 10, "GPT Image 호출 중...")
|
|
body = {
|
|
"model": params.get("model") or DEFAULT_MODEL,
|
|
"prompt": params["prompt"],
|
|
"size": params.get("size") or "1024x1024",
|
|
"n": 1,
|
|
}
|
|
resp = requests.post(
|
|
OPENAI_URL,
|
|
headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}", "Content-Type": "application/json"},
|
|
json=body,
|
|
timeout=120,
|
|
)
|
|
if resp.status_code != 200:
|
|
webhook_update_task(task_id, "failed", 0, "", error=f"OpenAI {resp.status_code}: {resp.text[:200]}")
|
|
return
|
|
b64 = resp.json()["data"][0]["b64_json"]
|
|
url = save_b64_png(task_id, b64)
|
|
webhook_update_task(task_id, "succeeded", 100, "완료", image_url=url)
|
|
except Exception as e:
|
|
logger.exception("gpt_image task=%s 실패", task_id)
|
|
webhook_update_task(task_id, "failed", 0, "", error=str(e))
|