สอนใช้ API ของ Bigjpg: ขยายภาพทีละมาก ๆ ด้วย Python

ถ้ามีไม่กี่ภาพ หน้าเว็บก็พอ แต่ถ้าเป็นโฟลเดอร์ห้าร้อยไฟล์ การใช้ API สบายกว่า: มีสี่ endpoint ไม่ต้องเปิดเบราว์เซอร์ และรันสคริปต์ซ้ำได้ไม่จำกัด เวอร์ชันด้านล่างคือเวอร์ชันที่เราโยนให้เพื่อนร่วมงานได้เลย — อัปโหลด สร้างงาน เฝ้าดูผล ดาวน์โหลด พร้อมการจัดการความล้มเหลวที่เจอจริงหน้างาน

ก่อนเริ่ม

• API ใช้ได้เฉพาะบัญชีแบบเสียเงิน พอเข้าสู่ระบบด้วยแพ็กเกจที่ยังใช้งานได้ คุณจะเจอคีย์ที่ หน้า API
• Python 3 กับ requests ติดตั้งไว้ แค่นั้น
• การขยายภาพทำงานแบบ async บน worker ของเรา สคริปต์จึงสร้างงานก่อนแล้วค่อยวนเช็ก ผลก็คือ งาน 500 ภาพจะใช้เวลาเท่างานที่ช้าที่สุดในคิวที่คุณอยู่ ไม่ใช่ผลรวมของทุกภาพ

สคริปต์

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" สำหรับภาพวาด, "photo" สำหรับภาพถ่าย
NOISE = "1"            # -1 ไม่ลด, 0 น้อย, 1 ปานกลาง, 2 มาก, 3 สูงสุด
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":
                # ลองใหม่อีกครั้งเดียว ถ้าไม่ผ่านก็ปล่อยไฟล์นี้
                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:])

แต่ละคำสั่งทำอะไร

การทำพร้อมกันและมารยาท

• เช็กให้ห่างกว่าที่คิด การขยายภาพไม่ใช่เรื่องทันที 5–10 วินาทีต่อครั้งก็เพียงพอ และการถาม 100 id ในคำสั่งเดียวช่วยให้จำนวนคำขอไม่บาน
• จำกัดจำนวนอัปโหลด สองถึงสี่อัปโหลดพร้อมกันถือเป็นเพดานที่สมเหตุสมผลต่อหนึ่งเครื่อง ถ้าต้องการปริมาณมากขึ้น ให้เพิ่มจำนวนอินสแตนซ์ ไม่ใช่เพิ่มเธรดในอินสแตนซ์เดียว
• ลองใหม่แบบถอยเวลาออกไป เน็ตเวิร์ก error ไม่ใช่ความล้มเหลวของงาน ลองคำสั่ง HTTP ซ้ำได้ถึงสามครั้งโดยเว้นช่วงห่างขึ้นเรื่อย ๆ ก่อนจะตัดสินว่าไฟล์นั้นพัง
• ดู remaining_api_calls ทุกงานที่สร้างสำเร็จจะรายงานค่านี้ หยุดลูปที่ระดับต่ำ ๆ ดีกว่าไปรู้ข้อจำกัดตอนเจอกำแพง error
• บันทึกตารางจับคู่ เก็บไฟล์ที่ระบุ tid ชื่อไฟล์ input ค่าตั้ง และชื่อไฟล์ output ไว้ เมื่องานรันข้ามคืน ไฟล์บันทึกนั้นคือทางเดียวที่จะรู้ว่าอะไรเสร็จแล้ว

error ที่จะเจอจริง

• valid_api_key_required — ไม่มี header หรือไม่รู้จักคีย์ ตรวจชื่อ header และดูว่าแพ็กเกจของคีย์นั้นยังไม่หมดอายุ
• requires_vip — บัญชีฟรีหรือหมดอายุแล้ว API ใช้กับแพ็กเกจเสียเงินที่ยังใช้งานได้
• month_limit — โควตารายเดือนหมด ทุกงานจะรายงานส่วนที่เหลือมาด้วย จึงหยุดก่อนถึงจุดนั้นได้
• param_error — style ต้องเป็น art หรือ photo, noise ตั้งแต่ -1 ถึง 3, x2 ตั้งแต่ 1 ถึง 4
• input_url_error — input ต้องเป็น URL รูปภาพแบบ http หรือ https ที่เข้าถึงได้จากที่อื่น URL ส่วนตัวแบบ signed จะหมดอายุ ให้อัปโหลดก่อนแล้วส่ง URL สาธารณะที่ได้กลับมา

ค่าใช้จ่ายของงานหนึ่งรอบ

ที่ 2x, 4x และ 8x ภาพหนึ่งภาพนับเป็นหนึ่งเครดิต ส่วน 16x นับเป็นสอง ดังนั้น 500 ภาพที่ 4x คือ 500 เครดิต และงานเดิมที่ 16x คือ 1000 เพราะภาพถ่ายส่วนใหญ่ใช้ 4x อยู่แล้ว ตัวเลือกที่ถูกกว่าจึงมักเป็นตัวที่ดูดีกว่าด้วย เหตุผลอยู่ใน คู่มือเลือกตัวคูณ

เอกสาร endpoint ฉบับเต็ม ทั้งพารามิเตอร์และสถานะ error ทุกตัว อยู่ที่ หน้า API ถ้ามีอะไรในสคริปต์ไม่ตรงกับที่คุณได้รับกลับมา หน้านั้นคือเวอร์ชันที่เราอัปเดตอยู่เสมอ