Обзор открытого API ZroImg

Введение

Открытый API ZroImg позволяет вызывать модели генерации изображений и видео платформы из ваших программ. API построен в стиле REST, запросы и ответы передаются в формате JSON, а оплата идёт баллами с вашего аккаунта по тем же правилам, что и на сайте (см. Баллы и оплата).

Base URL

Все методы используют следующий адрес в качестве префикса:

https://api.zroimg.com/open/v1

Авторизация

API использует авторизацию по API-ключу. Войдите в аккаунт и создайте ключ в разделе Настройки аккаунта → API-ключи. Ключ начинается с zk- и полностью показывается только один раз — при создании, поэтому сразу сохраните его (сервер хранит только хеш, поэтому потерянный ключ можно лишь создать заново). У одного аккаунта может быть до 10 действующих ключей одновременно (не отозванных и не истёкших); для ключа можно задать срок действия, а также отозвать его в любой момент — отзыв вступает в силу сразу.

Передавайте ключ в заголовке Authorization каждого запроса (также можно использовать заголовок X-API-Key: zk-...):

Authorization: Bearer zk-xxxxxxxxxxxxxxxxxxxxxxxx

Если ключ не существует, отозван, истёк или аккаунт заблокирован, возвращается 401 invalid_api_key. API-ключ работает только с методами /open/v1.

Проверить, что ключ работает, можно с помощью метода баланса:

curl https://api.zroimg.com/open/v1/balance \
  -H "Authorization: Bearer zk-xxxxxxxxxxxxxxxxxxxxxxxx"

Асинхронные задачи

Генерация изображений и видео занимает время, поэтому API работает по асинхронной схеме «отправка + опрос»:

1. 1

   Отправьте задачу через POST /images/generations или POST /videos/generations. Сервер проверяет параметры, списывает баллы и сразу возвращает 202 Accepted; в ответе есть id задачи и списанная сумма cost_points.

2. 2

   Каждые 3–5 секунд запрашивайте статус задачи через GET /tasks/{id}. Изображения обычно готовы за 30–90 секунд, видео — за несколько минут.

3. 3

   Когда статус станет succeeded, возьмите адреса результатов из data и скачайте их; при статусе failed причина указана в error.message, а баллы уже возвращены автоматически.

Статусы задачи:

Идемпотентная отправка

Если повторить запрос на отправку после сетевого тайм-аута, может быть создано две задачи с двойным списанием баллов. Чтобы этого избежать, передавайте при отправке заголовок Idempotency-Key (рекомендуется UUID):

Idempotency-Key: 6f1c2a9e-3d4b-4c8e-9a51-2b7d0e8f1a33

• Повторная отправка с тем же API-ключом и тем же Idempotency-Key в течение 24 часов сразу вернёт задачу, созданную в первый раз (тоже с кодом 202 и заголовком ответа Idempotent-Replayed: true), без повторного списания.
• Если первый запрос ещё обрабатывается, повторная отправка вернёт 409 idempotency_conflict — подождите немного и повторите с тем же значением.
• Если первый запрос завершился ошибкой (задача не создана, баллы не списаны), можно отправить запрос заново с тем же значением.
• Для каждой новой задачи используйте новый Idempotency-Key (не более 128 символов); повторно используйте ключ только при «повторе той же отправки».

Лимиты запросов

Каждый ответ содержит заголовки X-RateLimit-Limit и X-RateLimit-Remaining — общий лимит в минуту и количество оставшихся запросов.

При превышении лимита возвращается HTTP 429 с кодом ошибки rate_limited или concurrency_limit и заголовком ответа Retry-After (в секундах) — подождите указанное время и повторите запрос. concurrency_limit означает, что выполняется слишком много задач, — можно дождаться завершения текущих, прежде чем отправлять новые.

Формат ошибок

При ошибке API возвращает соответствующий HTTP-код и объект ошибки единого формата:

{
  "error": {
    "code": "insufficient_points",
    "message": "Недостаточно баллов: требуется 40 баллов, текущий баланс — 12",
    "type": "invalid_request_error"
  }
}

Определяйте тип ошибки по code; message предназначено только для отображения и диагностики (пока на китайском). Полный список — в разделе Коды ошибок.

Формат времени

Поля времени в ответах открытого API (created_at, finished_at, expires_at) — это Unix-время в секундах.

Полный пример

Программа ниже отправляет задачу генерации изображения, опрашивает её до завершения, скачивает все результаты и выводит оставшийся баланс баллов. Она обрабатывает повторы при 429 и неудавшиеся задачи, поэтому её можно взять за основу интеграции. Чтобы запустить пример, замените API_KEY своим ключом.

"""Generate an image with the ZroImg API and save the results.

pip install requests
"""
import time
import uuid

import requests

API_KEY = "zk-xxxxxxxxxxxxxxxxxxxxxxxx"
BASE_URL = "https://api.zroimg.com/open/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


def call(method, path, **kwargs):
    """Send a request; wait and retry when the API answers 429."""
    headers = {**HEADERS, **kwargs.pop("headers", {})}
    for _ in range(5):
        response = requests.request(method, BASE_URL + path, headers=headers, timeout=30, **kwargs)
        if response.status_code != 429:
            break
        time.sleep(int(response.headers.get("Retry-After", "5")))
    if response.status_code >= 400:
        error = response.json()["error"]
        raise RuntimeError(f"{response.status_code} {error['code']}: {error['message']}")
    return response.json()


def submit(prompt):
    # Reuse the same Idempotency-Key if you retry this call: you are charged only once.
    idempotency_key = str(uuid.uuid4())
    return call(
        "POST",
        "/images/generations",
        headers={"Idempotency-Key": idempotency_key},
        json={"model": "gpt-image-2", "prompt": prompt, "ratio": "1:1", "quality": "medium", "n": 1},
    )


def wait(task_id, interval=4, timeout=900):
    deadline = time.time() + timeout
    while time.time() < deadline:
        task = call("GET", f"/tasks/{task_id}")
        if task["status"] == "succeeded":
            return task
        if task["status"] == "failed":
            # Points for failed tasks are refunded automatically.
            raise RuntimeError(f"task failed: {task['error']['message']}")
        progress = task.get("progress") or {}
        print(f"{task['status']} {progress.get('percent', '')}")
        time.sleep(interval)
    raise TimeoutError(f"task {task_id} is still running, check it again later")


if __name__ == "__main__":
    task = submit("An isometric cozy reading room, warm evening light, highly detailed")
    print(f"submitted {task['id']}, cost {task['cost_points']} points")

    result = wait(task["id"])
    for index, item in enumerate(result["data"]):
        # Result URLs are signed and expire after 24 hours: download them now.
        content = requests.get(item["url"], timeout=120).content
        filename = f"{task['id']}-{index}.png"
        with open(filename, "wb") as file:
            file.write(content)
        print("saved", filename)

    print("balance:", call("GET", "/balance")["points"])