中国供应网 item_search 接口是批量检索工业商品的核心入口，专注 B2B 采购场景，支持通过关键词（商品名称、型号、材质、行业属性等）组合筛选，快速获取目标商品列表及核心信息（价格、规格、供应商、库存状态），广泛应用于采购选品、供应商挖掘、竞品分析、市场调研等业务，是工业场景下批量获取商品数据的必备工具。
一、接口核心认知：先明确 “能做什么”“适配什么场景”
1. 接口定位与核心价值
2. 核心参数与返回字段（工业场景适配版）
（1）请求参数（必填 + 可选，按优先级排序）
（2）返回核心字段（按业务场景分类）
3. 接口限制与注意事项
二、对接前准备：3 步搞定前置条件
1. 注册与获取密钥（核心步骤）
2. 技术环境准备
（1）支持语言与协议
（2）必备工具与依赖
3. 业务需求梳理
三、实操步骤：从调试到落地（Python 示例）
步骤 1：理解请求流程
步骤 2：签名生成规则（关键！避免调用失败）
签名示例（参数排序与拼接）
步骤 3：完整代码实现（Python）
（1）依赖安装
（2）完整代码（含签名生成、分页获取、数据保存）
import requests
import hashlib
import time
import json
import pandas as pd
from urllib.parse import urlencode
from typing import Dict, Optional, List
# 接口核心配置（替换为自己的密钥和接口地址）
APP_KEY = "你的appkey"
SECRET = "你的secret"
API_URL = "https://api.chinasilkroad.com/item_search" # 正式接口地址
SAVE_PATH = "工业商品搜索结果.xlsx" # 结果保存路径
def generate_sign(params: Dict) -> str:
"""生成接口签名（严格按平台规则实现，核心函数）"""
# 1. 按参数名ASCII升序排序（排除sign字段）
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接参数字符串（urlencode自动处理中文、空格等特殊字符）
param_str = urlencode(sorted_params, encoding="utf-8") + f"&secret={SECRET}"
# 3. MD5加密（32位大写）
md5 = hashlib.md5()
md5.update(param_str.encode("utf-8"))
return md5.hexdigest().upper()
def parse_product_specs(specs: List[Dict]) -> Dict:
"""解析商品规格参数（标准化字段命名，适配工业场景）"""
standard_specs = {
"model": "", # 型号
"brand": "", # 品牌
"material": "", # 材质
"protection_level": "", # 防护等级
"operating_temperature": "", # 工作温度
"executive_standard": "" # 执行标准
}
for spec in specs:
key = spec.get("key", "").lower()
value = spec.get("value", "").strip()
# 关键词匹配，标准化字段
if any(k in key for k in ["型号", "model"]):
standard_specs["model"] = value
elif any(k in key for k in ["品牌", "brand"]):
standard_specs["brand"] = value
elif any(k in key for k in ["材质", "material"]):
standard_specs["material"] = value
elif any(k in key for k in ["防护", "ip"]):
standard_specs["protection_level"] = value
elif any(k in key for k in ["温度", "temperature"]):
standard_specs["operating_temperature"] = value
elif any(k in key for k in ["标准", "standard", "gb", "iso"]):
standard_specs["executive_standard"] = value
return standard_specs
def search_products(
keywords: str,
page_no: int = 1,
page_size: int = 50,
category_id: Optional[str] = None,
price_start: Optional[float] = None,
price_end: Optional[float] = None,
region: Optional[str] = None,
qualification: Optional[str] = None,
stock_status: Optional[str] = None,
is_custom: int = -1,
refresh: int = 0
) -> Dict:
"""
调用item_search接口搜索商品（单页）
:param keywords: 搜索关键词
:param page_no: 页码
:param page_size: 每页条数
:param category_id: 分类ID
:param price_start: 最低价格
:param price_end: 最高价格
:param region: 地域筛选
:param qualification: 资质筛选
:param stock_status: 库存状态
:param is_custom: 是否筛选定制商品（1=仅定制，0=仅现货，-1=全部）
:param refresh: 是否强制刷新缓存（1=是，0=否，企业版可用）
:return: 标准化后的单页搜索结果
"""
# 1. 构建基础参数（必填项）
params = {
"appkey": APP_KEY,
"keywords": keywords,
"page_no": page_no,
"page_size": page_size,
"is_custom": is_custom,
"refresh": refresh,
"timestamp": int(time.time() * 1000),
# 按需筛选字段，减少数据传输量
"fields": "product_id,title,main_image,category,tags,price,stock,specs,seller,trade_terms"
}
# 2. 添加可选参数
if category_id:
params["category_id"] = category_id
if price_start is not None:
params["price_start"] = price_start
if price_end is not None:
params["price_end"] = price_end
if region:
params["region"] = region
if qualification:
params["qualification"] = qualification
if stock_status:
params["stock_status"] = stock_status
# 3. 生成签名
params["sign"] = generate_sign(params)
try:
# 4. 发送POST请求
response = requests.post(
url=API_URL,
data=json.dumps(params),
headers={"Content-Type": "application/json"},
timeout=10,
verify=True
)
result = response.json()
# 5. 处理响应
if result.get("code") == 200:
raw_data = result.get("data", {})
product_list = raw_data.get("list", [])
standard_products = []
for product in product_list:
# 解析规格参数
specs = parse_product_specs(product.get("specs", []))
# 标准化数据
standard_products.append({
"商品ID": product.get("product_id", ""),
"商品标题": product.get("title", ""),
"主图URL": product.get("main_image", ""),
"所属类目": product.get("category", ""),
"商品标签": ",".join(product.get("tags", [])),
"基础单价": product.get("price", {}).get("base_price", 0.0),
"含税状态": "含税" if product.get("price", {}).get("tax_inclusive", True) else "不含税",
"最小起订量": product.get("stock", {}).get("min_order", 1),
"库存状态": product.get("stock", {}).get("status", ""),
"库存数量": product.get("stock", {}).get("quantity", 0),
"型号": specs["model"],
"品牌": specs["brand"],
"材质": specs["material"],
"防护等级": specs["protection_level"],
"工作温度": specs["operating_temperature"],
"执行标准": specs["executive_standard"],
"供应商名称": product.get("seller", {}).get("company_name", ""),
"供应商资质": ",".join(product.get("seller", {}).get("qualification", [])),
"供应商地域": product.get("seller", {}).get("location", ""),
"交货周期": product.get("trade_terms", {}).get("delivery_cycle", ""),
"质保期": product.get("trade_terms", {}).get("warranty_period", "")
})
return {
"success": True,
"data": standard_products,
"total": raw_data.get("total", 0),
"page_no": page_no,
"page_size": page_size
}
else:
return {
"success": False,
"error_code": result.get("code"),
"error_msg": result.get("msg", "接口调用失败")
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error_code": -2,
"error_msg": f"网络异常：{str(e)}"
}
except Exception as e:
return {
"success": False,
"error_code": -3,
"error_msg": f"处理异常：{str(e)}"
}
def batch_search_products(
keywords: str,
max_page: int = 100,
**kwargs
) -> List[Dict]:
"""
批量搜索商品（自动分页，获取所有数据）
:param keywords: 搜索关键词
:param max_page: 最大页码（默认100）
:param kwargs: 其他可选参数（如price_start、region等）
:return: 所有页的标准化商品数据
"""
all_products = []
page_no = 1
total_count = 0
print(f"开始搜索关键词「{keywords}」的商品...")
while page_no <= max_page:
result = search_products(
keywords=keywords,
page_no=page_no,
page_size=50,
**kwargs
)
if not result["success"]:
print(f"第{page_no}页获取失败：{result['error_msg']}（错误码：{result['error_code']}）")
# 频率超限或服务器异常，重试1次
if result["error_code"] in [429, 500]:
time.sleep(10)
continue
break
current_page_data = result["data"]
if not current_page_data:
print(f"第{page_no}页无数据，停止获取")
break
all_products.extend(current_page_data)
total_count = result["total"]
print(f"第{page_no}页获取成功，累计获取{len(all_products)}/{total_count}条")
# 已获取全部数据，停止分页
if len(all_products) >= total_count:
print(f"已获取全部数据（共{total_count}条）")
break
# 控制调用频率（免费版3-5秒/页，企业版1-2秒/页）
time.sleep(3)
page_no += 1
return all_products
def save_search_results(products: List[Dict], save_path: str):
"""将搜索结果保存为Excel文件（便于采购分析）"""
if not products:
print("无数据可保存")
return
df = pd.DataFrame(products)
# 筛选常用字段，优化Excel可读性
columns = [
"商品ID", "商品标题", "品牌", "型号", "基础单价", "含税状态",
"最小起订量", "库存状态", "库存数量", "供应商名称", "供应商资质",
"供应商地域", "交货周期", "质保期"
]
df = df[columns].drop_duplicates(subset=["商品ID"]) # 去重
df.to_excel(save_path, index=False, engine="openpyxl")
print(f"数据已保存至：{save_path}（共{len(df)}条记录）")
# 调用示例（工业商品搜索场景）
if __name__ == "__main__":
# 搜索条件配置（电子元件采购场景）
SEARCH_KEYWORDS = "STM32F103C8T6 车规级 RoHS"
PRICE_START = 1.0
PRICE_END = 5.0
REGION = "广东省"
QUALIFICATION = "ISO9001,RoHS,IATF16949"
STOCK_STATUS = "现货"
# 批量搜索商品
product_list = batch_search_products(
keywords=SEARCH_KEYWORDS,
price_start=PRICE_START,
price_end=PRICE_END,
region=REGION,
qualification=QUALIFICATION,
stock_status=STOCK_STATUS,
is_custom=0 # 仅筛选现货商品
)
# 保存结果
if product_list:
save_search_results(product_list, SAVE_PATH)
else:
print("未获取到符合条件的商品数据")四、调试与验证：快速定位问题
1. 调试步骤（优先用 Postman 验证，避免代码干扰）
2. 常见调试问题排查（工业场景高频问题）
五、进阶优化：提升效率与稳定性（生产级必备）
1. 性能优化（批量搜索场景重点）
（1）关键词拆分策略
（2）异步并发请求
import aiohttp
import asyncio
async def async_search_products(session, keywords, page_no):
"""异步搜索单页商品"""
params = {
"appkey": APP_KEY,
"keywords": keywords,
"page_no": page_no,
"page_size": 50,
"timestamp": int(time.time() * 1000),
"sign": generate_sign(params)
}
async with session.post(
API_URL,
json=params,
headers={"Content-Type": "application/json"},
timeout=10
) as response:
return await response.json()
# 并发调用（控制并发数≤5，避免频率超限）
async def batch_async_search(keywords_list):
async with aiohttp.ClientSession() as session:
tasks = [async_search_products(session, kw, 1) for kw in keywords_list[:5]]
results = await asyncio.gather(*tasks)
return results