京东商品信息批量获取实战指南(2026最新版)

140
Category: 
开发交流

一、先选对接口:两条路,别走错
京东联盟接口 jd.union.open.goods.detail.query 京东开放平台 jingdong.item.read.get
适用场景 选品分析、比价工具、CPS推广 商家自有ERP、店铺数据同步
门槛 低,个人/企业均可 高,需企业认证+店铺授权
价格 ✅ 到手价/券后价/佣金 ✅ 原价/会员价/成本价
销量 ✅ monthSales 月销量 ✅ 更细粒度的销售数据
SKU ✅ 规格列表 ✅ 含真实库存/区域库存
推荐 大多数开发者首选这个 商家自用选这个
结论:做批量采集,优先走京东联盟接口,接入快、数据够用。

二、接入全流程(5步落地)
第1步:注册 & 拿凭证
京东联盟 →https://o0b.cn/lafite
└─ 注册 → 实名认证 → 创建应用 → 审核通过
└─ 获得:AppKey + AppSecret(务必保密)
第2步:获取 AccessToken(OAuth 2.0)
python
import requests

def 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:

python
import hashlib
import time

def 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步:发起请求(单个商品)
python
def 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步:批量获取(重点)
批量策略 = 分页 + 并发控制 + 断点续传

python
import time
import json
from concurrent.futures import ThreadPoolExecutor, as_completed

def 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
}
四、进阶:获取店铺全部商品
如果你要拉取某店铺所有商品,用搜索接口 + 分页:

python
def 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 就能跑:

python
import requests, hashlib, time, json

APP_KEY = '你的AppKey'
APP_SECRET = '你的AppSecret'
SKU_ID = '100012345678' # 商品ID

def 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次/分钟 → 批量+断点续传,稳。

Comments 0
/ 1000
0
0
Favorite