长时间的生成任务不必一直轮询：Pika 会主动回调 Webhook

长时间的生成任务不必一直轮询：Pika 会在任务完成或失败时主动回调你的 webhook。本篇写一个接收端：按 Standard Webhooks 规范用原始请求字节验签，按 webhook-id 去重（重试和重放都会复用这个 id），把结果记录到 SQLite。另外写一个本地发送脚本，模拟正常、重放、失败、篡改、过期五种情况。

先看清成品（box 上实测，完全离线，不需要 Pika Key）

• 测试： pytest 2 项通过。

• 五种情况：

  情况	结果
  正常完成事件	200，已记录
  同一个 webhook-id 重放	200，标记为重复，忽略
  失败事件	200，记录错误码 content_moderation
  body 被篡改	400 “No matching signature found”
  时间戳是 10 分钟前	400 “Message timestamp too old”

• 数据库： 只有 2 条记录。

没有在 box 上跑的部分： 没有收到过 Pika 的真实回调。真实回调要求地址是公网可访问的 HTTPS（443 端口），不能是内网地址。本篇的接收端只监听 127.0.0.1，上线时要放在反向代理或隧道后面，并且要 HTTPS。签名逻辑使用 Pika 文档推荐的 standardwebhooks 库，测试中的签名由本地的 secret 生成。

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

做完之后你会得到什么？

需要准备哪些环境？

文件结构

48-pika-C-src/
├── webhook_receiver.py
├── send_test_events.py
└── test_verify.py

第 1 步：代码

webhook_receiver.py：

"""接收 Pika 的 webhook（任务完成/失败时 Pika 主动回调），验签、去重后记录结果。
文档（dev.pika.art/llms/webhooks）：
  - 签名遵循 Standard Webhooks：请求头 webhook-id / webhook-timestamp / webhook-signature
  - 必须用“原始请求字节”验签，重新序列化的 JSON 验不过
  - 重试和重放会复用 webhook-id → 按 webhook-id 去重；投递不保证顺序，用 data.id 对应任务
  - 事件：media.job.completed、media.job.failed（看 data.error.code）、webhook.test（控制台的“发送测试”）
  - webhook 里的 billing 永远是 null，不能用来记账（记账见 48-pika-B）
  - 回调地址必须是公网可访问的 HTTPS 443 端口：本地开发要放在反向代理 / 隧道后面，本脚本只监听 127.0.0.1
环境变量：PIKA_WEBHOOK_SECRET（whsec_ 开头，在控制台 Webhooks 页或 GET /v1/webhooks/secret 获取）、DB（默认 webhooks.db）
用法：python webhook_receiver.py   （监听 127.0.0.1:18490，路径 /pika/webhooks）"""
import json, os, sqlite3, sys
from flask import Flask, request
from standardwebhooks import Webhook, WebhookVerificationError

app = Flask(__name__)
SECRET = os.environ.get("PIKA_WEBHOOK_SECRET") or sys.exit("请先设置 PIKA_WEBHOOK_SECRET")
wh = Webhook(SECRET)
db = sqlite3.connect(os.environ.get("DB", "webhooks.db"), check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS events (webhook_id TEXT PRIMARY KEY, type TEXT, job_id TEXT, status TEXT, video_url TEXT, error_code TEXT, received_at TEXT DEFAULT CURRENT_TIMESTAMP)")

@app.post("/pika/webhooks")
def receive():
    raw = request.get_data()                       # 原始字节，不要用 request.get_json() 再 dumps
    try: event = wh.verify(raw, dict(request.headers))
    except WebhookVerificationError as e:
        print(f"✗ 验签失败：{e}", flush=True); return {"error": "invalid signature"}, 400
    wid = request.headers["webhook-id"]
    d = event.get("data") or {}
    try:
        db.execute("INSERT INTO events (webhook_id, type, job_id, status, video_url, error_code) VALUES (?,?,?,?,?,?)",
                   (wid, event["type"], d.get("id"), d.get("status"), ((d.get("output") or {}).get("video") or {}).get("url"), (d.get("error") or {}).get("code")))
        db.commit()
    except sqlite3.IntegrityError:
        print(f"↷ {wid} 已处理过（重试/重放），忽略", flush=True); return {"ok": True, "duplicate": True}
    print(f"✓ {event['type']} job={d.get('id')} status={d.get('status')}" + (f" error={d['error']['code']}" if d.get("error") else ""), flush=True)
    return {"ok": True}                            # 尽快返回 2xx；下载视频等耗时操作放到后台任务里做

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=18490)

send_test_events.py：

"""本地测试：用同一个 secret 按 Standard Webhooks 签名，模拟 Pika 发几种回调（正常、重放、篡改、过期、重新序列化）。
用法：PIKA_WEBHOOK_SECRET=whsec_... python send_test_events.py"""
import json, os, time, uuid
from datetime import datetime, timezone
import requests
from standardwebhooks import Webhook

URL = os.environ.get("URL", "http://127.0.0.1:18490/pika/webhooks")
wh = Webhook(os.environ["PIKA_WEBHOOK_SECRET"])

def send(label, payload, wid=None, ts=None, tamper=False):
    wid = wid or f"msg_{uuid.uuid4()}"; ts = ts or datetime.now(timezone.utc)
    body = json.dumps(payload, ensure_ascii=False)
    sig = wh.sign(wid, ts, body)
    if tamper: body = body.replace("completed", "completed ")
    r = requests.post(URL, data=body.encode(), headers={"content-type": "application/json", "webhook-id": wid,
                      "webhook-timestamp": str(int(ts.timestamp())), "webhook-signature": sig}, timeout=10)
    print(f"{label:<28} → HTTP {r.status_code} {r.text.strip()}"); return wid

job = {"id": "media_demo-1", "status": "completed", "model_id": "pika/pika-2.5/image-to-video", "runtime_s": 42.7,
       "output": {"media_type": "video", "video": {"url": "https://api.dev.pika.art/v1/files/output.mp4", "content_type": "video/mp4"}},
       "error": None, "usage": {"video_output_tokens": 50638}, "billing": None}
ev = {"type": "media.job.completed", "timestamp": datetime.now(timezone.utc).isoformat(), "data": job}
w = send("1 正常的完成事件", ev)
send("2 同一个 webhook-id 重放", ev, wid=w)
send("3 失败事件", {**ev, "type": "media.job.failed", "data": {**job, "id": "media_demo-2", "status": "failed", "output": None, "error": {"code": "content_moderation", "message": "…"}}})
send("4 body 被篡改", ev, tamper=True)
send("5 时间戳是 10 分钟前", ev, ts=datetime.fromtimestamp(time.time() - 600, timezone.utc))

test_verify.py：

import base64, json, os
from datetime import datetime, timezone
import pytest
from standardwebhooks import Webhook, WebhookVerificationError
SECRET = "whsec_" + base64.b64encode(b"0123456789abcdef0123456789abcdef").decode()

def signed(body):
    w = Webhook(SECRET); ts = datetime.now(timezone.utc)
    return {"webhook-id": "msg_1", "webhook-timestamp": str(int(ts.timestamp())), "webhook-signature": w.sign("msg_1", ts, body)}

def test_raw_bytes_verify():
    body = '{"type": "media.job.completed", "data": {"id": "m1"}}'
    assert Webhook(SECRET).verify(body.encode(), signed(body))["data"]["id"] == "m1"

def test_reserialized_json_fails():
    body = '{"type": "media.job.completed", "data": {"id": "m1"}}'
    h = signed(body); again = json.dumps(json.loads(body), separators=(",", ":"))
    with pytest.raises(WebhookVerificationError): Webhook(SECRET).verify(again.encode(), h)

第 2 步：运行

$ export PIKA_WEBHOOK_SECRET="whsec_$(head -c 24 /dev/urandom | base64)"    # 本地测试用；正式使用填控制台给的密钥
$ python webhook_receiver.py &
$ python send_test_events.py
1 正常的完成事件                    → HTTP 200 {"ok":true}
2 同一个 webhook-id 重放          → HTTP 200 {"duplicate":true,"ok":true}
3 失败事件                       → HTTP 200 {"ok":true}
4 body 被篡改                   → HTTP 400 {"error":"invalid signature"}
5 时间戳是 10 分钟前                → HTTP 400 {"error":"invalid signature"}

# 接收端日志
✓ media.job.completed job=media_demo-1 status=completed
↷ msg_f73030a0-… 已处理过（重试/重放），忽略
✓ media.job.failed job=media_demo-2 status=failed error=content_moderation
✗ 验签失败：No matching signature found
✗ 验签失败：Message timestamp too old

检查点：数据库里只有 2 条记录（completed 和 failed），重放的那次没有重复写入。

调试示例

1. 重新序列化的 JSON 验不过。 Flask 里最顺手的写法是 request.get_json()，再用 json.dumps 转回字符串去验签。test_verify.py 的第二个测试复现了这个错误：同样的内容，只是序列化方式不同（这里是去掉了空格），签名就对不上了，抛出 WebhookVerificationError。Pika 文档也专门强调要传原始字节，所以代码里用的是 request.get_data()。

2. 后台日志是空的。 接收端用 nohup 放到后台运行，测试全部通过，日志文件里却什么都没有。原因是 Python 的标准输出在不连接终端时是块缓冲的。给 print 加上 flush=True 之后，日志就实时写入了。如果用 systemd 或 Docker 运行，也可以设置 PYTHONUNBUFFERED=1。

常见问题

模板规则

• 验签用原始字节，并且用规范的库。
• 以 webhook-id 去重，以 data.id 对应任务。
• 回调处理要快：收到后马上返回 2xx，耗时的工作交给后台。

项目资源

有了验签和去重，webhook 才能放心交给生产环境，不用担心伪造的或重复的回调。