京东工业（聚焦工业用品采购的 B2B 电商平台）的商品详情数据（如规格参数、批量价格、库存状态、供应商信息等）对企业采购决策、供应链管理、竞品分析等场景具有重要价值。由于平台无公开官方 API，开发者需通过页面解析或逆向工程实现商品详情（item_get）的获取。本文系统讲解接口逻辑、技术实现、工业场景适配及反爬应对，帮助构建稳定的京东工业商品详情获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（环境与 URL 结构）
三、接口调用流程（基于页面解析与动态接口）
四、代码实现示例（Python）
import requests
import time
import random
import re
import json
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import Dict, List
class JdIndustrialItemApi:
def __init__(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://b.jd.com/item/{product_id}.html"
self.api_url = "https://b.jd.com/industrial/product/detailAjax"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表
self.cookie = cookie # 企业账号登录态Cookie
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头"""
headers = {
"User-Agent": self.ua.random,
"Referer": "https://b.jd.com/",
"Accept": "application/json, text/javascript, */*; q=0.01",
"X-Requested-With": "XMLHttpRequest"
}
if self.cookie:
headers["Cookie"] = self.cookie
return headers
def _get_proxy(self) -> Dict[str, str]:
"""随机获取代理"""
if self.proxy_pool and len(self.proxy_pool) > 0:
proxy = random.choice(self.proxy_pool)
return {"http": proxy, "https": proxy}
return None
def _clean_price(self, price_str: str) -> float:
"""清洗价格（去除¥、/个等）"""
if not price_str:
return 0.0
price_str = re.sub(r"[^\d.]", "", price_str)
return float(price_str) if price_str else 0.0
def _parse_jsonp(self, jsonp_str: str) -> Dict:
"""解析JSONP响应为JSON"""
try:
json_str = re.search(r"jQuery\d+_\d+\((\{.*\})\)", jsonp_str).group(1)
return json.loads(json_str)
except Exception as e:
print(f"JSONP解析失败: {str(e)}")
return {}
def _parse_static_data(self, html: str) -> Dict:
"""解析主页面静态数据"""
soup = BeautifulSoup(html, "lxml")
# 提取规格选项
specs = []
for spec_item in soup.select("div.spec-items .spec-item"):
spec_name = spec_item.select_one(".spec-title")?.text.strip() or "规格"
spec_values = [
{
"name": option.text.strip(),
"spec_id": option.get("data-sku") or "", # 规格ID（多规格时使用）
"selected": "selected" in option.get("class", [])
}
for option in spec_item.select(".spec-value a")
]
if spec_values:
specs.append({
"name": spec_name,
"values": spec_values
})
return {
"title": soup.select_one("h1.sku-name")?.text.strip() or "",
"images": {
"main": [img.get("src") for img in soup.select("div.swiper-wrapper img") if img.get("src")],
"detail": [img.get("src") for img in soup.select("div.detail-img img") if img.get("src")]
},
"price": {
"single_str": soup.select_one("div.price .p-price")?.text.strip() or "",
"tax_tag": soup.select_one("div.tax-tag")?.text.strip() or ""
},
"supplier": {
"name": soup.select_one("div.seller-name")?.text.strip() or "",
"qualification": [q.text.strip() for q in soup.select("div.qualification-tags span")]
},
"purchase": {
"min_order": soup.select_one("div.min-order")?.text.strip() or "",
"service_tags": [s.text.strip() for s in soup.select("div.service-tags span")]
},
"specs": specs # 规格选项（如尺寸、型号）
}
def _fetch_api_data(self, product_id: str, spec_id: str = "", headers: Dict[str, str], proxy: Dict[str, str]) -> Dict:
"""调用动态API接口获取核心数据"""
api_data = {"price": {}, "stock": {}, "params": {}, "trade": {}}
try:
# 多规格时使用spec_id，否则用product_id
target_id = spec_id if spec_id else product_id
timestamp = int(time.time() * 1000)
# 生成随机callback参数（模拟前端）
callback = f"jQuery11240{random.randint(1000000000000, 9999999999999)}_{timestamp}"
params = {
"productId": target_id,
"callback": callback,
"_": timestamp
}
response = requests.get(
self.api_url,
params=params,
headers=headers,
proxies=proxy,
timeout=10
)
# 解析JSONP响应
data = self._parse_jsonp(response.text)
if not data:
return api_data
# 解析价格信息（含阶梯价）
price_info = data.get("price", {})
api_data["price"] = {
"singlePrice": price_info.get("singlePrice", 0),
"ladderPrices": price_info.get("ladderPrices", []), # 阶梯价列表
"taxPrice": price_info.get("taxPrice", 0)
}
# 解析库存信息
stock_info = data.get("stock", {})
api_data["stock"] = {
"totalStock": stock_info.get("totalStock", 0),
"warehouseStock": stock_info.get("warehouseStock", {}) # 区域库存
}
# 解析工业参数
api_data["params"] = data.get("params", {})
# 解析交易数据
trade_info = data.get("trade", {})
api_data["trade"] = {
"monthSales": trade_info.get("monthSales", 0),
"commentCount": trade_info.get("commentCount", 0),
"goodRate": trade_info.get("goodRate", 0)
}
except Exception as e:
print(f"API数据获取失败: {str(e)}")
return api_data
def _merge_multi_specs(self, static_specs: List[Dict], product_id: str, headers: Dict[str, str], proxy: Dict[str, str]) -> List[Dict]:
"""合并多规格商品的价格和库存"""
merged_specs = []
for spec_group in static_specs:
spec_name = spec_group["name"]
merged_values = []
for spec in spec_group["values"]:
spec_id = spec["spec_id"]
if not spec_id:
merged_values.append(spec)
continue
# 调用规格对应的API接口
spec_data = self._fetch_api_data(product_id, spec_id, headers, proxy)
merged_values.append({
**spec,** spec_data["price"],
"stock": spec_data["stock"]
})
merged_specs.append({
"name": spec_name,
"values": merged_values
})
return merged_specs
def item_get(self, product_id: str, timeout: int = 10) -> Dict:
"""
获取京东工业商品详情
:param product_id: 商品ID（如100012345678）
:param timeout: 超时时间
:return: 标准化商品数据
"""
try:
# 1. 主页面请求
url = self.base_url.format(product_id=product_id)
headers = self._get_headers()
proxy = self._get_proxy()
# 随机延迟，避免反爬
time.sleep(random.uniform(2, 4))
response = requests.get(
url=url,
headers=headers,
proxies=proxy,
timeout=timeout
)
response.raise_for_status()
main_html = response.text
# 2. 解析主页面数据
static_data = self._parse_static_data(main_html)
if not static_data["title"]:
return {"success": False, "error_msg": "商品不存在或已下架"}
# 3. 获取API核心数据（默认规格）
api_data = self._fetch_api_data(product_id, "", headers, proxy)
# 4. 处理多规格商品（若有）
merged_specs = static_data["specs"]
if static_data["specs"] and any(len(s["values"]) > 1 for s in static_data["specs"]):
merged_specs = self._merge_multi_specs(static_data["specs"], product_id, headers, proxy)
# 5. 整合结果
result = {
"success": True,
"data": {
"product_id": product_id,
**static_data,** api_data, # 合并price/stock/params/trade
"specs": merged_specs, # 多规格已合并价格库存
"url": url,
"update_time": time.strftime("%Y-%m-%d %H:%M:%S")
}
}
return result
except requests.exceptions.HTTPError as e:
if "403" in str(e):
return {"success": False, "error_msg": "触发反爬，建议更换代理或Cookie", "code": 403}
if "401" in str(e):
return {"success": False, "error_msg": "Cookie无效，请使用企业账号重新登录", "code": 401}
return {"success": False, "error_msg": f"HTTP错误: {str(e)}", "code": response.status_code}
except Exception as e:
return {"success": False, "error_msg": f"获取失败: {str(e)}", "code": -1}
# 使用示例
if __name__ == "__main__":
# 代理池（替换为有效代理）
PROXIES = [
"http://123.45.67.89:8888",
"http://98.76.54.32:8080"
]
# 企业账号登录态Cookie（从浏览器获取）
COOKIE = "session-id=xxx; user-key=xxx; pin=xxx; enterpriseId=xxx"
# 初始化API客户端
api = JdIndustrialItemApi(proxy_pool=PROXIES, cookie=COOKIE)
# 获取商品详情（示例product_id）
product_id = "100012345678" # 螺栓商品ID
result = api.item_get(product_id)
if result["success"]:
data = result["data"]
print(f"商品标题: {data['title']}")
print(f"价格信息: 单价¥{data['price']['singlePrice']}/个 | 含税价¥{data['price']['taxPrice']}/个")
if data['price']['ladderPrices']:
print("阶梯价:")
for ladder in data['price']['ladderPrices']:
print(f" 采购{ladder['count']}个及以上: ¥{ladder['price']}/个")
print(f"库存信息: 总库存{data['stock']['totalStock']}个")
if data['stock']['warehouseStock']:
print("区域库存:")
for warehouse, stock in data['stock']['warehouseStock'].items():
print(f" {warehouse}: {stock}个")
print(f"工业参数: 材质={data['params'].get('材质')} | 强度等级={data['params'].get('强度等级')} | 执行标准={data['params'].get('执行标准')}")
print(f"采购信息: {data['purchase']['min_order']} | 服务: {', '.join(data['purchase']['service_tags'])}")
print(f"供应商: {data['supplier']['name']} | 资质: {', '.join(data['supplier']['qualification'])}")
print(f"交易数据: 近30天成交{data['trade']['monthSales']}个 | 评价{data['trade']['commentCount']}条 | 好评率{data['trade']['goodRate']}%")
if data['specs']:
print(f"规格选项:")
for spec_group in data['specs'][:1]: # 第一个规格组（如尺寸）
print(f" {spec_group['name']}:")
for spec in spec_group['values'][:2]:
print(f" {spec['name']}: 单价¥{spec['singlePrice']}/个 | 库存{spec['stock']['totalStock']}个")
print(f"详情页: {data['url']}")
else:
print(f"获取失败: {result['error_msg']}（错误码: {result.get('code')}）")