Bigjpg API 튜토리얼: Python으로 이미지 여러 장 한 번에 확대

몇 장이면 웹 인터페이스로 충분합니다. 오백 장짜리 폴더를 앞에 두면 API가 편합니다. 엔드포인트 네 개, 브라우저 없이, 스크립트는 언제든 다시 돌릴 수 있습니다. 아래는 저희가 동료에게 그대로 건네는 버전입니다. 업로드, 생성, 폴링, 다운로드에 실무에서 실제로 필요한 실패 처리를 붙였습니다.

시작하기 전에

• API는 유료 계정 기능입니다. 활성 요금제로 로그인하면 키가 API 페이지에 있습니다.
• Python 3와 requests만 있으면 됩니다. 그 외에는 아무것도 필요 없습니다.
• 확대는 저희 워커에서 비동기로 돌아갑니다. 그래서 스크립트는 작업을 만든 뒤 폴링합니다. 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를 물으면 요청 수가 낮게 유지됩니다.
• 업로드 수를 제한하세요. 한 대당 동시 업로드 두세 개가 적당한 상한입니다. 처리량이 더 필요하면 인스턴스당 스레드를 늘리지 말고 인스턴스를 늘리세요.
• 간격을 늘려 재시도하세요. 네트워크 오류는 작업 실패가 아닙니다. 파일을 깨졌다고 판단하기 전에 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 — input은 공개적으로 접근 가능한 http 또는 https 이미지 URL이어야 합니다. 서명된 비공개 URL은 만료됩니다. 먼저 업로드하고, 돌려받은 공개 URL을 넘기세요.

배치 비용

2x, 4x, 8x는 한 장에 크레딧 1개, 16x는 2개입니다. 500장을 4x로 돌리면 500크레딧, 같은 작업을 16x로 돌리면 1000입니다. 대부분의 사진에는 어차피 4x가 맞는 배율이라, 더 싼 쪽이 결과도 더 좋은 경우가 많습니다. 이유는 배율 가이드에 정리했습니다.

파라미터와 모든 오류 상태를 포함한 전체 엔드포인트 레퍼런스는 API 페이지에 있습니다. 스크립트 내용이 실제 응답과 어긋난다면, 그 페이지가 저희가 계속 최신으로 관리하는 버전입니다.