用 Python 把上传到下载的放大全流程串起来
脚本要做的五件事
把前几篇的接口段拼成一条流水线：
- 申请临时上传地址（img_upload）；
- PUT 原图到上传地址；
- 创建放大任务（task，带四个参数）；
- 轮询任务到终态（GET task）；
- 失败走重试（POST task），成功下载结果到本地。
完整骨架
import time
import requests
API_KEY = "你的密钥" # 用户中心 API 页复制
ROOT = "bigjpg.com" # 接口域名（补协议在下方拼接）
HEADERS = {"X-API-KEY": API_KEY}
BASE = "https" + "://" + ROOT # 拼出接口根地址
def upload_image(path):
# 第一步：申请临时上传地址
r = requests.post(f"{BASE}/api/img_upload", headers=HEADERS)
info = r.json()
# 第二步：PUT 文件（Content-Type 保持一致）
with open(path, "rb") as f:
requests.put(info["fileurl"], data=f,
headers={"Content-Type": "image/jpeg"})
return info["fileurl"] # 作为任务的 input
def create_task(image_url, style="art", noise="3", scale="1"):
# 第三步：提交放大任务
payload = {"style": style, "noise": noise,
"x2": scale, "input": image_url}
r = requests.post(f"{BASE}/api/task/",
json=payload, headers=HEADERS)
return r.json() # 里面有任务 ID（字段名见实际返回）
def wait_done(tid, timeout=3600):
# 第四步：轮询到终态
deadline = time.time() + timeout
while time.time() < deadline:
r = requests.get(f"{BASE}/api/task/{tid}", headers=HEADERS)
task = r.json()
# 状态字段以实际返回为准，这里示意判断
if task.get("status") == "finished":
return task
if task.get("status") == "failed":
return None # 交给重试逻辑，见上一节
time.sleep(60)
raise TimeoutError(tid)
def download(url, out_path):
# 第五步：结果落盘
with requests.get(url, stream=True) as r, open(out_path, "wb") as f:
for chunk in r.iter_content(8192):
f.write(chunk)
if __name__ == "__main__":
image = upload_image("your_image.jpg")
task = create_task(image, style="art", noise="1", scale="1")
result = wait_done(task["tid"]) # 字段名以实际返回为准
if result:
download(result["url"], "output.png")运行前的清单
- [ ] Python 3.8+ 与 requests 库；
- [ ] 密钥已从用户中心复制，未写死进仓库；
- [ ] 接口根地址从 API 页核对（本文用占位写法，不代官方保证拼写）；
- [ ] 测试图放好，先用 2x（scale="1"）跑最小闭环；
- [ ] 账户有可用额度（免费 20 张/月或套餐额度）。
调试顺序建议
- 先打通上传段——img_upload 返回什么、PUT 返回码是否 2xx；
- 手动建一个任务——用 curl 建一个任务，把返回 JSON 完整打出来，确认任务 ID 与状态字段的真实名字，再回填脚本的字段解析；
- 再上轮询与批量——字段确认后，把 批量轮询与重试策略加回来。
与官方示例的关系
本文骨架的接口路径、参数名、鉴权头均整理自官方用户中心的 API 文档示例；返回体字段名以你实际收到的 JSON 为准——这是接入唯一可能「对不上文档」的地方，先打日志再写死。