Images

Edytowanie · multipart/form-data

: "${AILOOT_IMAGE_EDIT_MODEL:?Select a model whose actual card or GET /v1/models method supports /v1/images/edits}"
curl --fail-with-body --max-time 300 \
  -D image-edit-headers.txt -o image-edit.json \
  https://api.ailoot.ai/v1/images/edits \
  -H "Authorization: Bearer $AILOOT_API_KEY" \
  -F "model=$AILOOT_IMAGE_EDIT_MODEL" \
  -F 'prompt=Keep the subject and replace the background with a sunset' \
  -F 'image=@input.png'

input.png → image-edit.json → data[] → url | b64_json

import base64
import json
import mimetypes
import os
from pathlib import Path
import time
from urllib.parse import urljoin, urlsplit
from urllib.request import Request, urlopen

origin = "https://api.ailoot.ai"
out = Path("image-edit-results")
out.mkdir(exist_ok=True)

def save_all(data):
    rows = data.get("data", [])
    if not rows:
        raise RuntimeError("No completed image data; keep the original response")
    for index, row in enumerate(rows, 1):
        if row.get("b64_json"):
            content = base64.b64decode(row["b64_json"], validate=True)
            suffix = ".png" if content.startswith(b"\x89PNG") else ".bin"
        elif row.get("url"):
            url = row["url"]
            if urlsplit(url).scheme != "https":
                raise ValueError("Expected the exact returned HTTPS image URL")
            # Signed result URLs are downloaded without sending the API key to another host.
            with urlopen(url, timeout=300) as response:
                content = response.read()
                suffix = mimetypes.guess_extension(response.headers.get_content_type()) or ".bin"
        else:
            raise RuntimeError("Image has neither url nor b64_json")
        (out / ("image-%03d" % index + suffix)).write_bytes(content)


with open("image-edit.json", encoding="utf-8") as response:
    save_all(json.load(response))

Images · async

SDK OpenAI i Anthropic zwykle automatycznie ponawiają żądania po błędach sieciowych i 5xx. Przy generowaniu wyłącz tę funkcję (max_retries=0) i sprawdzaj X-Should-Retry przed kolejną próbą. Utrata odpowiedzi nie oznacza, że żądanie nie zostało wykonane.

import base64
import json
import mimetypes
import os
from pathlib import Path
import time
from urllib.parse import urljoin, urlsplit
from urllib.request import Request, urlopen

origin = "https://api.ailoot.ai"
key = os.environ["AILOOT_API_KEY"]
out = Path("image-results")
out.mkdir(exist_ok=True)

def api(url, payload=None):
    url = urljoin(origin, url)
    if urlsplit(url).scheme != "https" or urlsplit(url).netloc != "api.ailoot.ai":
        raise ValueError("Unexpected API origin")
    request = Request(url, data=None if payload is None else json.dumps(payload).encode(),
                      headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"})
    with urlopen(request, timeout=300) as response:
        headers = dict(response.headers)
        data = json.load(response)
    (out / "response-headers.json").write_text(json.dumps(headers, indent=2))
    return data, headers

def save_all(data):
    rows = data.get("data", [])
    if not rows:
        raise RuntimeError("No completed image data; keep the original response")
    for index, row in enumerate(rows, 1):
        if row.get("b64_json"):
            content = base64.b64decode(row["b64_json"], validate=True)
            suffix = ".png" if content.startswith(b"\x89PNG") else ".bin"
        elif row.get("url"):
            url = row["url"]
            if urlsplit(url).scheme != "https":
                raise ValueError("Expected the exact returned HTTPS image URL")
            # Signed result URLs are downloaded without sending the API key to another host.
            with urlopen(url, timeout=300) as response:
                content = response.read()
                suffix = mimetypes.guess_extension(response.headers.get_content_type()) or ".bin"
        else:
            raise RuntimeError("Image has neither url nor b64_json")
        (out / ("image-%03d" % index + suffix)).write_bytes(content)

# One POST. If its result is unknown, preserve it and check history; do not rerun creation.
task, headers = api("/v1/images/generations/async", {
    "model": "gpt-image-2.5", "prompt": "Watercolor lighthouse at dawn", "size": "1024x1024"})
(out / "created-task.json").write_text(json.dumps(task, indent=2))
task_id = task.get("task_id") or task.get("id")
poll_url = task.get("poll_url") or headers.get("Location") or headers.get("location")
if not task_id or not poll_url:
    raise RuntimeError("Keep the create response and Request ID; do not submit again")

for poll in range(60):
    task, headers = api(poll_url)  # Read the original task with the same key.
    (out / "task-status.json").write_text(json.dumps(task, indent=2))
    if task.get("status") == "completed":
        result = task["result"]
        (out / "image-response.json").write_text(json.dumps(result, indent=2))
        save_all(result)
        break
    if task.get("status") == "failed":
        raise RuntimeError(json.dumps(task.get("error")))
    if task.get("status") != "processing" or task.get("result_pending_check"):
        raise RuntimeError("Keep the original task ID and check it later; no new POST")
    time.sleep(min(60, max(1, int(headers.get("Retry-After", headers.get("retry-after", "3"))))))
else:
    raise TimeoutError("Polling stopped; resume GET reads using created-task.json, never repeat POST")