批量提交 Pika 生成任务之前，先把钱算清楚

批量提交 Pika 生成任务之前，先把钱算清楚：本地按 JSON Schema 校验 → 调用公开的 quote 接口报价（免费，服务端会按真实规则校验）→ 合计报价不超过你设定的预算，也不超过账户余额，才允许提交。提交时带上 Idempotency-Key，避免重试时重复扣费。轮询到任务结束，只在 billing.state 为 settled 时记账；实际扣费高于报价时给出提示。

先看清成品

• 测试： pytest 2 项通过，其中一项调用真实的 quote 接口。
• 报价（真实接口）： 3 个任务里有 1 个被拒，原因是“单张关键帧必须 loop=true”。本地 schema 校验没有查出这个问题。
• 提交（模拟服务器）： 没设预算、超预算、余额不足都被拦下；正常提交时，billing 状态 pending 会等待后再查；实际扣费高于报价时给出警告；同一个 Idempotency-Key 返回同一个任务，换了 body 返回 409。

没有在 box 上跑的部分： box 上没有 Pika Key，没有提交过真实任务，没有花过钱。目录和报价调用的是真实的公开接口；提交、轮询、余额都在我写的模拟服务器 fake_pika.py 上运行，扣费金额是假的。

合规提醒：PIKA_API_KEY 和 webhook 签名密钥只放在服务器端，不要写进浏览器代码、URL、截图或聊天里（Pika 文档原话也是这样要求的）。输入的图片和视频必须是你有权使用的，不要用他人肖像冒充真人，不要侵犯版权或商标。生成的内容发布时要按《人工智能生成合成内容标识办法》加上“AI 生成”标识，并遵守 Pika 的使用条款。Pika API 是预付费的，没有免费的生成额度，也没有测试环境，每次生成都按实时报价扣费。本文不列价格，示例输出里的金额已用“x.xx”代替，请以运行时的报价为准。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

48-pika-B-src/
├── pika_budget_run.py
├── jobs.json
├── fake_pika.py
└── test_budget.py

第 1 步：代码

pika_budget_run.py：

"""一批 Pika 生成任务：先校验、报价、对照预算和余额，确认后才提交；提交带 Idempotency-Key，轮询到结束，只按 settled 的 billing 记账。
步骤：
  1. 用公开目录里的 input_schema（JSON Schema）在本地校验 body —— 快，但查不出跨字段规则
  2. 调用公开的 quote 接口报价（免费、不需要 key）—— 服务端会按真实规则校验，比如“单张关键帧必须 loop=true”
  3. 合计报价 ≤ 预算（PIKA_BUDGET_USD，由你设置）且 ≤ 账户余额（GET /billing/balance）才继续；默认只报价不提交，加 --submit 才提交
  4. 提交：Idempotency-Key = body 的哈希（同一请求重试不会重复扣费；文档：失败的任务要换新 key 重试）
  5. 轮询 GET /v1/media/jobs/{id}，completed / failed 结束；billing.state 为 settled 才记账，pending 稍后再查
金额单位：micro-USD（美元 × 1,000,000），全部用整数计算，不用浮点
环境变量：PIKA_API_KEY、PIKA_BASE（默认 https://api.dev.pika.art）、PIKA_BUDGET_USD、POLL_SECONDS（默认 5）
用法：python pika_budget_run.py jobs.json [--submit]"""
import hashlib, json, os, sys, time
import jsonschema, requests

CATALOG = "https://api.dev.pika.art"           # 目录和报价永远用真实的公开接口
BASE = os.environ.get("PIKA_BASE", "https://api.dev.pika.art")
POLL = float(os.environ.get("POLL_SECONDS", 5))

def usd(micro): return f"{micro / 1e6:.2f} 美元"

def schema(api_id, cache={}):
    if api_id not in cache:
        r = requests.get(f"{CATALOG}/catalog/apis/{api_id}", params={"expand": "inputs"}, timeout=30); r.raise_for_status()
        cache[api_id] = r.json()
    return cache[api_id]

def validate_local(api_id, body):
    try: jsonschema.validate(body, schema(api_id)["input_schema"]); return None
    except jsonschema.ValidationError as e: return f"本地校验未通过：{'/'.join(map(str, e.path)) or '(body)'} {e.message}"

def quote(api_id, body):
    r = requests.post(f"{CATALOG}/catalog/apis/{api_id}/quote", json=body, timeout=30)
    if r.status_code != 200: return None, f"报价被拒（HTTP {r.status_code}）：{r.json().get('message', r.text)}"
    return int(r.json()["micro_usd"]), None

def headers(body=None):
    k = os.environ.get("PIKA_API_KEY") or sys.exit("提交需要环境变量 PIKA_API_KEY")
    h = {"X-API-Key": k, "Content-Type": "application/json"}
    if body is not None: h["Idempotency-Key"] = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()[:32]
    return h

def run_job(job):
    path = schema(job["api_id"])["call"]["path"]
    r = requests.post(BASE + path, json=job["body"], headers=headers(job["body"]), timeout=60)
    j = r.json()
    if r.status_code != 200 or j.get("status") == "failed":
        return {"error": (j.get("error") or {}).get("code") or j.get("message") or r.status_code}
    jid = j["id"]; print(f"  已提交 {jid}")
    while True:
        j = requests.get(f"{BASE}/v1/media/jobs/{jid}", headers=headers(), timeout=30).json()
        if j["status"] == "failed": return {"id": jid, "error": j["error"]["code"]}
        if j["status"] == "completed":
            b = j.get("billing") or {}
            if b.get("state") == "pending": print("  billing 还是 pending，稍后再查"); time.sleep(POLL); continue
            return {"id": jid, "url": j["output"]["video"]["url"], "charge": b.get("charge_micro_usd") if b.get("state") == "settled" else None}
        time.sleep(POLL)

def main():
    jobs = json.load(open(sys.argv[1], encoding="utf-8")); submit = "--submit" in sys.argv
    ok, total = [], 0
    for i, job in enumerate(jobs, 1):
        err = validate_local(job["api_id"], job["body"])
        price, qerr = (None, None) if err else quote(job["api_id"], job["body"])
        if err or qerr: print(f"✗ 任务 {i}（{job['api_id']}）：{err or qerr}"); continue
        total += price; ok.append({**job, "quote": price}); print(f"✓ 任务 {i}（{job['api_id']}）：报价 {usd(price)}")
    budget = os.environ.get("PIKA_BUDGET_USD")
    print(f"合计 {len(ok)} 个可提交任务，报价 {usd(total)}" + (f"，预算 {budget} 美元" if budget else "（未设置 PIKA_BUDGET_USD）"))
    if not submit: print("只报价，没有提交。确认后加 --submit。"); return
    if not budget: sys.exit("✗ 提交前必须设置 PIKA_BUDGET_USD")
    if total > round(float(budget) * 1e6): sys.exit("✗ 报价超出预算，没有提交")
    bal = requests.get(f"{BASE}/billing/balance", headers=headers(), timeout=30).json()["balance_micro_usd"]
    if total > bal: sys.exit(f"✗ 余额 {usd(bal)} 不够，没有提交")
    spent = 0
    for job in ok:
        res = run_job(job); print(f"  → {res}")
        c = res.get("charge")
        if c is not None and c > job["quote"]: print(f"  ⚠ 实际扣费 {usd(c)} 高于报价 {usd(job['quote'])}，请核对账单")
        spent += c or 0
    print(f"实际记账 {usd(spent)}（只计 settled 的 billing）")

if __name__ == "__main__":
    main()

jobs.json：

[
  {"api_id": "pika/pika-2.5/image-to-video", "body": {"image": "https://raw.githubusercontent.com/python-pillow/Pillow/main/Tests/images/hopper.png", "prompt": "镜头缓慢推近", "resolution": "720p", "duration_s": 5}},
  {"api_id": "pika/pikaframes/image-to-video", "body": {"images": ["https://raw.githubusercontent.com/python-pillow/Pillow/main/Tests/images/hopper.png", "https://raw.githubusercontent.com/python-pillow/Pillow/main/Tests/images/hopper.webp"], "transition_duration_s": 4, "resolution": "720p"}},
  {"api_id": "pika/pikaframes/image-to-video", "body": {"images": ["https://raw.githubusercontent.com/python-pillow/Pillow/main/Tests/images/hopper.png"], "resolution": "720p"}}
]

fake_pika.py（仅用于离线测试）：

"""Pika API 提交/轮询/余额接口的模拟服务器（仅用于离线测试；结构按 dev.pika.art 的规格编写，视频是 ffmpeg 测试画面）。
报价（quote）不在这里：报价走真实的公开接口，免费、不需要 key。
- GET  /billing/balance                    {"balance_micro_usd": 余额}，余额由 FAKE_BALANCE_USD 设置（默认 1.00）
- POST /v1/media/<path>                    返回 {"id","status":"queued"}；Idempotency-Key 相同且 body 相同 → 返回同一个 job；body 不同 → 409
- GET  /v1/media/jobs/<id>                 queued → running → completed（billing.state 先 pending 一次，再 settled）
- GET  /v1/media/jobs/<id>/content         未完成时 409
测试 key：pk-test（X-API-Key 或 Authorization: Bearer）
用法：python fake_pika.py   （监听 127.0.0.1:18480）"""
import hashlib, json, os, subprocess, tempfile, uuid
from flask import Flask, jsonify, request, send_file
app = Flask(__name__); JOBS, IDEM = {}, {}
CLIP = os.path.join(tempfile.gettempdir(), "fake_pika.mp4")

@app.before_request
def auth():
    if request.path.startswith("/files"): return None
    k = request.headers.get("X-API-Key") or request.headers.get("Authorization", "").removeprefix("Bearer ")
    if k != "pk-test": return jsonify(message="Invalid API key"), 401

@app.get("/billing/balance")
def balance(): return jsonify(balance_micro_usd=int(float(os.environ.get("FAKE_BALANCE_USD", "1.00")) * 1e6))

@app.post("/v1/media/<path:api>")
def submit(api):
    raw = request.get_data(); key = request.headers.get("Idempotency-Key")
    if key in IDEM:
        jid, h = IDEM[key]
        if h != hashlib.sha256(raw).hexdigest(): return jsonify(message="Idempotency-Key reused with a different body"), 409
        return jsonify(id=jid, status=JOBS[jid]["status"])
    jid = "media_" + str(uuid.uuid4()); JOBS[jid] = {"api": api, "n": 0, "status": "queued", "quote": json.loads(raw).get("_fake_charge", 200000)}
    if key: IDEM[key] = (jid, hashlib.sha256(raw).hexdigest())
    return jsonify(id=jid, status="queued")

@app.get("/v1/media/jobs/<jid>")
def job(jid):
    j = JOBS.get(jid)
    if not j: return jsonify(message="media job not found"), 404
    j["n"] += 1; j["status"] = ["queued", "running", "completed"][min(j["n"], 2)]
    out = {"id": jid, "status": j["status"]}
    if j["status"] == "completed":
        if not os.path.exists(CLIP): subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc=size=1280x720:rate=24:duration=5", "-pix_fmt", "yuv420p", CLIP], check=True)
        out.update(output={"media_type": "video", "video": {"url": f"{request.host_url}files/{jid}.mp4", "content_type": "video/mp4"}},
                   usage={"output_seconds": 5}, billing={"state": "pending"} if j["n"] == 3 else {"state": "settled", "charge_micro_usd": j["quote"]})
    return jsonify(out)

@app.get("/v1/media/jobs/<jid>/content")
def content(jid):
    j = JOBS.get(jid)
    if not j or j["status"] != "completed": return jsonify(message="media job is not ready"), 409
    return jsonify(url=f"{request.host_url}files/{jid}.mp4")

@app.get("/files/<name>")
def files(name): return send_file(CLIP, mimetype="video/mp4")

if __name__ == "__main__":
    app.run(port=18480)

test_budget.py：

import pika_budget_run as p
def test_local_schema_catches_bad_enum():
    assert "duration_s" in p.validate_local("pika/pika-2.5/image-to-video", {"image": "https://example.com/a.png", "duration_s": 7})
def test_local_schema_misses_cross_field_rule_but_quote_catches_it():
    body = {"images": ["https://example.com/a.png"]}
    assert p.validate_local("pika/pikaframes/image-to-video", body) is None
    price, err = p.quote("pika/pikaframes/image-to-video", body)
    assert price is None and "loop=true" in err

第 2 步：运行

$ python pika_budget_run.py jobs.json                     # 只报价（真实的公开 quote 接口）
✓ 任务 1（pika/pika-2.5/image-to-video）：报价 x.xx 美元
✓ 任务 2（pika/pikaframes/image-to-video）：报价 x.xx 美元
✗ 任务 3（pika/pikaframes/image-to-video）：报价被拒（HTTP 422）：Value error, A single keyframe requires loop=true
合计 2 个可提交任务，报价 x.xx 美元（未设置 PIKA_BUDGET_USD）
只报价，没有提交。确认后加 --submit。

# 下面几次的提交、余额、扣费都是模拟服务器的（PIKA_BASE=http://127.0.0.1:18480）
$ python pika_budget_run.py jobs.json --submit
✗ 提交前必须设置 PIKA_BUDGET_USD
$ PIKA_BUDGET_USD=<低于报价合计> python pika_budget_run.py jobs.json --submit
✗ 报价超出预算，没有提交
$ PIKA_BUDGET_USD=1 python pika_budget_run.py jobs.json --submit     # 模拟服务器余额设得很低
✗ 余额 x.xx 美元 不够，没有提交
$ PIKA_BUDGET_USD=1 python pika_budget_run.py jobs.json --submit     # 余额充足
  已提交 media_43e14ce4-…
  billing 还是 pending，稍后再查
  → {'id': 'media_43e14ce4-…', 'url': 'http://127.0.0.1:18480/files/…mp4', 'charge': …}
  已提交 media_b437c394-…
  → {'id': 'media_b437c394-…', 'url': …, 'charge': …}
  ⚠ 实际扣费 x.xx 美元 高于报价 x.xx 美元，请核对账单          ← 模拟服务器固定扣一个数，故意触发
实际记账 x.xx 美元（只计 settled 的 billing）

幂等提交的单独测试（模拟服务器）：同一个 Key 提交两次，返回同一个 media_d6dbe273-…；同一个 Key 换了 body，返回 409 Idempotency-Key reused with a different body。

调试示例

1. 本地 schema 校验通过了，报价却被拒。 第 3 个任务只有一张关键帧，没有设置 loop。本地 jsonschema.validate 认为没问题（minItems=1）。真实的 quote 接口返回：

HTTP 422 {"message":"Value error, A single keyframe requires loop=true"}

同样，3 张关键帧、transition_duration_s: 8 也被报价接口拒绝：“Durations above 5 seconds are only supported for a single transition”。这些跨字段规则只写在参数说明里，schema 表达不了。所以流程是先本地校验（快，查拼写和类型），再报价（免费，查真实规则），两步都通过才算这个任务可以提交。

2. 余额检查“失灵”了。 第一次测试余额不足时，我在运行客户端的命令前面设置了 FAKE_BALANCE_USD=0.10，结果任务照样提交了。其实这个变量是模拟服务器读取的，服务器启动时用的是默认余额。重启服务器并设置变量后，程序正确输出“✗ 余额不够，没有提交”。测试本身也会写错，所以检查结果要对照预期。

3. 记账只认 settled。 规格里写明，billing.state 为 pending 时要稍后再查，null（提交时的响应、webhook）不代表扣费。模拟服务器第一次返回 pending，程序等待后再查。实际扣费高于报价时，程序给出警告。

常见问题

模板规则

• 先报价，再提交；预算由人设定，不设预算就不提交。
• 金额用 micro-USD 整数计算。
• 只按 settled 记账，实际扣费和报价要对账。

项目资源

预付费的 API，第一道防线是报价和预算，之后才轮到写生成代码。