Tutorial API Bigjpg: ingrandire immagini in batch con Python

Per qualche immagine l’interfaccia web basta. Con una cartella da cinquecento file conviene l’API: quattro endpoint, nessun browser e uno script rilanciabile quando vuoi. Questa è la versione che passeremmo a un collega, con la gestione dei guasti che conta.

Prima di iniziare

• L’accesso all’API è una funzione degli account a pagamento. La chiave sta nella pagina API quando accedi con un piano attivo.
• Python 3 con requests installato. Nient’altro.
• L’ingrandimento gira in modo asincrono sui nostri worker, quindi lo script crea i task e poi li interroga. Un lotto da 500 immagini dura quanto il task più lento nella tua coda, non quanto la somma delle immagini.

Lo script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" per l’illustrazione, "photo" per le fotografie
NOISE = "1"            # -1 nessuno, 0 basso, 1 medio, 2 alto, 3 massimo
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":
                # Un solo nuovo tentativo, poi molliamo questo file.
                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:])

Cosa fa ogni chiamata

Concorrenza e buone maniere

• Interroga meno spesso di quanto pensi. L’ingrandimento non è istantaneo: 5–10 secondi tra un controllo e l’altro bastano, e chiedere 100 id in una sola chiamata tiene basso il numero di richieste.
• Limita i caricamenti. Da due a quattro caricamenti in volo è un tetto ragionevole per una macchina. Se serve più throughput, avvia più istanze invece di aggiungere thread alla singola istanza.
• Riprova con attesa crescente. Un errore di rete non è un task fallito. Ripeti la chiamata HTTP fino a tre volte con pause sempre più lunghe prima di dare il file per perso.
• Tieni d’occhio remaining_api_calls. Ogni task creato lo riporta; ferma il ciclo a una soglia bassa invece di scoprire il limite davanti a un muro di errori.
• Registra la corrispondenza. Tieni un file con tid, nome di input, impostazioni e nome di output. Quando il lavoro gira tutta la notte, quel registro è l’unico modo per sapere cosa è finito.

Errori che incontrerai davvero

• valid_api_key_required — header assente o chiave sconosciuta. Controlla il nome dell’header e che il piano di quella chiave non sia scaduto.
• requires_vip — account gratuito o scaduto. L’API è per i piani a pagamento attivi.
• month_limit — quota mensile esaurita. Ogni task riporta quanto resta, quindi puoi fermarti prima.
• param_error — style deve essere art o photo, noise da -1 a 3, x2 da 1 a 4.
• input_url_error — l’input dev’essere un URL immagine http o https raggiungibile pubblicamente. Gli URL privati firmati scadono: carica prima il file, poi passa l’URL pubblico restituito.

Quanto costa un lotto

Un’immagine vale un credito a 2x, 4x e 8x; a 16x ne vale due. Un lotto da 500 immagini a 4x costa quindi 500 crediti, lo stesso lotto a 16x ne costa 1000. Dato che 4x è comunque il fattore giusto per la maggior parte delle fotografie, l’opzione economica è spesso anche quella che si vede meglio: il ragionamento è nella guida ai fattori.

Il riferimento completo degli endpoint, con parametri e tutti gli stati di errore, è sulla pagina API. Se qualcosa nello script non combacia con le risposte che ricevi, quella pagina è la versione che teniamo aggiornata.