Tutorial de la API de Bigjpg: ampliar imágenes por lotes con Python

Para un puñado de imágenes, la interfaz web va bien. Con una carpeta de quinientas, conviene la API: cuatro endpoints, sin navegador y un script reutilizable. Esta es la versión que le pasaríamos a un compañero, con el manejo de fallos que de verdad importa.

Antes de empezar

• El acceso a la API es una función de cuenta de pago. La clave está en la página de la API cuando inicias sesión con un plan activo.
• Python 3 con requests instalado. Nada más.
• La ampliación corre en asíncrono en nuestros workers, así que el script crea tareas y luego consulta. Un lote de 500 imágenes tarda lo que la tarea más lenta de tu cola, no la suma de las imágenes.

El script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" para ilustración, "photo" para fotografías
NOISE = "1"            # -1 ninguna, 0 baja, 1 media, 2 alta, 3 máxima
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 reintento y, si falla, descartamos el archivo.
                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:])

Qué hace cada llamada

Concurrencia y buenos modales

• Consulta menos de lo que crees. La ampliación no es instantánea: 5–10 segundos entre consultas sobra, y pedir 100 ids en una llamada mantiene bajas las peticiones.
• Limita las subidas. Dos a cuatro subidas simultáneas es un techo razonable por máquina. Si necesitas más caudal, levanta más instancias en vez de más hilos por instancia.
• Reintenta con espera creciente. Un error de red no es un fallo de tarea. Repite la llamada HTTP hasta tres veces con pausas cada vez mayores antes de dar el archivo por roto.
• Vigila remaining_api_calls. Cada tarea creada lo informa; corta el bucle en un umbral bajo en lugar de descubrir el límite con un muro de errores.
• Registra el mapeo. Guarda un archivo con tid, entrada, ajustes y salida. Si el trabajo corre de noche, ese registro dice qué terminó.

Errores que vas a ver de verdad

• valid_api_key_required — falta la cabecera o la clave es desconocida. Revisa el nombre de la cabecera y que el plan de esa clave no haya caducado.
• requires_vip — cuenta gratuita o caducada. La API es para planes de pago activos.
• month_limit — cuota mensual agotada. Cada tarea informa del resto, así que puedes parar antes de llegar ahí.
• param_error — style debe ser art o photo, noise de -1 a 3 y x2 de 1 a 4.
• input_url_error — el input tiene que ser una URL de imagen http o https accesible públicamente. Las URLs privadas prefirmadas caducan: sube primero y luego pasa la URL pública que te devuelvan.

Lo que cuesta un lote

Una imagen cuenta como un crédito en 2x, 4x y 8x; en 16x cuenta dos. Un lote de 500 imágenes en 4x son 500 créditos, y en 16x son 1000. Como 4x es el factor correcto para la mayoría de las fotografías, la opción barata suele verse mejor; el razonamiento está en la guía de factores.

La referencia completa de endpoints, con parámetros y todos los estados de error, está en la página de la API. Si algo del script no cuadra con lo que recibes, esa página es la versión al día.