Images

Szerkesztés · 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

Az OpenAI és Anthropic SDK-k hálózati hibák vagy 5xx válaszok esetén általában automatikusan újraküldik a kérést. Generálásnál kapcsolja ki ezt (max_retries=0) és ellenőrizze az X-Should-Retry értékét új próbálkozás előtt. Az elveszett válasz nem jelenti azt, hogy a kérés nem hajtódott végre.

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")