京东工业（专注工业用品 B2B 采购的电商平台）的工业品搜索功能（item_industries接口，非官方命名）是获取特定品类或关键词工业品列表的核心入口，数据包含批量价格、规格参数、供应商资质等关键信息，对企业采购选品、供应链比价、竞品分析等场景具有重要价值。由于平台无公开官方 API，开发者需通过页面解析或逆向工程实现搜索对接。本文系统讲解接口逻辑、参数解析、技术实现及工业场景适配策略，助你构建稳定的京东工业商品列表获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（参数与 URL 结构）
三、接口调用流程（基于页面解析与动态接口）
四、代码实现示例（Python）
import requests
import time
import random
import re
import json
import urllib.parse
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import List, Dict
class JdIndustrialSearchApi:
def __init__(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://b.jd.com/search/"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表，如["http://ip:port", ...]
self.cookie = cookie # 企业账号登录态Cookie
# 分类ID映射（简化版）
self.category_map = {
"紧固件-螺栓": "131946_131950",
"气动工具-风批": "131960_131962",
"轴承-深沟球轴承": "131970_131971"
}
# 供应商ID映射（简化版）
self.vendor_map = {
"京东自营": "1000000123",
"XX工业供应链": "1000000456"
}
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头"""
headers = {
"User-Agent": self.ua.random,
"Referer": "https://b.jd.com/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
}
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 _clean_stock(self, stock_str: str) -> int:
"""清洗库存（提取数字）"""
if not stock_str:
return 0
stock_num = re.search(r"\d+", stock_str)
return int(stock_num.group()) if stock_num else 0
def _clean_sales(self, sales_str: str) -> int:
"""清洗销量（提取数字）"""
if not sales_str:
return 0
sales_num = re.search(r"\d+", sales_str)
return int(sales_num.group()) if sales_num else 0
def _parse_item(self, item_soup) -> Dict[str, str]:
"""解析单条商品数据"""
# 提取商品ID
product_id = item_soup.get("data-sku") or ""
# 提取规格参数（转换为字典）
params_text = item_soup.select_one(".product-params")?.text.strip() or ""
params = {}
if params_text:
for param in params_text.split("|"):
if "：" in param:
key, value = param.split("：", 1)
params[key.strip()] = value.strip()
# 提取资质标签
qualification_tags = [tag.text.strip() for tag in item_soup.select(".qualification-tags span")]
return {
"product_id": product_id,
"title": item_soup.select_one(".product-title")?.text.strip() or "",
"main_image": item_soup.select_one(".product-img img")?.get("src") or "",
"url": f"https://b.jd.com/item/{product_id}.html" if product_id else "",
"price": {
"single": self._clean_price(item_soup.select_one(".product-price")?.text or ""),
"single_str": item_soup.select_one(".product-price")?.text.strip() or "",
"ladder_str": item_soup.select_one(".ladder-price")?.text.strip() or "",
"tax": "含税" in (item_soup.select_one(".tax-tag")?.text or "")
},
"params": params, # 工业参数（如材质、规格）
"stock": {
"stock_str": item_soup.select_one(".stock-text")?.text.strip() or "",
"total": self._clean_stock(item_soup.select_one(".stock-text")?.text or "")
},
"purchase": {
"min_order": item_soup.select_one(".min-order")?.text.strip() or "",
"delivery": item_soup.select_one(".delivery-time")?.text.strip() or ""
},
"vendor": {
"name": item_soup.select_one(".vendor-name")?.text.strip() or "",
"qualifications": qualification_tags
},
"sales": {
"volume": self._clean_sales(item_soup.select_one(".sales-volume")?.text or ""),
"volume_str": item_soup.select_one(".sales-volume")?.text.strip() or "",
"good_rate": item_soup.select_one(".good-rate")?.text.strip() or ""
}
}
def _parse_page(self, html: str) -> List[Dict]:
"""解析页面的商品列表"""
soup = BeautifulSoup(html, "lxml")
# 商品列表容器（需根据实际页面结构调整）
item_list = soup.select("div.product-list .product-item")
return [self._parse_item(item) for item in item_list if item.get("data-sku")]
def _get_total_pages(self, html: str) -> int:
"""获取总页数"""
soup = BeautifulSoup(html, "lxml")
page_box = soup.select_one(".pager-box")
if not page_box:
return 1
# 提取最后一页页码
last_page = page_box.select("a.page")[-1].text.strip() if page_box.select("a.page") else "1"
return int(last_page) if last_page.isdigit() else 1
def item_industries(self,
keyword: str = "",
category: str = "",
price_from: float = None,
price_to: float = None,
params: Dict = None,
vendor: str = "",
min_order: int = None,
sort: str = "",
page_limit: int = 5) -> Dict:
"""
搜索京东工业工业品列表
:param keyword: 搜索关键词
:param category: 分类名称（如“紧固件-螺栓”）或分类ID
:param price_from: 最低单价（元）
:param price_to: 最高单价（元）
:param params: 工业参数字典（如{"材质": "304不锈钢", "规格": "M10"}）
:param vendor: 供应商名称（如“京东自营”）或供应商ID
:param min_order: 最大起订量（个/件）
:param sort: 排序方式（sales-desc/price-asc等）
:param page_limit: 最大页数（默认5）
:return: 标准化搜索结果
"""
try:
# 1. 参数预处理
if not keyword and not category:
return {"success": False, "error_msg": "关键词（keyword）和分类（category）至少需提供一个"}
# 转换分类名称为ID
if category in self.category_map:
cat_id = self.category_map[category]
else:
cat_id = category if category else ""
# 转换供应商名称为ID
vendor_id = self.vendor_map.get(vendor, "") if vendor else ""
# 编码关键词
encoded_keyword = urllib.parse.quote(keyword, encoding="utf-8") if keyword else ""
# 处理工业参数（JSON序列化并编码）
encoded_params = ""
if params and isinstance(params, dict):
encoded_params = urllib.parse.quote(json.dumps(params, ensure_ascii=False))
all_items = []
current_page = 1
while current_page <= page_limit:
# 构建参数
params_dict = {
"page": current_page
}
if encoded_keyword:
params_dict["keyword"] = encoded_keyword
if cat_id:
params_dict["cat"] = cat_id
if price_from is not None:
params_dict["priceFrom"] = price_from
if price_to is not None:
params_dict["priceTo"] = price_to
if encoded_params:
params_dict["params"] = encoded_params
if vendor_id:
params_dict["vendorId"] = vendor_id
if min_order is not None:
params_dict["minOrder"] = min_order
if sort:
params_dict["sort"] = sort
# 发送请求（带随机延迟）
time.sleep(random.uniform(3, 5)) # 控制频率，避免反爬
headers = self._get_headers()
proxy = self._get_proxy()
response = requests.get(
url=self.base_url,
params=params_dict,
headers=headers,
proxies=proxy,
timeout=10
)
response.raise_for_status()
html = response.text
# 解析当前页商品
items = self._parse_page(html)
if not items:
break # 无数据，终止分页
all_items.extend(items)
# 获取总页数（仅第一页需要）
if current_page == 1:
total_pages = self._get_total_pages(html)
# 修正最大页数（不超过page_limit和50）
total_pages = min(total_pages, page_limit, 50)
if total_pages < current_page:
break
# 若当前页是最后一页，终止
if current_page >= total_pages:
break
current_page += 1
# 去重（基于product_id）
seen_ids = set()
unique_items = []
for item in all_items:
if item["product_id"] not in seen_ids:
seen_ids.add(item["product_id"])
unique_items.append(item)
return {
"success": True,
"total": len(unique_items),
"page_processed": current_page,
"items": unique_items
}
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客户端
search_api = JdIndustrialSearchApi(proxy_pool=PROXIES, cookie=COOKIE)
# 搜索“不锈钢螺栓”，分类“紧固件-螺栓”，价格0.5-1元，参数{材质:304不锈钢,规格:M10}，起订量≤100，销量降序，最多3页
result = search_api.item_industries(
keyword="不锈钢螺栓",
category="紧固件-螺栓",
price_from=0.5,
price_to=1.0,
params={"材质": "304不锈钢", "规格": "M10"},
min_order=100,
sort="sales-desc",
page_limit=3
)
if result["success"]:
print(f"搜索成功：共找到 {result['total']} 件商品，处理 {result['page_processed']} 页")
for i, item in enumerate(result["items"][:5]): # 打印前5条
print(f"\n商品 {i+1}:")
print(f"标题：{item['title'][:50]}...") # 截断长标题
print(f"价格：{item['price']['single_str']} | 阶梯价：{item['price']['ladder_str'] or '无'} | {'含税' if item['price']['tax'] else '不含税'}")
print(f"规格参数：{'; '.join([f'{k}:{v}' for k, v in item['params'].items()])}")
print(f"库存销量：{item['stock']['stock_str']} | 30天成交{item['sales']['volume']}个 | 好评率{item['sales']['good_rate']}")
print(f"采购信息：{item['purchase']['min_order']} | 配送：{item['purchase']['delivery']}")
print(f"供应商：{item['vendor']['name']} | 资质：{', '.join(item['vendor']['qualifications'])}")
print(f"详情页：{item['url']}")
else:
print(f"搜索失败：{result['error_msg']}（错误码：{result.get('code')}）")