开发者··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 页面。如果脚本里有什么和你实际拿到的返回不一致，以那一页为准，我们一直在维护它。

Bigjpg Team ·

← 全部文章