3 种集成模式

按业务规模选择最合适的对接方式。

模式 1：同步查询（单条/少量，实时返回）

适用场景：电商订单详情页、客服查询界面、用户输入即查。

sequenceDiagram
    participant C as 客户端
    participant T as TrackEasy API
    participant L as 物流商 (USPS/UPS/...)

    C->>T: POST /api/v1/trackings/create {tracking_number: "..."}
    T->>L: 异步查询（后台线程）
    T-->>C: 200 OK {id, status: "查询中"}
    Note over C,T: 等待 5-15 秒
    C->>T: GET /api/v1/trackings/get?tracking_numbers=...
    T-->>C: 200 OK {success: [{status, events: [...]}]}

代码示例（Python）

import requests, time

API = "https://api.168tracking.com"
KEY = "tk_your_key"
H = {"X-API-Key": KEY, "Content-Type": "application/json"}

def track_one(tracking_number):
    # 1) 注册
    r = requests.post(f"{API}/api/v1/trackings/create",
        headers=H, json={"tracking_number": tracking_number})
    r.raise_for_status()
    data = r.json()["data"]

    # 2) 等几秒
    time.sleep(8)

    # 3) 查询轨迹
    r = requests.get(f"{API}/api/v1/trackings/get",
        headers=H, params={"tracking_numbers": tracking_number})
    r.raise_for_status()
    return r.json()["data"]["success"][0]

print(track_one("9205512345600001234567"))

模式 2：异步批量（几十~几千条，结果后台写入）

适用场景：电商订单导出、定时同步 ERP、跨境批量发货。

sequenceDiagram
    participant C as 客户端
    participant T as TrackEasy API
    participant L as 物流商

    C->>T: POST /api/v1/track/async {tracking_numbers: [...]}
    T-->>C: 200 OK {accepted: 40, poll_url: "..."}
    Note over T,L: 后台并发查询（每单 5-15s）
    C->>T: GET /api/v1/shipments?status=查询中&limit=50
    T-->>C: {shipments: [{id, status, ...}]}
    Note over C: 轮询直到 status 全为 已签收/异常

代码示例（Python）

import requests, time

API = "https://api.168tracking.com"
KEY = "tk_your_key"
H = {"X-API-Key": KEY, "Content-Type": "application/json"}

def batch_track(tracking_numbers, chunk_size=40):
    """分片批量提交（每片 ≤40）"""
    for i in range(0, len(tracking_numbers), chunk_size):
        chunk = tracking_numbers[i:i+chunk_size]
        r = requests.post(f"{API}/api/v1/track/async",
            headers=H, json={"tracking_numbers": chunk})
        r.raise_for_status()
        print(f"提交 {len(chunk)} 单: {r.json()['data']}")

def poll_until_done(timeout=300):
    """轮询直到所有 tracking 完成"""
    start = time.time()
    while time.time() - start < timeout:
        r = requests.get(f"{API}/api/v1/shipments",
            headers=H, params={"status": "查询中", "limit": 200})
        pending = r.json()["data"]["shipments"]
        if not pending:
            print("✅ 全部完成")
            return
        print(f"⏳ 还有 {len(pending)} 单查询中...")
        time.sleep(15)

nums = ["9205512345600001234567", "1Z999AA10123456784", ...]
batch_track(nums)
poll_until_done()

模式 3：Webhook 推送（长期监控，自动接收）

适用场景：长期监控订单状态、SaaS 应用通知、电商平台订阅推送。

sequenceDiagram
    participant C as 客户端系统
    participant T as TrackEasy API
    participant L as 物流商

    C->>T: POST /api/v1/webhooks {url: "https://...", events: [...]}
    T-->>C: 200 OK {secret: "whsec_..."}
    C->>T: POST /api/v1/trackings/create {tracking_number: "..."}
    T->>L: 定期轮询（5 分钟/次）
    Note over T,L: 轨迹更新
    T->>C: POST https://your-server.com/webhook
    Note over C: HMAC 验签 → 写库 → 用户可查

代码示例（Python — Flask 接收端）

from flask import Flask, request, abort
import hmac, hashlib, time

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret_from_create_response"

@app.post("/webhook/trackeasy")
def handle():
    sig_header = request.headers.get("X-Webhook-Signature", "")
    ts = request.headers.get("X-Webhook-Timestamp", "")
    body = request.get_data()

    # Stripe-style 签名验证
    expected = "t=" + ts + ",v1=" + hmac.new(
        WEBHOOK_SECRET.encode(),
        (ts + ".").encode() + body,
        hashlib.sha256
    ).hexdigest()

    # 防 replay：5 分钟 drift
    if abs(time.time() - int(ts)) > 300:
        abort(400, "timestamp drift")

    if not hmac.compare_digest(sig_header, expected):
        abort(401, "bad signature")

    # 处理事件
    event = request.json
    print(f"📦 {event['event']}: {event['data']['tracking_number']} → {event['data']['status']}")

    # 2xx 表示成功，否则会重试 14 次
    return "", 200

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

详细签名 / 重试机制见 Webhook 规格。

怎么选？