开始之前
- 已有一物一码模块使用权限（qcode.manage）
- 外部服务有公网可达的 HTTPS 地址
操作步骤
事件类型一共 9 种
订阅键：scan（扫码）、form_submit（表单提交）、status_change（状态流转）、item_voided（作废）、approved（审批通过）、rejected（审批驳回）、returned（退回）、inspection_due（巡检到期）、inspection_overdue（巡检超期）。按需订阅，不要全开 —— 每个事件都会单独发一次 HTTP 请求。
在系统里建 Webhook
「一物一码 → Webhook 回调」新建：填 URL、勾选 events、填 secret（强烈建议填）。填完可在「一物一码 → Webhook 投递日志」看每次投递的状态与响应码。
三个请求头，一个都不能漏
X-Qcode-Signature：hex(HMAC_SHA256(secret, raw_body))，未配 secret 时这个头不存在。X-Qcode-Event：事件名。X-Qcode-Delivery：本次投递的 UUID，做幂等用（重试时 UUID 不变）。
验签必须用原始 body
HMAC 算的是收到的原始字节流。如果你先把 JSON 反序列化再重新序列化去算签名，几乎一定不匹配（键顺序、空格、中文转义都会变）。正确做法：拿到 raw bytes → 先验签 → 再 json.loads。
2xx 才算成功，必须 10 秒内响应
超时 10 秒。建议收到后立刻返回 200 并在后台异步处理，不要在请求里做耗时业务（写库、调第三方）。响应慢了会被判失败并触发重试，导致同一事件被处理多次。
重试策略要接受「可能重复」
失败后按 30×2^n 退避重试，最多 3 次；重试耗尽保持 failed。另有兜底扫描任务每 10 分钟扫一次 pending 记录重新入队。因为存在重试和兜底补偿，你的处理逻辑必须幂等 —— 用 X-Qcode-Delivery 去重。
配置变更 60 秒才生效
Webhook 配置有 60 秒进程内缓存。新建、改地址、停用后不要立刻测试然后怀疑没配好，等一分钟。
代码示例
import hmac, hashlib, json
from fastapi import FastAPI, Request, Header, HTTPException, BackgroundTasks
app = FastAPI()
SECRET = b"your-webhook-secret"
SEEN: set[str] = set() # 生产环境换成 Redis 或数据库表
def verify(raw: bytes, signature: str) -> bool:
expect = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
return hmac.compare_digest(expect, signature)
@app.post("/hooks/qcode")
async def qcode_hook(
request: Request,
background: BackgroundTasks,
x_qcode_signature: str = Header(default=""),
x_qcode_event: str = Header(default=""),
x_qcode_delivery: str = Header(default=""),
):
raw = await request.body() # 必须用原始字节验签
if not verify(raw, x_qcode_signature):
raise HTTPException(status_code=401, detail="bad signature")
if x_qcode_delivery in SEEN: # 重试去重
return {"ok": True, "dup": True}
SEEN.add(x_qcode_delivery)
payload = json.loads(raw.decode("utf-8"))
background.add_task(handle_event, x_qcode_event, payload)
return {"ok": True} # 立刻返回，耗时逻辑丢后台
def handle_event(event: str, payload: dict):
if event == "scan":
code = payload.get("code") or payload.get("qcode")
print("扫码:", code, payload)
elif event == "inspection_overdue":
print("巡检超期:", payload)BODY='{"code":"QC0001","event":"scan"}'
SIG=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac 'your-webhook-secret' | sed 's/^.*= //')
curl -X POST http://127.0.0.1:9000/hooks/qcode \
-H 'Content-Type: application/json' \
-H "X-Qcode-Signature: $SIG" \
-H 'X-Qcode-Event: scan' \
-H 'X-Qcode-Delivery: test-0001' \
-d "$BODY"示例中的域名、租户编码、密钥请替换成你自己的。生产环境不要把密钥写进前端代码。