本文将详细介绍如何使用 Python 调用京东商品详情 API 接口，涵盖官方开放平台 API、第三方代理接口以及网页逆向三种方案，包含完整的代码示例和最佳实践。
一、方案概述
1. 官方开放平台 API（推荐）
核心接口信息：
- 接口地址：
https://api.jd.com/routerjson - 请求方式：POST/GET
- 认证方式：OAuth2.0 + 签名验证（MD5/HMAC-SHA256）
2. 第三方代理 API
3. 网页逆向方案
通过分析京东网页接口（如
item.so.m.jd.com）直接获取数据。二、官方 API 方案详解
2.1 准备工作
- 注册开发者账号：访问 京东开放平台
- 创建应用：获取
AppKey和AppSecret - 申请权限：申请
jd.item.get接口权限 - 获取 Access Token：通过 OAuth2.0 授权流程获取
2.2 签名算法实现
京东 API 使用 MD5 签名验证，参数需按字典序排序：
Python
import hashlibimport urllib.parseimport jsonimport timefrom collections import OrderedDictclass JDAPISigner:
"""京东 API 签名工具类"""
@staticmethod
def generate_sign(params: dict, app_secret: str, sign_method: str = "md5") -> str:
"""
生成京东 API 请求签名
Args:
params: 请求参数（不含 sign）
app_secret: 应用密钥
sign_method: 签名方法，支持 md5 和 hmac-sha256
"""
# 按参数名升序排序
sorted_params = OrderedDict(sorted(params.items()))
# 拼接字符串：参数名+参数值
sign_str = app_secret for key, value in sorted_params.items():
if key != 'sign' and value is not None and value != '':
sign_str += f"{key}{value}"
sign_str += app_secret
# 生成签名
if sign_method.lower() == 'md5':
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
elif sign_method.lower() == 'hmac-sha256':
import hmac return hmac.new(
app_secret.encode('utf-8'),
sign_str.encode('utf-8'),
hashlib.sha256 ).hexdigest().upper()
else:
raise ValueError(f"不支持的签名方法: {sign_method}")
@staticmethod
def url_encode(params: dict) -> str:
"""URL 编码参数"""
encoded = []
for key, value in sorted(params.items()):
if value is not None:
encoded.append(f"{urllib.parse.quote(key)}={urllib.parse.quote(str(value))}")
return "&".join(encoded)2.3 完整调用代码
Python
import requestsimport jsonimport timeclass JDOpenAPI:
"""京东开放平台 API 客户端"""
def __init__(self, app_key: str, app_secret: str, access_token: str = None):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.base_url = "https://api.jd.com/routerjson"
self.signer = JDAPISigner()
def _build_params(self, method: str, api_params: dict) -> dict:
"""构建请求参数"""
params = {
'app_key': self.app_key,
'method': method,
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
'format': 'json',
'v': '1.0',
'sign_method': 'md5',
}
# 添加业务参数
if api_params:
params['360buy_param_json'] = json.dumps(api_params)
if self.access_token:
params['access_token'] = self.access_token
# 生成签名
params['sign'] = self.signer.generate_sign(params, self.app_secret)
return params
def get_item_detail(self, sku_id: str, fields: str = None) -> dict:
"""
获取商品详情
Args:
sku_id: 商品SKU ID
fields: 指定返回字段，逗号分隔
"""
api_params = {
'skuId': sku_id }
if fields:
api_params['fields'] = fields
params = self._build_params('jd.item.get', api_params)
try:
response = requests.post(self.base_url, data=params, timeout=30)
response.raise_for_status()
result = response.json()
# 处理京东 API 响应格式
if 'error_response' in result:
error = result['error_response']
print(f"API错误: {error.get('zh_desc', error.get('en_desc', '未知错误'))}")
return None
return result.get('jd_item_get_response', {})
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
def parse_item_info(self, data: dict) -> dict:
"""解析商品信息"""
if not data:
return None
item = data.get('item', {})
return {
'sku_id': item.get('skuId'),
'name': item.get('name'),
'brand': item.get('brand'),
'category': item.get('category'),
'price': item.get('price'),
'market_price': item.get('marketPrice'),
'stock': item.get('stock'),
'weight': item.get('weight'),
'image_url': item.get('imageUrl'),
'detail_url': f"https://item.jd.com/{item.get('skuId')}.html",
'status': item.get('status'), # 1:上架, 0:下架
'attributes': item.get('attributes', []),
'images': item.get('images', []),
'shop_id': item.get('shopId'),
'shop_name': item.get('shopName'),
}# 使用示例if __name__ == "__main__":
# 配置信息（实际应用中应从配置文件或环境变量读取）
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
ACCESS_TOKEN = "your_access_token"
SKU_ID = "100012043978" # 替换为实际商品SKU
# 初始化API客户端
api = JDOpenAPI(APP_KEY, APP_SECRET, ACCESS_TOKEN)
# 获取商品详情
raw_data = api.get_item_detail(
sku_id=SKU_ID,
fields="skuId,name,brand,price,stock,imageUrl,status"
)
if raw_data:
# 解析数据
item_info = api.parse_item_info(raw_data)
print(json.dumps(item_info, ensure_ascii=False, indent=2))三、第三方代理方案
3.1 方案特点
| 维度 | 官方 API | 第三方代理 |
|---|---|---|
| 认证门槛 | 需企业资质 | 无需资质 |
| 数据完整度 | 基础字段 | 更完整（含评论、促销等） |
| 调用成本 | 按量计费 | 按条计费 |
| 稳定性 | 高 | 中等 |
| 适用场景 | 企业级应用 | 个人/研究 |
3.2 第三方接口调用示例
Python
import requestsimport jsonclass JDThirdPartyAPI:
"""第三方京东商品详情 API 封装"""
def __init__(self, api_key: str, api_secret: str, base_url: str = "https://api-gw.onebound.cn/jd"):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = base_url
def get_item_detail(self, num_iid: str) -> dict:
"""
获取商品详情
Args:
num_iid: 商品ID（SKU ID）
"""
url = f"{self.base_url}/item_get"
params = {
'key': self.api_key,
'secret': self.api_secret,
'num_iid': num_iid }
try:
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if data.get('error') != '0':
print(f"API错误: {data.get('error', '未知错误')}")
return None
return data.get('item', {})
except requests.exceptions.RequestException as e:
print(f"请求异常: {e}")
return None
def extract_key_fields(self, item_data: dict) -> dict:
"""提取关键字段"""
if not item_data:
return None
return {
'title': item_data.get('title'),
'price': item_data.get('price'),
'original_price': item_data.get('orginal_price'),
'num_iid': item_data.get('num_iid'),
'detail_url': item_data.get('detail_url'),
'pic_url': item_data.get('pic_url'),
'brand': item_data.get('brand'),
'category': item_data.get('rootCatId'),
'cid': item_data.get('cid'),
'desc': item_data.get('desc'),
'item_imgs': item_data.get('item_imgs', []),
'sku': item_data.get('sku', []),
'props_list': item_data.get('props_list', {}),
'seller_id': item_data.get('seller_id'),
'shop_name': item_data.get('shop_name'),
'sales': item_data.get('sales'),
'comment_count': item_data.get('comment_count'),
'star': item_data.get('star'),
}# 使用示例if __name__ == "__main__":
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
NUM_IID = "10335871600" # 商品ID
api = JDThirdPartyAPI(API_KEY, API_SECRET)
item_data = api.get_item_detail(NUM_IID)
if item_data:
clean_data = api.extract_key_fields(item_data)
print(json.dumps(clean_data, ensure_ascii=False, indent=2))四、网页逆向方案
4.1 移动端 H5 接口
京东移动端 H5 页面提供 JSON 数据接口，无需签名验证：
Python
import requestsimport jsonimport reclass JDWebCrawler:
"""京东网页爬虫"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Referer': 'https://so.m.jd.com/',
})
def get_item_from_h5(self, sku_id: str) -> dict:
"""
通过 H5 接口获取商品详情
Args:
sku_id: 商品SKU ID
"""
# H5 详情页接口
url = f"https://item.so.m.jd.com/ware/detail.json"
params = {
'wareId': sku_id,
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
# 提取商品信息
ware_info = data.get('wareInfo', {})
return {
'sku_id': sku_id,
'title': ware_info.get('wname'),
'price': ware_info.get('jdPrice', {}).get('p'),
'original_price': ware_info.get('jdPrice', {}).get('op'),
'stock': ware_info.get('stockDesc'),
'brand': ware_info.get('brand'),
'category': ware_info.get('category'),
'images': ware_info.get('wareImage', []),
'detail': ware_info.get('wdis'),
'shop_id': ware_info.get('shopId'),
'shop_name': ware_info.get('shopName'),
'score': ware_info.get('score'),
'comments': ware_info.get('commentCount'),
}
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
def get_item_from_pc(self, sku_id: str) -> dict:
"""
通过 PC 端页面解析获取商品详情（需要处理反爬）
Args:
sku_id: 商品SKU ID
"""
url = f"https://item.jd.com/{sku_id}.html"
try:
response = self.session.get(url, timeout=10)
response.raise_for_status()
# 从页面中提取商品数据
html = response.text
# 提取标题
title_match = re.search(r'<div class="sku-name">(.*?)</div>', html, re.DOTALL)
title = re.sub(r'<[^>]+>', '', title_match.group(1)).strip() if title_match else None
# 提取价格（需要从价格接口单独获取）
# 京东 PC 端价格通过异步接口加载：p.3.cn/prices/mgets
return {
'sku_id': sku_id,
'title': title,
'detail_url': url,
}
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
def get_price(self, sku_ids: list) -> dict:
"""
获取商品价格（京东价格接口）
Args:
sku_ids: SKU ID 列表
"""
# 京东价格接口
url = "https://p.3.cn/prices/mgets"
# 构建 SKU 参数
sku_params = ','.join([f"J_{sid}" for sid in sku_ids])
params = {
'skuIds': sku_params,
'type': 1,
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
price_map = {}
for item in data:
sku_id = item.get('id', '').replace('J_', '')
price_map[sku_id] = {
'price': item.get('p'), # 京东价
'market_price': item.get('m'), # 市场价
'plus_price': item.get('tpp'), # Plus会员价
}
return price_map
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return {}五、关键注意事项
5.1 频率控制
京东对 API 和网页请求有严格限流，建议：
| 接口类型 | 频率限制 | 建议延迟 |
|---|---|---|
| 官方 API | 按应用等级分配 | 遵循返回的限流头 |
| 网页 H5 | 10-30 次/分钟 | 2-5 秒/请求 |
| 价格接口 | 较高限制 | 1-2 秒/请求 |
Python
import timeimport randomclass RateLimiter:
"""请求频率限制器"""
def __init__(self, min_delay: float = 1.0, max_delay: float = 3.0):
self.min_delay = min_delay
self.max_delay = max_delay
self.last_request_time = 0
def wait(self):
"""等待随机时间"""
elapsed = time.time() - self.last_request_time if elapsed < self.min_delay:
sleep_time = random.uniform(self.min_delay - elapsed, self.max_delay - elapsed)
time.sleep(max(0, sleep_time))
self.last_request_time = time.time()# 使用示例limiter = RateLimiter(min_delay=2.0, max_delay=5.0)def safe_request(func):
"""请求装饰器"""
def wrapper(self, *args, **kwargs):
if hasattr(self, 'rate_limiter'):
self.rate_limiter.wait()
return func(self, *args, **kwargs)
return wrapper5.2 错误处理
| 错误码 | 含义 | 处理建议 |
|---|---|---|
isv.item-not-exist | 商品不存在 | 检查 SKU ID 是否正确 |
isv.invalid-parameter | 参数错误 | 检查必填参数格式 |
isv.invalid-signature | 签名错误 | 检查签名算法和密钥 |
isv.access-token-invalid | Token 失效 | 重新获取 Access Token |
isv.rate-limit | 频率超限 | 降低请求频率，启用队列 |
5.3 合规建议
- 遵守 robots.txt：京东禁止高频抓取商品详情页
- 数据使用范围：仅限内部使用，不得转售或公开发布
- 用户授权：获取店铺数据需获得商家明确授权
- 反爬应对：网页方案需处理验证码、IP 封禁等问题
六、数据存储与扩展
6.1 数据库存储示例
Python
import sqlite3from datetime import datetimeimport jsonclass JDDataStorage:
"""京东商品数据存储"""
def __init__(self, db_path: str = "jd_products.db"):
self.conn = sqlite3.connect(db_path)
self._init_tables()
def _init_tables(self):
"""初始化数据表"""
# 商品主表
self.conn.execute("""
CREATE TABLE IF NOT EXISTS jd_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sku_id TEXT UNIQUE NOT NULL,
title TEXT,
brand TEXT,
category TEXT,
price REAL,
market_price REAL,
stock INTEGER,
shop_id TEXT,
shop_name TEXT,
status INTEGER DEFAULT 1,
raw_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 价格历史表
self.conn.execute("""
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sku_id TEXT NOT NULL,
price REAL,
market_price REAL,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (sku_id) REFERENCES jd_items(sku_id)
)
""")
self.conn.commit()
def save_item(self, item: dict):
"""保存商品信息"""
try:
self.conn.execute("""
INSERT INTO jd_items
(sku_id, title, brand, category, price, market_price, stock, shop_id, shop_name, status, raw_data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(sku_id) DO UPDATE SET
title=excluded.title,
brand=excluded.brand,
category=excluded.category,
price=excluded.price,
market_price=excluded.market_price,
stock=excluded.stock,
shop_id=excluded.shop_id,
shop_name=excluded.shop_name,
status=excluded.status,
raw_data=excluded.raw_data,
updated_at=CURRENT_TIMESTAMP
""", (
item.get('sku_id'),
item.get('title'),
item.get('brand'),
item.get('category'),
item.get('price'),
item.get('market_price'),
item.get('stock'),
item.get('shop_id'),
item.get('shop_name'),
item.get('status', 1),
json.dumps(item, ensure_ascii=False)
))
self.conn.commit()
# 记录价格历史
if item.get('price'):
self.conn.execute("""
INSERT INTO price_history (sku_id, price, market_price)
VALUES (?, ?, ?)
""", (item.get('sku_id'), item.get('price'), item.get('market_price')))
self.conn.commit()
except Exception as e:
print(f"保存失败: {e}")
self.conn.rollback()
def get_item_history(self, sku_id: str, days: int = 30):
"""获取商品价格历史"""
cursor = self.conn.execute("""
SELECT price, market_price, recorded_at
FROM price_history
WHERE sku_id = ?
AND recorded_at > datetime('now', '-{} days')
ORDER BY recorded_at ASC
""".format(days), (sku_id,))
return [{
'price': row[0],
'market_price': row[1],
'date': row[2]
} for row in cursor.fetchall()]七、总结
本文介绍了三种获取京东商品详情的方案：
| 方案 | 适用场景 | 难度 | 稳定性 | 数据完整度 |
|---|---|---|---|---|
| 官方 API | 企业级应用 | 中 | 高 | 基础字段 |
| 第三方代理 | 快速验证/个人 | 低 | 中 | 完整 |
| 网页逆向 | 研究学习 | 高 | 低 | 完整 |
