Tutoriel API Bigjpg : agrandir des images par lot en Python

Pour quelques images, l’interface web suffit. Avec un dossier de cinq cents fichiers, l’API est plus confortable : quatre endpoints, pas de navigateur, un script relançable à volonté. Voici la version qu’on passerait à un collègue, avec la gestion des échecs qui compte vraiment.

Avant de commencer

• L’accès à l’API est réservé aux comptes payants. La clé se trouve sur la page API une fois connecté avec une formule active.
• Python 3 avec requests installé. Rien d’autre.
• L’agrandissement tourne en asynchrone sur nos workers : le script crée les tâches, puis les interroge. Un lot de 500 images prend le temps de la tâche la plus lente de votre file, pas la somme des images.

Le script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" pour l’illustration, "photo" pour les photos
NOISE = "1"            # -1 aucun, 0 faible, 1 moyen, 2 fort, 3 maximal
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":
                # Une seule nouvelle tentative, puis on abandonne ce fichier.
                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:])

Ce que fait chaque appel

Concurrence et bonnes manières

• Interrogez moins souvent que vous ne le pensez. L’agrandissement n’est pas instantané : 5–10 secondes entre deux requêtes suffisent, et demander 100 ids d’un coup garde le nombre d’appels bas.
• Limitez les téléversements. Deux à quatre envois simultanés, c’est un plafond raisonnable par machine. Pour plus de débit, lancez plus d’instances plutôt que plus de threads par instance.
• Réessayez avec un délai croissant. Une erreur réseau n’est pas un échec de tâche. Relancez l’appel HTTP jusqu’à trois fois avec une attente de plus en plus longue avant de déclarer le fichier perdu.
• Surveillez remaining_api_calls. Chaque tâche créée l’indique ; coupez la boucle à un seuil bas au lieu de découvrir la limite devant un mur d’erreurs.
• Tenez le journal des correspondances. Gardez un fichier avec tid, entrée, réglages et sortie. Si le travail tourne toute la nuit, ce journal dit ce qui est terminé.

Les erreurs que vous verrez vraiment

• valid_api_key_required — en-tête absent ou clé inconnue. Vérifiez le nom de l’en-tête et que la formule associée à la clé n’a pas expiré.
• requires_vip — compte gratuit ou expiré. L’API est prévue pour les formules payantes actives.
• month_limit — quota mensuel épuisé. Chaque tâche indique le reste, vous pouvez donc vous arrêter avant.
• param_error — style doit valoir art ou photo, noise de -1 à 3, x2 de 1 à 4.
• input_url_error — l’input doit être une URL d’image http ou https accessible publiquement. Les URL privées signées expirent : téléversez d’abord, puis transmettez l’URL publique renvoyée.

Ce que coûte un lot

Une image compte pour un crédit en 2x, 4x et 8x ; en 16x elle en compte deux. Un lot de 500 images en 4x coûte 500 crédits, le même en 16x coûte 1000. Comme 4x est de toute façon le bon facteur pour la plupart des photos, l’option la moins chère est souvent la plus belle — l’explication est dans le guide des facteurs.

La référence complète des endpoints, paramètres et états d’erreur compris, est sur la page API. Si un point du script ne correspond pas à ce que vous recevez, c’est cette page que nous tenons à jour.