Tutorial da API do Bigjpg: ampliar imagens em lote com Python

Para meia dúzia de imagens, a interface web resolve. Com uma pasta de quinhentos arquivos, a API rende mais: quatro endpoints, sem navegador e um script reutilizável. Esta é a versão que a gente entregaria a um colega, com o tratamento de falhas do dia a dia.

Antes de começar

• O acesso à API é recurso de conta paga. A chave fica na página da API quando você entra com um plano ativo.
• Python 3 com requests instalado. Só isso.
• A ampliação roda de forma assíncrona nos nossos workers, então o script cria tarefas e depois consulta. Um lote de 500 imagens leva o tempo da tarefa mais lenta da sua fila, não a soma das imagens.

O script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" para ilustração, "photo" para fotografias
NOISE = "1"            # -1 nenhum, 0 baixo, 1 médio, 2 alto, 3 máximo
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":
                # Uma tentativa extra; se falhar, desistimos do arquivo.
                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:])

O que cada chamada faz

Concorrência e boas maneiras

• Consulte menos do que você imagina. A ampliação não é instantânea; 5–10 segundos entre consultas bastam, e perguntar por 100 ids numa chamada só mantém o número de requisições baixo.
• Limite os uploads. Dois a quatro uploads simultâneos é um teto razoável por máquina. Se precisar de mais vazão, suba mais instâncias em vez de mais threads por instância.
• Repita com espera crescente. Erro de rede não é falha de tarefa. Repita a chamada HTTP até três vezes com pausas cada vez maiores antes de marcar o arquivo como perdido.
• Acompanhe remaining_api_calls. Toda tarefa criada informa esse número; pare o laço num limite baixo em vez de descobrir a cota com uma parede de erros.
• Registre o mapeamento. Mantenha um arquivo com tid, entrada, ajustes e saída. Se o trabalho roda a noite inteira, esse registro diz o que terminou.

Erros que você vai encontrar mesmo

• valid_api_key_required — o cabeçalho está ausente ou a chave é desconhecida. Confira o nome do cabeçalho e se o plano dessa chave não expirou.
• requires_vip — conta gratuita ou expirada. A API é para planos pagos ativos.
• month_limit — a cota mensal acabou. Cada tarefa informa o que sobrou, então dá para parar antes disso.
• param_error — style precisa ser art ou photo, noise de -1 a 3 e x2 de 1 a 4.
• input_url_error — o input tem que ser uma URL de imagem http ou https acessível publicamente. URLs privadas pré-assinadas expiram; envie o arquivo primeiro e depois passe a URL pública devolvida.

Quanto custa um lote

Uma imagem vale um crédito em 2x, 4x e 8x; em 16x vale dois. Um lote de 500 imagens em 4x são 500 créditos, e em 16x vira 1000. Como 4x é o fator certo para a maioria das fotos, a opção barata costuma ficar melhor; o raciocínio está no guia de fatores.

A referência completa dos endpoints, com parâmetros e todos os estados de erro, está na página da API. Se algo no script não bater com o que você recebe, essa página é a versão atualizada.