API Bigjpg: powiększanie zdjęć w partiach w Pythonie

Przy kilku obrazach interfejs webowy w zupełności wystarczy. Przy folderze z pięciuset plikami wygodniej jest użyć API: cztery endpointy, żadnej przeglądarki i skrypt, który możesz uruchamiać dowolną liczbę razy. Poniżej wersja, którą dalibyśmy koledze z zespołu: wgraj, utwórz, odpytaj, pobierz, plus obsługa awarii, które naprawdę się zdarzają.

Zanim zaczniesz

• Dostęp do API mają konta płatne. Klucz znajdziesz na stronie API, gdy zalogujesz się z aktywnym planem.
• Python 3 z zainstalowanym requests. Nic więcej.
• Powiększanie działa asynchronicznie na naszych workerach, więc skrypt najpierw tworzy zadania, a potem je odpytuje. Partia 500 obrazów zajmuje tyle, ile najwolniejsze zadanie w twojej kolejce, a nie sumę czasów wszystkich plików.

Skrypt

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" do ilustracji, "photo" do zdjęć
NOISE = "1"            # -1 brak, 0 niski, 1 średni, 2 wysoki, 3 najwyższy
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":
                # Jedna ponowna próba, potem odpuszczamy ten plik.
                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:])

Co robi każde wywołanie

Współbieżność i dobre obyczaje

• Odpytuj rzadziej, niż podpowiada intuicja. Powiększanie nie jest natychmiastowe; 5–10 sekund przerwy w zupełności wystarczy, a 100 id w jednym wywołaniu utrzymuje liczbę zapytań na niskim poziomie.
• Ogranicz wgrywanie. Dwa do czterech równoczesnych wgrań to rozsądny limit dla jednej maszyny. Potrzebujesz większej przepustowości, uruchom więcej instancji, a nie więcej wątków w jednej instancji.
• Ponawiaj z rosnącym opóźnieniem. Błąd sieci to nie awaria zadania. Powtórz wywołanie HTTP do trzech razy z coraz dłuższą przerwą, zanim uznasz plik za stracony.
• Patrz na remaining_api_calls. Każde utworzone zadanie je podaje; przerwij pętlę na niskim progu, zamiast poznawać limit przy ścianie błędów.
• Zapisuj mapowanie. Trzymaj plik z tid, nazwą wejściową, ustawieniami i nazwą wyjściową. Gdy zadanie idzie na całą noc, ten log jest jedynym sposobem, żeby wiedzieć, co się skończyło.

Błędy, które faktycznie zobaczysz

• valid_api_key_required — brakuje nagłówka albo klucz jest nieznany. Sprawdź nazwę nagłówka i to, czy plan tego klucza nie wygasł.
• requires_vip — konto darmowe albo wygasłe. API jest dla aktywnych planów płatnych.
• month_limit — miesięczny limit wykorzystany. Każde zadanie podaje resztę, więc możesz zatrzymać się wcześniej.
• param_error — style musi być art albo photo, noise od -1 do 3, x2 od 1 do 4.
• input_url_error — input musi być publicznie dostępnym adresem obrazu po http lub https. Prywatne podpisane adresy wygasają: najpierw wgraj plik, potem przekaż zwrócony publiczny adres.

Ile kosztuje partia

Jeden obraz to jeden kredyt przy 2x, 4x i 8x; przy 16x dwa. Partia 500 obrazów w 4x to więc 500 kredytów, a ta sama partia w 16x to 1000. Ponieważ 4x i tak jest właściwym mnożnikiem dla większości zdjęć, tańsza opcja zwykle wygląda też lepiej — uzasadnienie znajdziesz w przewodniku po mnożnikach.

Pełne opisanie endpointów, razem z parametrami i wszystkimi statusami błędów, jest na stronie API. Jeśli coś w skrypcie nie zgadza się z tym, co dostajesz w odpowiedzi, aktualna wersja jest właśnie tam.