開發者··4 分鐘讀完

Bigjpg API 教學:用 Python 批次放大圖片

一支能直接跑的腳本:上傳圖片、建立放大任務、輪詢結果、下載成品,外加併發控制與錯誤處理,讓 500 張的批次工作不會跑到一半就斷。

Bigjpg API 教學:用 Python 批次放大圖片

處理幾張圖,網頁介面就夠了。要面對一個五百張的資料夾,還是用 API 省事:四個接口、不用開瀏覽器、腳本隨時能重跑。下面這一版就是我們平常會直接丟給同事的程式——上傳、建立、輪詢、下載,再加上實務上真的會遇到的失敗處理。

開始之前

  • API 只對付費帳號開放。登入後帶著有效方案,金鑰就在 API 參考頁 上。
  • Python 3,裝好 requests,其他都不用。
  • 放大在我們的 worker 上非同步執行,所以腳本是「先建任務、再輪詢」。五百張的工作,耗時取決於你所在佇列裡最慢的那一張,而不是把每張的時間加起來。

腳本

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:])

每個接口做什麼

呼叫作用說明
POST /api/img_upload回傳預簽名的上傳網址和圖片公開網址用你宣告過的 Content-Type PUT 上傳檔案,再把 fileurl 當成任務的 input
POST /api/task/建立放大任務回傳 tid、預估的 minute,以及 remaining_api_calls
GET /api/task/{ids}查詢一個或多個任務的狀態id 以逗號分隔,單次最多 100 個
POST /api/task/{ids}重試失敗的任務只有處於 error 狀態的任務會回到佇列

併發與禮數

  • 輪詢別太勤。放大不是瞬間完成的,5–10 秒查一次就夠;一次問 100 個 id,請求數也壓得住。
  • 限制上傳併發。單台機器同時跑兩到四個上傳比較合適。想要更高吞吐就多開幾個實例,別在同一個實例裡加執行緒。
  • 退避重試。網路錯誤不等於任務失敗。HTTP 呼叫最多重試三次,間隔一次比一次長,再判定檔案真的壞了。
  • 盯著 remaining_api_calls。每個建立成功的任務都會帶上它;設一個低位門檻就停下,別等錯誤訊息洗版才發現額度見底。
  • 把對應關係記下來。用一份檔案記下 tid、輸入檔名、參數和輸出檔名。任務跑一整夜時,這份日誌是唯一能查證的東西。

實際會撞上的錯誤

  • valid_api_key_required —— 標頭缺失或無法辨識。檢查標頭名稱,並確認這把金鑰所屬的方案還沒過期。
  • requires_vip —— 免費或已過期帳號。API 只面向有效期內的付費方案。
  • month_limit —— 當月額度用完。每個任務都會回報剩餘量,完全可以在此之前主動收手。
  • param_error —— style 只能是 art 或 photo,noise 取 -1 到 3,x2 取 1 到 4。
  • input_url_error —— 輸入必須是可以公開存取的 http 或 https 圖片網址。預簽名的私有網址會過期;先上傳,再把回傳的公開網址傳進去。

批次要花多少

2x、4x、8x 每張算一個額度,16x 算兩個。五百張按 4x 就是 500 額度,同一批走 16x 就是 1000。多數照片本來用 4x 就夠,所以便宜的那個選項往往也是更好看的那個——原因見 倍率選擇指南。

完整的接口文件,包含所有參數和每一種錯誤狀態,都在 API 頁面。如果腳本裡有什麼和你實際拿到的回傳不一致,以那一頁為準,我們一直在維護它。

← 全部文章