一、先选对接口:两条路,别走错
京东联盟接口 jd.union.open.goods.detail.query | 京东开放平台 jingdong.item.read.get | |
|---|---|---|
| 适用场景 | 选品分析、比价工具、CPS推广 | 商家自有ERP、店铺数据同步 |
| 门槛 | 低,个人/企业均可 | 高,需企业认证+店铺授权 |
| 价格 | ✅ 到手价/券后价/佣金 | ✅ 原价/会员价/成本价 |
| 销量 | ✅ monthSales 月销量 | ✅ 更细粒度的销售数据 |
| SKU | ✅ 规格列表 | ✅ 含真实库存/区域库存 |
| 推荐 | 大多数开发者首选这个 | 商家自用选这个 |
结论:做批量采集,优先走京东联盟接口,接入快、数据够用。
二、接入全流程(5步落地)
第1步:注册 & 拿凭证
京东联盟 → https://union.jd.com └─ 注册 → 实名认证 → 创建应用 → 审核通过 └─ 获得:AppKey + AppSecret(务必保密)
第2步:获取 AccessToken(OAuth 2.0)
pythonimport requestsdef get_access_token(app_key, app_secret): url = 'https://api.jd.com/oauth2/access_token' params = { 'grant_type': 'client_credentials', 'client_id': app_key, 'client_secret': app_secret } resp = requests.post(url, data=params).json() return resp['access_token']⚠️
access_token有效期 2小时,过期需刷新,别存死。
第3步:生成签名(核心!最多人栽在这里)
京东联盟用 HMAC-SHA256,不是MD5:
pythonimport hashlibimport timedef generate_sign(params, app_secret): """params 不包含 sign 字段,按 key 字典序排序""" sorted_params = sorted(params.items()) query_str = app_secret for k, v in sorted_params: query_str += f"{k}{v}" query_str += app_secret return hashlib.sha256(query_str.encode('utf-8')).hexdigest().upper()常见签名错误:
| 错误现象 | 原因 |
|---|---|
{"code":10001,"message":"无效签名"} | 排序错了 / app_secret 拼接位置错了 / 没转大写 |
{"code":13,"message":"频率超限"} | 超过 100次/分钟,加 time.sleep(0.6) |
第4步:发起请求(单个商品)
pythondef get_goods_detail(sku_id, app_key, app_secret, access_token): url = 'https://api.jd.com/routerjson' timestamp = str(int(time.time() * 1000)) params = { 'method': 'jd.union.open.goods.detail.query', 'app_key': app_key, 'access_token': access_token, 'timestamp': timestamp, 'format': 'json', 'v': '2.0', 'sign_method': 'hmac-sha256', 'param_json': {'skuId': sku_id} # 商品ID,从URL中提取 } params['sign'] = generate_sign(params, app_secret) resp = requests.post(url, json=params).json() return resp第5步:批量获取(重点)
批量策略 = 分页 + 并发控制 + 断点续传
pythonimport timeimport jsonfrom concurrent.futures import ThreadPoolExecutor, as_completeddef batch_fetch(sku_id_list, app_key, app_secret, access_token, max_workers=5): """批量获取商品详情,自动控制频率""" results = [] semaphore = time.sleep # 简单限流:每次请求间隔0.6秒 def fetch_one(sku_id): semaphore(0.6) # 限速 ~100次/分钟 try: data = get_goods_detail(sku_id, app_key, app_secret, access_token) if data.get('code') == '0': goods = data['jd_union_open_goods_detail_query_responce']['result']['goodsInfo'] return { 'sku_id': goods['skuId'], 'title': goods.get('title', ''), 'price': goods.get('price', 0), # 原价 'low_price': goods.get('lowPrice', 0), # 到手价 'coupon_price': goods.get('couponPrice', 0), # 券后价 'month_sales': goods.get('monthSales', 0), # 月销量 'comment_count': goods.get('commentCount', 0), 'is_self': goods.get('isSelf', 0), # 是否自营 'stock_state': goods.get('stockState', 0), # 有货/无货 'sku_list': goods.get('skuList', []), # SKU规格列表 } except Exception as e: print(f"Error fetching {sku_id}: {e}") return None with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {executor.submit(fetch_one, sid): sid for sid in sku_id_list} for future in as_completed(futures): result = future.result() if result: results.append(result) # 每100条存一次,防止中断全丢 if len(results) % 100 == 0: with open('jd_goods_batch.json', 'w') as f: json.dump(results, f, ensure_ascii=False, indent=2) return results# 执行sku_list = ['100012345678', '100012345679', ...] # 你的商品ID列表access_token = get_access_token(APP_KEY, APP_SECRET)data = batch_fetch(sku_list, APP_KEY, APP_SECRET, access_token)print(f"完成,共获取 {len(data)} 条")三、核心字段速查表(你要的全在这)
| 字段 | 含义 | 获取方式 |
|---|---|---|
price | 京东原价 | goodsInfo.price |
lowPrice | 到手价(促销后) | goodsInfo.lowPrice |
couponPrice | 券后价 | goodsInfo.couponPrice |
monthSales | 月销量 | goodsInfo.monthSales |
commentCount | 评价数 | goodsInfo.commentCount |
goodCommentRate | 好评率 | goodsInfo.goodCommentRate |
skuList | SKU列表(颜色/尺码/配置) | goodsInfo.skuList,每个含 specId+specName |
stockState | 有货(1)/无货(0) | goodsInfo.stockState |
isSelf | 是否自营(1/0) | goodsInfo.isSelf |
SKU数据结构示例:
json{ "skuId": "1000123456789", "specId": "12345", "specName": "颜色:黑色/内存:256GB", "price": 5999.00, "stockState": 1}四、进阶:获取店铺全部商品
如果你要拉取某店铺所有商品,用搜索接口 + 分页:
pythondef fetch_shop_all_goods(shop_id, app_key, app_secret, access_token): """拉取店铺全部商品,自动翻页""" all_goods = [] page_no = 1 page_size = 100 # 京东API上限 while True: params = { 'method': 'jingdong.goods.search', 'app_key': app_key, 'access_token': access_token, 'timestamp': str(int(time.time() * 1000)), 'format': 'json', 'v': '2.0', 'shop_id': shop_id, 'page_no': page_no, 'page_size': page_size, } params['sign'] = generate_sign(params, app_secret) resp = requests.post('https://api.jd.com/routerjson', json=params).json() result = resp['jingdong_goods_search_responce']['result'] goods_list = result['goods_list'] total = result['total'] all_goods.extend(goods_list) print(f"第{page_no}页: {len(goods_list)}条, 累计{len(all_goods)}") total_pages = (total + page_size - 1) // page_size if page_no >= total_pages: break page_no += 1 time.sleep(0.6) return all_goods五、避坑清单(90%的人踩过)
| 坑 | 解决方案 |
|---|---|
| 签名永远报错 | 严格按 key字典序排序,app_secret 前后各拼一次 |
| 频率超限被封 | 控制在 ≤100次/分钟,加 time.sleep(0.6) |
| access_token 过期 | 每次请求前检查,过期则重新获取,缓存2小时内的token |
| SKU里没库存数 | 联盟接口只返回有货/无货,真实库存要走开放平台接口 |
| 月销量为0 | 部分新品无销量数据,属正常,不是接口问题 |
| 商品ID提取错 | URL是 item.jd.com/100012345678.html → ID是 100012345678 |
六、快速启动模板
把下面复制下来,替换 APP_KEY / APP_SECRET / SKU_ID 就能跑:
pythonimport requests, hashlib, time, jsonAPP_KEY = '你的AppKey'APP_SECRET = '你的AppSecret'SKU_ID = '100012345678' # 商品IDdef sign(params): s = sorted(params.items()) q = APP_SECRET + ''.join(f'{k}{v}' for k,v in s) + APP_SECRET return hashlib.sha256(q.encode()).hexdigest().upper()def get_token(): r = requests.post('https://api.jd.com/oauth2/access_token', data={'grant_type':'client_credentials','client_id':APP_KEY,'client_secret':APP_SECRET}).json() return r['access_token']def get_detail(sku_id, token): p = { 'method':'jd.union.open.goods.detail.query', 'app_key':APP_KEY, 'access_token':token, 'timestamp':str(int(time.time()*1000)), 'format':'json','v':'2.0','sign_method':'hmac-sha256', 'param_json':json.dumps({'skuId':sku_id}) } p['sign'] = sign(p) r = requests.post('https://api.jd.com/routerjson', json=p).json() return r['jd_union_open_goods_detail_query_responce']['result']['goodsInfo']token = get_token()goods = get_detail(SKU_ID, token)print(f"{goods['title']} | ¥{goods['lowPrice']} | 月销{goods['monthSales']}")一句话总结:走京东联盟接口 → OAuth拿token → HMAC-SHA256签名 → 控制100次/分钟 → 批量+断点续传,稳。