Bigjpg API: Python ile toplu görsel büyütme

Birkaç görsel için web arayüzü yeter. Elinde beş yüz dosyalık bir klasör varsa API daha rahat: dört uç nokta, tarayıcı yok, betiği istediğin kadar yeniden çalıştırabilirsin. Aşağıdaki, iş arkadaşına doğrudan verdiğimiz sürüm: yükle, oluştur, yokla, indir ve pratikte işine yarayacak hata yönetimi.

Başlamadan önce

• API ücretli hesap özelliği. Etkin bir paketle giriş yaptığında anahtarın API sayfasında durur.
• Python 3 ve kurulu requests. Başka bir şey gerekmiyor.
• Büyütme işlemleri worker’larımızda asenkron çalışır; betik önce görev oluşturur, sonra yoklar. 500 görsellik bir iş, görsellerin toplamı kadar değil, bulunduğun kuyruktaki en yavaş iş kadar sürer.

Betik

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" çizim, "photo" fotoğraf için
NOISE = "1"            # -1 yok, 0 az, 1 orta, 2 yüksek, 3 en yüksek
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":
                # Bir kez daha dene, olmazsa bu dosyadan vazgeç.
                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:])

Her çağrı ne yapar

Eşzamanlılık ve görgü

• Yoklamayı düşündüğünden seyrek yap. Büyütme anında bitmiyor; 5–10 saniye aralık fazlasıyla yeter ve 100 id’yi tek çağrıda sormak istek sayısını düşük tutar.
• Yüklemeleri sınırla. Tek makine için aynı anda iki ile dört yükleme makul bir tavan. Daha fazla hız gerekiyorsa örnek başına iş parçacığını değil, örnek sayısını artır.
• Gecikmeli tekrar dene. Ağ hatası görev hatası değildir. Dosyayı bozuk saymadan önce HTTP çağrısını artan beklemeyle üç kez yenile.
• remaining_api_calls değerini izle. Kurulan her görev bunu bildirir; hata duvarını görmeden önce düşük bir eşikte döngüyü durdur.
• Eşleşmeyi kaydet. tid, giriş dosya adı, ayarlar ve çıktı adını bir dosyada tut. İş gece boyunca dönerken neyin bittiğini anlamanın tek yolu o kayıt.

Gerçekten karşılaşacağın hatalar

• valid_api_key_required — başlık yok ya da anahtar tanınmıyor. Başlık adını ve anahtarın süresi geçmemiş bir pakete ait olduğunu kontrol et.
• requires_vip — ücretsiz veya süresi dolmuş hesap. API etkin ücretli paketler için.
• month_limit — aylık kota bitmiş. Kalan miktar her görevle birlikte gelir, yani bu noktaya gelmeden durabilirsin.
• param_error — style art ya da photo olmalı, noise -1 ile 3, x2 1 ile 4 arasında.
• input_url_error — input, herkesin erişebildiği bir http veya https görsel adresi olmalı. İmzalı özel adresler zamanla geçersiz olur; önce yükle, sonra dönen genel adresi ver.

Bir toplu işin maliyeti

2x, 4x ve 8x’te görsel başına bir kredi, 16x’te iki. Yani 500 görsel 4x’te 500 kredi, aynı iş 16x’te 1000 kredi. Çoğu fotoğraf için doğru faktör zaten 4x olduğundan ucuz olan seçenek bir de daha iyi görünür; gerekçesi faktör rehberi sayfasında.

Parametreler ve tüm hata durumları dahil eksiksiz uç nokta referansı API sayfasında. Betikteki bir şey aldığın yanıtla uyuşmuyorsa güncel tuttuğumuz sürüm o sayfadır.