Bigjpg API: Bilder mit Python im Stapel vergrößern

Für ein paar Bilder reicht die Weboberfläche. Bei einem Ordner mit fünfhundert Dateien ist die API angenehmer: vier Endpunkte, kein Browser, und das Skript lässt sich beliebig oft erneut starten. Das hier ist die Fassung, die wir einem Kollegen in die Hand drücken würden — hochladen, anlegen, abfragen, herunterladen, mit der Fehlerbehandlung, die in der Praxis zählt.

Bevor du anfängst

• Der API-Zugang gehört zum bezahlten Konto. Den Schlüssel findest du auf der API-Referenz, sobald du mit einem aktiven Tarif angemeldet bist.
• Python 3 mit installiertem requests. Sonst nichts.
• Die Vergrößerung läuft asynchron auf unseren Workern. Das Skript legt also Aufträge an und fragt danach nach. Ein Stapel mit 500 Bildern dauert ungefähr so lange wie der langsamste Auftrag in deiner Warteschlange, nicht wie die Summe aller Bilder.

Das Skript

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" für Illustrationen, "photo" für Fotos
NOISE = "1"            # -1 aus, 0 niedrig, 1 mittel, 2 hoch, 3 höchste
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":
                # Einmal wiederholen, dann diese Datei aufgeben.
                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:])

Was jeder Aufruf macht

Nebenläufigkeit und Manieren

• Seltener abfragen, als du denkst. Die Vergrößerung braucht Zeit; 5–10 Sekunden Abstand reichen völlig, und 100 IDs pro Anfrage halten die Zahl der Requests klein.
• Uploads begrenzen. Zwei bis vier gleichzeitige Uploads sind für eine Maschine ein vernünftiger Deckel. Mehr Durchsatz bekommst du über mehr Instanzen, nicht über mehr Threads pro Instanz.
• Mit Backoff wiederholen. Ein Netzwerkfehler ist kein Auftragsfehler. Wiederhole den HTTP-Aufruf bis zu dreimal mit wachsender Pause, bevor du eine Datei als kaputt abhakst.
• remaining_api_calls im Blick behalten. Jeder angelegte Auftrag meldet den Rest; brich die Schleife an einer niedrigen Schwelle ab, statt das Limit an einer Wand voller Fehler zu entdecken.
• Die Zuordnung protokollieren. Eine Datei mit tid, Eingabename, Einstellungen und Ausgabenamen. Wenn ein Job über Nacht läuft, ist dieses Protokoll der einzige Beleg, was fertig geworden ist.

Fehler, die du wirklich siehst

• valid_api_key_required — Header fehlt oder Schlüssel unbekannt. Prüfe den Header-Namen und ob der Schlüssel zu einem nicht abgelaufenen Tarif gehört.
• requires_vip — kostenloses oder abgelaufenes Konto. Die API ist für aktive bezahlte Tarife gedacht.
• month_limit — Monatskontingent aufgebraucht. Der Rest steht bei jedem Auftrag, du kannst also vorher aufhören.
• param_error — style muss art oder photo sein, noise von -1 bis 3, x2 von 1 bis 4.
• input_url_error — der input muss eine öffentlich erreichbare Bild-URL über http oder https sein. Signierte private URLs laufen ab: erst hochladen, dann die zurückgegebene öffentliche URL übergeben.

Was ein Stapel kostet

Ein Bild zählt bei 2x, 4x und 8x als ein Credit; 16x zählt als zwei. Ein Stapel mit 500 Bildern kostet bei 4x also 500 Credits, derselbe Stapel bei 16x 1000. 4x ist für die meisten Fotos ohnehin der richtige Faktor, die günstige Variante sieht also meistens auch besser aus — die Begründung steht im Faktor-Leitfaden.

Die vollständige Endpunkt-Referenz mit Parametern und allen Fehlerzuständen findest du auf der API-Seite. Wenn etwas im Skript nicht zu dem passt, was du zurückbekommst, ist diese Seite die aktuelle Fassung.