多店铺京东订单统一拉取的核心在于：每个店铺独立 AccessToken + 增量时间窗拉取
jingdong.pop.order.search+ 以 (tid, shop_id)为唯一键做数据库幂等 + 分布式锁/Redis NX 防多节点并发重复处理 + 断点续跑防漏单。下面给你一套可直接落地的 Python 实战源码。一、整体架构
┌────────────────────┐ │ 分布式调度器 │ APScheduler / Celery Beat │ 遍历 shop_auth 表 │ └───────┬────────────┘ │ 每店铺独立 token ▼ ┌────────────────────┐ │ JdOrderSyncPerShop │ │ 1. 获取分布式锁 │ Redis SET shop_id:sync NX EX │ 2. 增量拉取订单 │ start_modified ~ end_modified │ 3. 逐单查明细 │ jd.order.detail.get │ 4. 幂等入库 │ INSERT ON CONFLICT(tid,shop_id) DO NOTHING │ 5. 更新断点时间 │ last_sync_time │ 6. 释放锁 │ └────────────────────┘
二、核心源码（Python · 含签名 / 分布式锁 / 幂等 / 断点）
# jd_multi_shop_order_sync.py
"""
多店铺京东订单统一拉取实战
- JOS MD5签名（秒级timestamp）
- Redis分布式锁（防止多节点同时跑同一店铺）
- 增量同步 modified 时间窗
- 幂等入库 (tid, shop_id) UNIQUE
- 断点续跑（记录 last_sync_time）
依赖:
pip install requests redis apscheduler
"""
import hashlib
import json
import time
import datetime
import requests
import redis
from typing import Dict, List
# ================== 配置 ==================
APP_KEY = "YOUR_JD_APP_KEY"
APP_SECRET = "YOUR_JD_APP_SECRET"
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
SYNC_INTERVAL_MIN = 30 # 增量窗口分钟数
LOCK_TTL_SEC = 300 # 分布式锁过期时间
# ==========================================
rds = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
# ---------- JOS Client ----------
class JdJosClient:
GW = "https://api.jd.com/routerjson"
def __init__(self, app_key, app_secret):
, self.ask = app_key, app_secret
def _sign(self, params: Dict) -> str:
filt = sorted(
(k, v) for k, v in params.items()
if k != "sign" and v is not None and str(v).strip() != ""
)
qs = "".join(f"{k}{v}" for k, v in filt)
return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()).hexdigest().upper()
def call(self, method, biz, token):
p = {
"app_key": self.ak,
"method": method,
"timestamp": str(int(time.time())), # 秒级！
"format": "json",
"v": "2.0",
"sign_method": "md5",
"360buy_param_json": json.dumps(biz, ensure_ascii=False,
separators=(',', ':')),
"access_token": token
}
p["sign"] = self._sign(p)
r = requests.post(self.GW, data=p, timeout=15)
r.raise_for_status()
d = r.json()
resp_key = method.replace(".", "_") + "_response"
if resp_key not in d:
for k in d:
if k.endswith("_response"):
resp_key = k
break
data = d.get(resp_key, d)
if "error_response" in str(data):
err = d.get(resp_key, {}).get("error_response") or d.get("error_response")
if err:
raise Exception(f"JOS[{err.get('code')}]: {err.get('zh_desc')}")
return data
def list_orders(self, token, start, end, page=1, page_size=100):
return self.call("jingdong.pop.order.search", {
"start_modified": start,
"end_modified": end,
"order_state": "WAIT_SELLER_STOCK_OUT,WAIT_BUYER_CONFIRM_GOODS,FINISHED",
"page": page,
"page_size": min(page_size, 100)
}, token)
def order_detail(self, token, order_id):
return self.call("jd.order.detail.get", {
"orderId": str(order_id),
"fields": "orderId,status,payment,buyer_nick,receiver_name,"
"receiver_mobile,orders.skuId,orders.outerSkuId,"
"orders.num,orders.price,created,modified"
}, token)
# ---------- 分布式锁 ----------
def acquire_lock(shop_id: str) -> str:
"""Redis SET NX EX，返回 identifier 或 None"""
identifier = str(time.time())
lock_key = f"jd_sync_lock:{shop_id}"
ok = rds.set(lock_key, identifier, nx=True, ex=LOCK_TTL_SEC)
return identifier if ok else None
def release_lock(shop_id: str, identifier: str):
lock_key = f"jd_sync_lock:{shop_id}"
# Lua等效：仅当值匹配才删，防误释放
with rds.pipeline() as pipe:
while True:
try:
pipe.watch(lock_key)
if pipe.get(lock_key) == identifier.encode():
pipe.multi()
pipe.delete(lock_key)
pipe.execute()
break
except redis.WatchError:
continue
# ---------- 断点 & 幂等（SQLite示例，生产换 MySQL/PG） ----------
def init_db(db="jd_erp.db"):
conn = __import__("sqlite3").connect(db)
conn.execute("""
CREATE TABLE IF NOT EXISTS erp_order(
tid TEXT,
shop_id TEXT,
status TEXT,
payment REAL,
buyer_nick TEXT,
created TEXT,
PRIMARY KEY(tid, shop_id)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS shop_sync_point(
shop_id TEXT PRIMARY KEY,
last_sync_time TEXT
)
""")
conn.commit()
return conn
def load_last_time(conn, shop_id):
cur = conn.execute("SELECT last_sync_time FROM shop_sync_point WHERE shop_id=?",
(shop_id,))
row = cur.fetchone()
return row[0] if row else None
def save_last_time(conn, shop_id, t):
conn.execute("""
INSERT INTO shop_sync_point(shop_id,last_sync_time)
VALUES(?,?) ON CONFLICT(shop_id) DO UPDATE SET last_sync_time=excluded.last_sync_time
""", (shop_id, t))
conn.commit()
def upsert_order(conn, shop_id, order):
conn.execute("""
INSERT INTO erp_order(tid,shop_id,status,payment,buyer_nick,created)
VALUES(?,?,?,?,?,?)
ON CONFLICT(tid,shop_id) DO NOTHING
""", (
str(order.get("orderId")),
shop_id,
order.get("status"),
float(order.get("payment") or 0),
order.get("buyer_nick", ""),
order.get("created", "")
))
conn.commit()
# ---------- 单店铺同步 ----------
def sync_one_shop(client: JdJosClient, conn, shop: Dict):
shop_id = shop["shop_id"]
token = shop["access_token"]
lock_id = acquire_lock(shop_id)
if not lock_id:
print(f"⚠️ 店铺[{shop_id}] 正在其他节点同步，跳过")
return
try:
now = datetime.datetime.now()
last = load_last_time(conn, shop_id)
if last:
start = last
else:
start = (now - datetime.timedelta(minutes=SYNC_INTERVAL_MIN * 2)
).strftime("%Y-%m-%d %H:%M:%S")
end = now.strftime("%Y-%m-%d %H:%M:%S")
print(f"▶ 店铺[{shop_id}] 增量 {start} ~ {end}")
page = 1
while True:
rsp = client.list_orders(token, start, end, page=page)
order_list = (rsp.get("popOrderSearch", {})
.get("orderSearch", {})
.get("orderInfoList", [])) or []
total = rsp.get("popOrderSearch", {}).get("orderSearch", {}
).get("orderTotal", 0)
for o in order_list:
tid = str(o.get("orderId"))
# 明细（按需，列表已有基础字段可先upsert）
detail = client.order_detail(token, tid)
order_data = detail.get("orderInfo", o)
upsert_order(conn, shop_id, order_data)
if len(order_list) < 100 or page * 100 >= total:
break
page += 1
time.sleep(0.15) # QPS保护
# 成功更新断点（用当前end，下次从此继续）
save_last_time(conn, shop_id, end)
print(f"✅ 店铺[{shop_id}] 同步完成，断点更新至 {end}")
except Exception as e:
print(f"❌ 店铺[{shop_id}] 同步异常: {e}")
finally:
release_lock(shop_id, lock_id)
# ---------- 多店铺入口 ----------
def sync_all_shops(shop_list: List[Dict]):
client = JdJosClient(APP_KEY, APP_SECRET)
conn = init_db()
for shop in shop_list:
try:
sync_one_shop(client, conn, shop)
except Exception as e:
print(f"❌ 顶层异常 shop={shop.get('shop_id')}: {e}")
# ===================== 示例 shop_list =====================
if __name__ == "__main__":
SHOP_LIST = [
{"shop_id": "SHOP001", "access_token": "SELLER_TOKEN_A"},
{"shop_id": "SHOP002", "access_token": "SELLER_TOKEN_B"},
]
sync_all_shops(SHOP_LIST)
# ---- 常驻调度示例（取消注释）----
# from apscheduler.schedulers.blocking import BlockingScheduler
# sched = BlockingScheduler()
# sched.add_job(lambda: sync_all_shops(SHOP_LIST),
# 'interval', minutes=SYNC_INTERVAL_MIN, id='jd_multi_shop_sync')
# sched.start()三、关键设计点说明
- 多店铺隔离每个店铺一条
shop_auth记录，存shop_id / access_token / refresh_token，同步时按shop_id独立跑、独立锁、独立断点时间。 - 分布式锁（Redis SET NX EX）防止多节点 / 多进程同时跑同一店铺导致重复调用API、重复入库。
- 幂等设计本地订单表 PRIMARY KEY
(tid, shop_id)，处理前数据库层ON CONFLICT DO NOTHING，即使重试/重复推送也不重复建单。 - 增量 + 断点续跑记录
last_sync_time，每次用start_modified=last_sync_time, end_modified=now拉取，服务重启从中继续，避免漏单与超日额度。 - QPS保护分页间
time.sleep(0.15)+ 令牌桶（前文京东限流篇已给），遇code=16 / ISP_FLOW_CONTROL指数退避重试。 - 对账兜底每日凌晨全量比对一次最近 N 天订单，修正极端情况下漏掉的状态变更（回调丢失/网络抖动）。
四、避坑清单
坑 | 现象 | 解决 |
|---|---|---|
用毫秒时间戳 | Invalid Timestamp | JOS用秒级 int(time.time()) |
订单403 | 传买家token / 个人应用 | 企业认证 + 卖家OAuth AccessToken |
多节点重复入库 | 无幂等/无锁 | (tid,shop_id)唯一索引 + Redis锁 |
重启后重复拉全量 | 无断点记录 | 持久化 last_sync_time |
超日额度 | limited-by-app-access-count | 增量同步 + 断点，避免全量翻页 |
一句话总结：多店铺京东订单同步 = 每店铺独立 Seller Token → 增量 modified 时间窗 → Redis NX 锁防并发 →
(tid,shop_id)幂等入库 → 记录 last_sync_time 断点续跑 → 日对账兜底，签名/QPS 同单店铺，仅调度与隔离维度不同。