虽然标题是《仿电商ERP：Spring Boot整合京东JOS...》，但你明确要求了附Python源码，下面我给你一套“架构思路（Spring Boot视角）+ 可直接跑的Python实现”，聚焦商品 / 订单 / 库存三件套自动同步，方便你对照理解或先用Python做原型验证。
一、仿ERP同步整体架构（对照Spring Boot思路）
在Java Spring Boot里通常是分层设计，Python这边可以一一对应：
- 配置层：
AppKey / AppSecret / AccessToken统一管理→ 对应@ConfigurationProperties - 平台适配层：京东JOS签名、网关、鉴权封装→ 对应
JdClient / JosTemplate - 业务同步层：商品、订单、库存增量拉取 + 本地落库→ 对应
XxxSyncService - 调度层：定时任务（APScheduler / Spring
@Scheduled）→ 对应ScheduledExecutorService - 幂等与重试：订单号去重、异常重试、分页游标→ 对应 数据库唯一索引 +
@Retryable
二、京东JOS通用API Client（Python，含MD5签名）
京东JOS特点：
- 网关：
https://api.jd.com/routerjson - 签名：MD5(AppSecret + 排序KV + AppSecret) → 大写
timestamp是秒级（10位），别用毫秒
# jd_jos_client.py
import hashlib
import time
import requests
from typing import Dict
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class JdJosClient:
GATEWAY = "https://api.jd.com/routerjson"
def __init__(self, app_key: str, app_secret: str, access_token: str = ""):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
def _sign(self, params: Dict) -> str:
# 剔除 sign、空值
filtered = sorted(
(k, v) for k, v in params.items()
if k != "sign" and v is not None and str(v).strip() != ""
)
kv = "".join(f"{k}{v}" for k, v in filtered)
raw = self.app_secret + kv + self.app_secret
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def execute(self, method: str, biz_params: Dict) -> dict:
params = {
"app_key": self.app_key,
"method": method,
"timestamp": str(int(time.time())),
"format": "json",
"v": "2.0",
"sign_method": "md5",
**biz_params,
}
if self.access_token:
params["access_token"] = self.access_token
params["sign"] = self._sign(params)
resp = requests.post(self.GATEWAY, data=params, timeout=15)
resp.raise_for_status()
return resp.json()三、三件套同步示例（商品 / 订单 / 库存）
1️⃣ 商品同步（SKU读取）
对应JOS接口：
jingdong.ware.sku.readdef sync_products(client: JdJosClient, sku_ids: str):
"""
sku_ids: 逗号分隔的skuId字符串
"""
result = client.execute("jingdong.ware.sku.read", {
"sku_id": sku_ids,
"field": "sku_id,outer_id,stock_num,price,ware_name"
})
data = result.get("jingdong_ware_sku_read_responce", {})
skus = data.get("skus", [])
for sku in skus:
print({
"sku_id": sku.get("sku_id"),
"name": sku.get("ware_name"),
"price": sku.get("price"),
"stock": sku.get("stock_num"),
})2️⃣ 订单同步（增量分页拉取）
对应接口：
jd.order.search/ jingdong.pop.order.searchdef sync_orders(client: JdJosClient, start_time: str, end_time: str, page: int = 1):
"""
start_time / end_time: yyyy-MM-dd HH:mm:ss
"""
result = client.execute("jingdong.pop.order.search", {
"start_date": start_time,
"end_date": end_time,
"order_state": "WAIT_SELLER_STOCK_OUT,WAIT_BUYER_CONFIRM_GOODS,FINISHED",
"page": page,
"page_size": 100,
})
orders = result.get("jingdong_pop_order_search_responce", {}).get("order_list", [])
for o in orders:
print({
"order_id": o.get("order_id"),
"state": o.get("order_state"),
"amount": o.get("order_total_price"),
})
return len(orders)ERP实践：
modified时间做增量order_id做唯一索引防重
3️⃣ 库存同步（读取 + 回写思路）
- 读取：复用
jingdong.ware.sku.read里的stock_num - 回写（ERP→京东）：用库存更新接口（需对应权限），一般按
sku_id + outer_id映射本地仓库库存
伪代码思路：
def sync_stock_to_jd(client, sku_id, new_stock):
# 实际接口以京东文档为准，示例仅为结构
client.execute("jingdong.ware.stock.update", {
"sku_id": sku_id,
"quantity": new_stock,
})四、定时任务（仿 @Scheduled）
from apscheduler.schedulers.blocking import BlockingScheduler
def job():
client = JdJosClient("APP_KEY", "APP_SECRET", "ACCESS_TOKEN")
sync_products(client, "123456,789012")
sync_orders(client, "2026-07-13 00:00:00", "2026-07-14 23:59:59")
if __name__ == "__main__":
scheduler = BlockingScheduler()
scheduler.add_job(job, "interval", minutes=30, id="jd_sync_job")
scheduler.start()五、常见坑（京东JOS特有）
- 时间戳用秒：京东是10位秒级，淘宝才是毫秒，混用必报签名错误
- 参数名ASCII排序：必须严格排序再拼串
- 订单接口要access_token：商品公开读可能不用，但订单/库存一定需要OAuth授权
- 分页+增量：不要全量扫历史订单，用
modified时间窗口 + 页游标
如果你愿意，我可以帮你把这套 Python 同步脚本改成标准 Spring Boot 分层结构（Config / Client / Service / Scheduled）的Java示例代码，方便你直接往ERP项目里搬。
