企业官网建设流程全解析

《京东订单API jingdong.pop.order.search 落地方案：鼎内调用与QPS配额管理》（附 Python 源码）

一、先拍结论

jingdong.pop.order.search是POP 商家订单检索接口，不是订单真相源（真相是jd.order.detail.get/ 订单消息）。

它的生产定位是：“按状态+时间窗捞订单列表”，再拿orderId去拉详情。

官方已经明说：

• 创建时间支持查近 2 年，按更新时间只支持近 3 个月
• 开始/结束时间不得相差超过 1 个月
• “建议通过时间滚动的方式拉取订单，页数要尽量少”​
• 云鼎内调用是官方标注的调用方式之一

所以京东订单同步的成熟写法不是“翻页翻到天荒地老”，而是：

时间窗滚动（30天/窗） → pop.order.search 捞 orderId → jd.order.detail.get 拉真相 → (order_id, order_state, operate_time) 做幂等 → JMQ/订单消息做触发器，search 做兜底

二、接口边界（别和这些搞混）

POP 和 VOP（企业采购）权限完全隔离，别用同一个 AppKey 想通吃。

三、时间窗与分页：官方给的“反模式”

官方文档原话要点：

• start_date/end_date差值不得超过 1 个月
• WAIT_SELLER_STOCK_OUT时起止可空，默认前一个月
• 其他状态start_date必传
• page_size最大 100
• “建议您之前频繁的翻页，页数较大，请尽快调整为时间滚动方式”​

❌ 错误写法

page=1..5000，固定最近30天，order_state=ALL

问题：

• 大商家一窗几万单 → 翻页深 → 超时/限流
• 订单在窗口内被修改 → 翻页过程中顺序漂移 → 漏单
• 状态 ALL + 深翻页 = 京东最喜欢限你流

✅ 正确写法

按 order_state 分桶 每个状态：时间窗 30 天，sortType=0 升序，page 不超 50 窗滚动：end = now - 30s，start = end - 30d 每次同步记录 (state, last_operate_time) 下次从 last_operate_time 继续

四、鼎内调用是什么

京东有“云鼎/鼎内”概念：应用/服务在京东云鼎环境内调用，某些订单/数据/合规链路更稳，官方接口文档也会标“云鼎内调用”。

生产含义：

• 鼎内：延迟低、订单/隐私字段合规链路更完整、大促抖动小
• 鼎外（公网 IDC）：能调，但隐私字段/大促稳定性/审计不如鼎内
• 订单类接口必须卖家 OAuth access_token，不管鼎内鼎外

经验口径（非官方保证）：

京东商家 JOS 订单类接口，鼎内通常比鼎外更适合做“持续同步 + 大促兜底”；

但“鼎内”不是免限流金牌，QPS 配额照样按应用/权限包来。

五、QPS 与配额：社区经验口径

官方不把“所有应用的 QPS”写死成一刀切，社区实测大致是：

费用口径（社区经验）：

• 商家 JOS 有日免费额度
• pop.order.search超量按量计费
• 鼎内/云内单价低于公网（类似淘宝聚石塔内外价差逻辑）
• 联盟 API（jd.union.open.*）是 CPS 选品体系，和商家订单同步是两回事

六、Python：京东 JOS 基础 Client（秒级时间！）

# jd/jos_client.py import hashlib import time import requests # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex class JdJosClient: GATEWAY = "https://api.jd.com/routerjson" def __init__(self, app_key, app_secret, access_token=None, timeout=5): self.app_key = app_key self.app_secret = app_secret self.access_token = access_token self.timeout = timeout def _sign(self, params: dict) -> str: flat = {k: v for k, v in params.items() if v is not None and k != "sign"} qs = "".join(f"{k}{v}" for k, v in sorted(flat.items())) raw = f"{self.app_secret}{qs}{self.app_secret}" return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() def execute(self, method: str, biz: dict): params = { "app_key": self.app_key, "method": method, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), # 京东用秒级/字符串 "format": "json", "v": "2.0", "sign_method": "md5", } if self.access_token: params["access_token"] = self.access_token params.update(biz) params["sign"] = self._sign(params) resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout) return resp.json()

京东坑：把 1688 的毫秒时间戳带过来 →timestamp invalid。

京东要yyyy-MM-dd HH:mm:ss，允许误差 10 分钟。

七、Python：时间滚动订单检索器

# jd/pop_order_scanner.py from datetime import datetime, timedelta from ratelimit.token_bucket import TokenBucket # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex class PopOrderScanner: MAX_WINDOW = timedelta(days=30) SLIDE_BACK = timedelta(seconds=30) # 防修改时间漂移漏单 def __init__(self, client: JdJosClient, bucket: TokenBucket): self.client = client self.bucket = bucket def scan_state(self, order_state: str, start: datetime, end: datetime): while start < end: window_end = min(start + self.MAX_WINDOW, end) yield from self._scan_window(order_state, start, window_end) # 下一窗往前滚，并回退 30s 防丢 start = window_end - self.SLIDE_BACK def _scan_window(self, order_state, start, end, page_size=100): page = 1 while True: self.bucket.acquire() resp = self.client.execute("jingdong.pop.order.search", { "start_date": start.strftime("%Y-%m-%d %H:%M:%S"), "end_date": end.strftime("%Y-%m-%d %H:%M:%S"), "order_state": order_state, "dateType": 0, # 0=按创建时间；1=按修改时间 "sortType": 0, # 0=升序，滚动拉取稳 "page": str(page), "page_size": str(page_size), "optional_fields": ( "orderId,orderState,operateTime,orderStartTime," "itemInfoList,payType,orderSellerPrice,consigneeInfo" ), }) node = resp.get("jingdong_pop_order_search_responce") or {} result = node.get("searchorderinfo_result") or {} api_result = result.get("apiResult") or {} if api_result.get("success") == "false": raise JdApiError(api_result.get("chineseErrCode"), api_result.get("numberCode")) orders = result.get("orderInfoList") or [] if not orders: break for o in orders: yield o # 官方建议：别深翻页 if len(orders) < page_size or page >= 50: break page += 1

状态分桶示例：

STATES = [ "NOT_PAY", "WAIT_SELLER_STOCK_OUT", "WAIT_GOODS_RECEIVE_CONFIRM", "FINISHED_L", "TRADE_CANCELED", "LOCKED", ]

八、QPS 配额管理器（单实例 + Redis 分布式）

单实例令牌桶

# ratelimit/token_bucket.py import time class TokenBucket: def __init__(self, rate: float, capacity: float = None, jitter: float = 0.05): self.rate = rate self.capacity = capacity or rate self.tokens = self.capacity self.last = time.monotonic() self.jitter = jitter def acquire(self, n: int = 1): while True: now = time.monotonic() self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate) self.last = now if self.tokens >= n: self.tokens -= n return sleep = (n - self.tokens) / self.rate + self.jitter time.sleep(sleep)

多实例分布式限流

# ratelimit/redis_sliding.py import time class RedisQpsLimiter: def __init__(self, redis, key: str, qps: int): self.r = redis self.key = key self.qps = qps def allow(self) -> bool: now = time.time() self.r.zremrangebyscore(self.key, 0, now - 1) self.r.zadd(self.key, {str(now): now}) self.r.expire(self.key, 2) return self.r.zcard(self.key) <= self.qps

调用侧：

if not redis_limiter.allow(): time.sleep(0.1) continue bucket.acquire()

规则：

• 企业 10 QPS → 客户端设 8
• 鼎内企业 20 QPS → 客户端设 16
• 多店铺：按shop_id分桶，别全店共享一个令牌桶
• 列表 8 QPS + 详情 5 QPS 分开桶，别互相挤死

九、限流/错误分类（京东特有）

class JdApiError(Exception): def __init__(self, msg, code=None): super().__init__(msg) self.code = code def classify(resp): node = resp.get("jingdong_pop_order_search_responce", {}) api = (node.get("searchorderinfo_result") or {}).get("apiResult") or {} if api.get("success") == "false": msg = api.get("chineseErrCode", "") code = api.get("numberCode", "") # 限额/限流 if "accesscontrol" in msg or "limited" in msg or str(code) == "16": raise RetriableError(f"jd limit: {msg}") # 参数/权限 if "permission" in msg or "validate" in msg: raise FatalError(f"jd param/perm: {msg}") raise RetriableError(f"jd api err: {msg}") # HTTP 层 # code=16 / 429 / ISP_FLOW_CONTROL_LIMIT → 退避

重试策略：

• code=16/ 429 / flow control → 指数退避（10s, 30s, 60s）
• token 过期 → 刷 access_token 再试 1 次
• 参数错 / 无权限 / 店铺未授权 → 不重试，告警
• 网络超时 → 重试，但别把“超时”当成“没数据”

十、幂等与状态机

pop.order.search只给列表，别直接写业务：

def upsert_from_search(order): order_id = str(order["orderId"]) state = order["orderState"] op_time = parse_time(order.get("operateTime") or order.get("orderStartTime")) cur = order_repo.get(order_id) version = int(op_time.timestamp() * 1000) if cur and version < cur.version: return # 旧窗数据晚到 # 只存“列表级快照”，详情再去 jd.order.detail.get 拉 order_repo.upsert({ "order_id": order_id, "order_state": state, "operate_time": op_time, "version": version, "need_detail": True, })

详情拉取：

def fill_detail(order_id): resp = client.execute("jd.order.detail.get", {"order_id": order_id}) trade = extract_trade(resp) order_repo.update_detail(trade)

十一、消息 + 搜索 双通道（生产必做）

JMQ 订单消息 └─> 入队 order_change(order_id) 定时器（每 60s） └─> pop.order.search 按状态滚动 └─> 补消息漏掉的 order_id 两者都只干一件事： mark need_detail = True Worker： jd.order.detail.get → 写 order_snapshot → 跑状态机

铁律：

• 消息：快返回、msgId 去重、异步处理
• 搜索：兜底、时间滚动、别深翻页
• 详情：真相源
• (order_id, order_state, operate_time)：幂等键

十二、大促/店铺多时的配置

JD_POP_CONFIG = { "bucket_qps": 8, # 企业 10 留余量 "detail_qps": 5, "window_days": 30, "slide_back_sec": 30, "max_page": 50, "states": STATES, "per_shop_bucket": True, "use_jmq": True, "search_fallback_interval": 60, # 秒 "dingnei_preferred": True, # 鼎内网关/部署优先 }

鼎内部署时：

• 订单同步服务放鼎内容器
• Redis/DB 也走鼎内内网
• 公网只暴露“内部调度 API”，不暴露 JOS 调用层
• 大促前压测的是detail + 发货 + 面单，不是 search 翻页

十三、落地检查表

• [ ] 时间戳用yyyy-MM-dd HH:mm:ss，别用毫秒
• [ ]pop.order.search只捞列表，不写业务
• [ ] 时间窗 ≤ 30 天，状态分桶，回退 30s
• [ ] page 不超 50，不深翻页
• [ ] 列表 QPS / 详情 QPS / 面单 QPS 分桶
• [ ] 多实例用 Redis 滑动窗口，不用单机 time.sleep 假装限流
• [ ] 限流退避、token 过期刷一次、权限错告警
• [ ] JMQ 做触发器，search 做兜底，detail 做真相
• [ ](order_id, order_state, operate_time)幂等
• [ ] POP / FBP / VOP 应用分开，不串 AppKey
• [ ] 鼎内优先，但鼎内也不假设“无限 QPS”

十四、一句话收口

jingdong.pop.order.search不是“订单同步主接口”，是“订单 ID 捕捞网”。

生产里把它当主角 → 深翻页、限流、漏单、大促崩；

把它当配角 → 时间滚动捞 ID、消息做闹钟、detail 做法官、QPS 桶做安全带。

鼎内调用解决的是合规和稳定性，不解决“你不会限流”的问题。