用一个 API Key 把商品图批量生成 15 秒视频

curl https://api.makernice.com/v1/generations/quote \
  -H "Authorization: Bearer $MAKERNICE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "wan-3.0",
    "mode": "i2v",
    "prompt": "The ceramic mug turns slowly on a white tabletop, soft studio light, steam rising, slow push-in",
    "image_url": "https://example.com/products/sku-1042.jpg",
    "duration": 15,
    "resolution": "720p"
  }'

curl https://api.makernice.com/v1/generations \
  -H "Authorization: Bearer $MAKERNICE_API_KEY" \
  -H "Idempotency-Key: product-video-sku-1042-v1" \
  -H "Content-Type: application/json" \
  -d '{
    "quote_id": "qt_...",
    "input": {
      "model": "wan-3.0",
      "mode": "i2v",
      "prompt": "The ceramic mug turns slowly on a white tabletop, soft studio light, steam rising, slow push-in",
      "image_url": "https://example.com/products/sku-1042.jpg",
      "duration": 15,
      "resolution": "720p"
    }
  }'

import csv
import os
import time
from decimal import Decimal

import requests

API = "https://api.makernice.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['MAKERNICE_API_KEY']}"}
FINAL = {"succeeded", "failed", "canceled"}


def call(method, path, headers=None, **kwargs):
    # Waits out 429 (rate_limited, concurrency_limit_reached) using Retry-After; raises on anything else
    for _ in range(20):
        r = requests.request(method, API + path, headers={**HEADERS, **(headers or {})}, timeout=30, **kwargs)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5")))
            continue
        if not r.ok:
            raise RuntimeError(f"{method} {path} -> {r.status_code}: {r.text}")
        return r.json()
    raise RuntimeError(f"{method} {path}: still rate limited")


# products.csv columns: sku,image_url,prompt
with open("products.csv", newline="") as f:
    products = list(csv.DictReader(f))

for p in products:
    job = {
        "model": "wan-3.0",
        "mode": "i2v",
        "prompt": p["prompt"],
        "image_url": p["image_url"],
        "duration": 15,
        "resolution": "720p",
    }
    # Quote right before creating: a quote is valid for 12 minutes and one generation
    quote = call("POST", "/generations/quote", json=job)
    print(p["sku"], "quoted", Decimal(quote["amount"]).scaleb(-6), "USD")
    gen = call(
        "POST",
        "/generations",
        json={"quote_id": quote["id"], "input": job},
        headers={"Idempotency-Key": f"product-video-{p['sku']}-v1"},
    )
    while gen["status"] not in FINAL:
        time.sleep(5)
        gen = call("GET", f"/generations/{gen['id']}")
    print(p["sku"], gen["status"], gen["billing"], gen.get("error"))