×

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

admin admin 发表于2026-06-18 15:22:49 浏览74 评论0

抢沙发发表评论

一、先选对接口:两条路,别走错


京东联盟接口 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
skuListSKU列表(颜色/尺码/配置)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次/分钟 → 批量+断点续传,稳。


群贤毕至

访客