Hướng dẫn API Bigjpg: phóng to ảnh hàng loạt bằng Python

Vài tấm ảnh thì giao diện web là đủ. Nhưng với thư mục năm trăm tệp, API nhàn hơn: bốn endpoint, không cần trình duyệt, chạy lại lúc nào cũng được. Đây là bản chúng tôi đưa cho đồng nghiệp, kèm xử lý lỗi thực tế.

Trước khi bắt đầu

• API thuộc về tài khoản trả phí. Khóa nằm ở trang API sau khi bạn đăng nhập bằng gói còn hiệu lực.
• Python 3 cùng requests. Không cần gì thêm.
• Phóng to chạy bất đồng bộ trên worker của chúng tôi, nên script tạo tác vụ rồi thăm dò. Lô 500 ảnh tốn thời gian bằng tác vụ chậm nhất trong hàng đợi.

Script

import json
import os
import time
import requests

API = "https://bigjpg.com"
KEY = "YOUR_API_KEY"
STYLE = "art"          # "art" cho tranh vẽ, "photo" cho ảnh chụp
NOISE = "1"            # -1 không, 0 thấp, 1 vừa, 2 cao, 3 cao nhất
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":
                # Thử lại một lần, không được thì bỏ tệp này.
                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:])

Mỗi lệnh gọi làm gì

Đồng thời và phép lịch sự

• Thăm dò thưa hơn bạn nghĩ. Phóng to không tức thì; 5–10 giây một lần là đủ, và hỏi 100 id trong một lần gọi giữ số lượt gọi thấp.
• Giới hạn số lần tải lên. Hai đến bốn lượt cùng lúc là mức trần hợp lý cho một máy. Cần nhanh hơn thì chạy thêm instance, đừng thêm thread.
• Thử lại với độ trễ tăng dần. Lỗi mạng không phải tác vụ hỏng. Thử lại lệnh HTTP tối đa ba lần với khoảng chờ dài dần trước khi bỏ tệp.
• Theo dõi remaining_api_calls. Mỗi tác vụ đều báo con số này; dừng ở ngưỡng thấp thay vì biết giới hạn nhờ một bức tường lỗi.
• Ghi lại bảng đối chiếu. Giữ một tệp ghi tid, tên ảnh vào, thiết lập và tên ảnh ra. Chạy qua đêm thì đó là cách duy nhất để biết cái gì đã xong.

Những lỗi bạn chắc chắn gặp

• valid_api_key_required — thiếu header hoặc khóa không nhận ra. Kiểm tra tên header và hạn của gói.
• requires_vip — tài khoản miễn phí hoặc đã hết hạn. API dành cho gói trả phí còn hiệu lực.
• month_limit — hết hạn mức tháng. Mỗi tác vụ đều báo phần còn lại, nên bạn dừng trước mức đó.
• param_error — style phải là art hoặc photo, noise từ -1 đến 3, x2 từ 1 đến 4.
• input_url_error — input phải là URL ảnh http hoặc https truy cập công khai. URL có chữ ký sẽ hết hạn; tải lên trước rồi truyền URL công khai trả về.

Một lô tốn bao nhiêu

Mỗi ảnh tính một credit ở 2x, 4x và 8x; ở 16x tính hai. Lô 500 ảnh ở 4x là 500 credit, ở 16x là 1000. Vì 4x vốn đúng cho phần lớn ảnh chụp, lựa chọn rẻ hơn thường cũng đẹp hơn — xem hướng dẫn chọn hệ số.

Tài liệu đầy đủ về endpoint, gồm tham số và mọi trạng thái lỗi, nằm ở trang API. Nếu script không khớp với phản hồi nhận được, trang đó là bản mới nhất.