Разработчикам··4 мин чтения

Bigjpg API: пакетное увеличение картинок на Python

Готовый скрипт: загрузка файлов, создание задач на увеличение, опрос результатов и скачивание. Плюс параллельность и обработка ошибок, чтобы пакет из 500 изображений не разваливался на середине.

Bigjpg API: пакетное увеличение картинок на Python

Для десятка картинок хватает веб-интерфейса. Когда в папке пятьсот файлов, нужен API: четыре эндпоинта, никакого браузера, скрипт можно запускать сколько угодно раз. Ниже — та версия, которую мы бы дали коллеге: загрузка, создание, опрос, скачивание и обработка сбоев, которые встречаются на практике.

Что нужно знать заранее

  • API доступен только на платном аккаунте. Ключ лежит на странице API, когда вы вошли с активным тарифом.
  • Python 3 и установленный requests. Больше ничего.
  • Увеличение выполняется асинхронно на наших воркерах, поэтому скрипт сначала создаёт задачи, а потом опрашивает их. 500 картинок займут столько же, сколько самая медленная задача в вашей очереди, а не сумму по всем файлам.

Скрипт

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" для иллюстраций, "photo" для фотографий
NOISE = "1"            # -1 без шума, 0 низкий, 1 средний, 2 высокий, 3 максимальный
X2 = "2"               # 1 = 2x, 2 = 4x, 3 = 8x, 4 = 16x
POLL_SECONDS = 8
MAX_RETRY = 1

HEADERS = {"X-API-KEY": KEY}


def upload(path):
    """Get a presigned URL, PUT the file, return its public URL."""
    name = path.split("/")[-1]
    r = requests.post(
        API + "/api/img_upload",
        headers=HEADERS,
        data={"filename": name, "filetype": "image/jpeg"},
        timeout=60,
    )
    info = r.json()
    if info.get("status") != "ok":
        raise RuntimeError("upload failed: " + str(info))

    with open(path, "rb") as image:
        r = requests.put(
            info["url"],
            data=image,
            headers={"Content-Type": "image/jpeg"},
            timeout=300,
        )
    if r.status_code != 200:
        raise RuntimeError("upload failed with HTTP %s" % r.status_code)
    return info["fileurl"]


def create(image_url):
    """Create an enlargement task, return its task id."""
    payload = {
        "style": STYLE,
        "noise": NOISE,
        "x2": X2,
        "input": image_url,
    }
    r = requests.post(
        API + "/api/task/",
        headers=HEADERS,
        data=json.dumps(payload),
        timeout=60,
    )
    data = r.json()
    if "tid" not in data:
        raise RuntimeError("task failed: " + str(data))
    return data["tid"]


def poll(tids):
    """Query up to 100 task ids at once. Returns task id -> result dict."""
    r = requests.get(API + "/api/task/" + ",".join(tids), timeout=60)
    return r.json()


def download(url, out):
    """Download a finished image, checking the transfer really succeeded."""
    r = requests.get(url, timeout=300)
    if r.status_code != 200:
        raise RuntimeError("download failed with HTTP %s" % r.status_code)
    with open(out, "wb") as f:
        f.write(r.content)
    return out


def main(paths):
    pending = {}
    attempts = {}
    for path in paths:
        try:
            tid = create(upload(path))
        except Exception as e:
            print("skipped", path, e)
            continue
        pending[tid] = path
        attempts[tid] = 0
        print("queued", path, tid)

    while pending:
        for tid, info in poll(list(pending)[:100]).items():
            if info.get("status") == "success":
                out = os.path.splitext(pending[tid])[0] + "_bigjpg.jpg"
                try:
                    download(info["url"], out)
                except Exception as e:
                    print("download failed", pending[tid], e)
                    pending.pop(tid)
                    continue
                print("done", pending.pop(tid), "->", out)

            elif info.get("status") == "error":
                # Одна повторная попытка, потом файл считаем потерянным.
                attempts[tid] += 1
                if attempts[tid] <= MAX_RETRY:
                    requests.post(API + "/api/task/" + tid, headers=HEADERS, timeout=60)
                    print("retrying", pending[tid])
                else:
                    print("giving up on", pending.pop(tid))

        if pending:
            time.sleep(POLL_SECONDS)


if __name__ == "__main__":
    import sys
    main(sys.argv[1:])

Что делает каждый вызов

ВызовНазначениеЗаметки
POST /api/img_uploadВозвращает ссылку для загрузки по подписи и публичный URL картинкиЗагрузите файл через PUT с тем же Content-Type, что указали, и передайте fileurl как input задачи
POST /api/task/Создаёт задачу на увеличениеВозвращает tid, оценку minute и remaining_api_calls
GET /api/task/{ids}Статус одной задачи или несколькихid через запятую, до 100 за запрос
POST /api/task/{ids}Перезапускает упавшие задачиВ очередь возвращаются только задачи в статусе error

Параллельность и вежливость

  • Опрашивайте реже, чем кажется нужным. Увеличение не мгновенно: 5–10 секунд между запросами более чем достаточно, а 100 id за один вызов держат их число низким.
  • Ограничьте загрузки. Две-четыре параллельные загрузки с одной машины — разумный потолок. Нужна пропускная способность выше — запускайте больше экземпляров, а не больше потоков внутри одного.
  • Повторяйте с задержкой. Ошибка сети — это ещё не провал задачи. Повторите HTTP-вызов до трёх раз с растущей паузой, прежде чем помечать файл сломанным.
  • Следите за remaining_api_calls. Его возвращает каждая созданная задача; остановите цикл на невысоком пороге вместо того, чтобы узнавать лимит по стене ошибок.
  • Ведите журнал соответствий. Храните файл с tid, именем входа, настройками и именем результата. Если задача ушла на ночь, этот журнал — единственный способ понять, что завершилось.

Ошибки, которые вы правда увидите

  • valid_api_key_required — заголовка нет или ключ неизвестен. Проверьте имя заголовка и то, что тариф по ключу не истёк.
  • requires_vip — бесплатный или просроченный аккаунт. API рассчитан на активные платные тарифы.
  • month_limit — месячная квота выбрана. Остаток приходит с каждой задачей, так что можно остановиться заранее.
  • param_error — style только art или photo, noise от -1 до 3, x2 от 1 до 4.
  • input_url_error — input должен быть публично доступным URL картинки по http или https. Приватные подписанные ссылки истекают: сначала загрузите файл, затем передайте полученный публичный URL.

Сколько стоит пакет

2x, 4x и 8x — один кредит за картинку, 16x — два. Пакет из 500 изображений на 4x это 500 кредитов, тот же пакет на 16x — 1000. Для большинства фотографий 4x и так оптимален, поэтому дешёвый вариант обычно же и выглядит лучше. Почему — в гайде по множителям.

Полный справочник по эндпоинтам, с параметрами и всеми статусами ошибок, — на странице API. Если что-то в скрипте расходится с фактическими ответами, актуальная версия именно там.

Bigjpg Team ·

← Все статьи