中国工控网 item_search 接口是工控领域批量检索商品列表的核心入口，专注工业自动化 B2B 场景，支持通过关键词（型号、品牌、品类、属性等）组合多维度筛选，快速获取 PLC、传感器、变频器、伺服系统等工控商品的基础信息、价格区间、供应商资质等核心数据，广泛应用于工控设备采购选品、供应商挖掘、市场价格调研、备件库存盘点等核心业务，是工控场景下批量获取商品数据的必备工具。
一、接口核心认知：先明确 “能做什么”“适配什么场景”
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.gongkong.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_industrial_item(item: Dict) -> Dict:
"""解析工控商品数据（标准化字段命名，适配工控场景）"""
# 提取核心规格参数（适配工控品类特性）
specs = item.get("specs", [])
core_specs = {
"model": "", # 核心型号
"key_attribute": "" # 关键属性（如4-20mA、IP67）
}
for spec in specs:
key = spec.get("key", "").lower()
value = spec.get("value", "").strip()
if any(k in key for k in ["型号", "model"]):
core_specs["model"] = value
elif any(k in key for k in ["属性", "特性", "信号", "防护"]):
core_specs["key_attribute"] += value + "|" if value else ""
core_specs["key_attribute"] = core_specs["key_attribute"].rstrip("|")
# 标准化返回数据
return {
"商品ID": item.get("item_id", ""),
"商品标题": item.get("title", ""),
"品牌": item.get("brand", ""),
"核心型号": core_specs["model"],
"关键属性": core_specs["key_attribute"],
"所属类目": item.get("category", ""),
"商品标签": ",".join(item.get("tags", [])),
"基础单价": item.get("price", {}).get("base_price", 0.0),
"含税状态": "含税" if item.get("price", {}).get("tax_inclusive", True) else "不含税",
"批量阶梯价": "; ".join([f"{ladder['min_qty']}-{ladder['max_qty']}台：¥{ladder['price']}"
for ladder in item.get("price", {}).get("ladder_prices", [])]),
"库存状态": item.get("stock", {}).get("status", "现货"),
"最小起订量": item.get("stock", {}).get("min_order", 1),
"备货周期": item.get("stock", {}).get("lead_time", ""),
"供应商名称": item.get("seller", {}).get("company_name", ""),
"供应商资质": ",".join(item.get("seller", {}).get("qualification", [])),
"是否授权供应商": "是" if item.get("seller", {}).get("is_authorized", False) else "否",
"供应商地域": item.get("seller", {}).get("location", ""),
"核心认证": ",".join([
f"CE={item.get('certification', {}).get('ce', '')}",
f"CCC={item.get('certification', {}).get('ccc', '')}",
f"ROHS={item.get('certification', {}).get('rohs', '')}"
]),
"质保期": item.get("after_sales", {}).get("warranty_period", ""),
"技术支持": ",".join(item.get("after_sales", {}).get("support_method", [])),
"详情页URL": item.get("url", "")
}
def search_industrial_items(
keywords: str,
page_no: int = 1,
page_size: int = 50,
category_id: Optional[str] = None,
brand: 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_authorized: int = 0,
refresh: int = 0
) -> Dict:
"""
调用item_search接口搜索工控商品（单页）
:param keywords: 搜索关键词
:param page_no: 页码
:param page_size: 每页条数
:param category_id: 工控品类ID
:param brand: 品牌筛选
:param price_start: 最低价格
:param price_end: 最高价格
:param region: 地域筛选
:param qualification: 资质筛选
:param stock_status: 库存状态
:param is_authorized: 是否筛选授权供应商（1=是，0=否）
:param refresh: 是否强制刷新缓存（1=是，0=否，企业版可用）
:return: 标准化后的单页搜索结果
"""
# 1. 构建基础参数（必填项）
params = {
"appkey": APP_KEY,
"keywords": keywords,
"page_no": page_no,
"page_size": page_size,
"is_authorized": is_authorized,
"refresh": refresh,
"timestamp": int(time.time() * 1000),
# 按需筛选字段，减少数据传输量
"fields": "item_id,title,brand,specs,category,tags,price,stock,seller,certification,after_sales,url"
}
# 2. 添加可选参数
if category_id:
params["category_id"] = category_id
if brand:
params["brand"] = brand
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", {})
item_list = raw_data.get("list", [])
standard_items = [parse_industrial_item(item) for item in item_list]
return {
"success": True,
"data": standard_items,
"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_industrial_items(
keywords: str,
max_page: int = 100,
**kwargs
) -> List[Dict]:
"""
批量搜索工控商品（自动分页，获取所有数据）
:param keywords: 搜索关键词
:param max_page: 最大页码（默认100）
:param kwargs: 其他可选参数（如brand、price_start等）
:return: 所有页的标准化商品数据
"""
all_items = []
page_no = 1
total_count = 0
print(f"开始搜索关键词「{keywords}」的工控商品...")
while page_no <= max_page:
result = search_industrial_items(
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_items.extend(current_page_data)
total_count = result["total"]
print(f"第{page_no}页获取成功，累计获取{len(all_items)}/{total_count}条")
# 已获取全部数据，停止分页
if len(all_items) >= total_count:
print(f"已获取全部数据（共{total_count}条）")
break
# 控制调用频率（免费版5秒/页，企业版2秒/页）
time.sleep(5)
page_no += 1
return all_items
def save_search_results(products: List[Dict], save_path: str):
"""将搜索结果保存为Excel文件（便于采购分析）"""
if not products:
print("无数据可保存")
return
df = pd.DataFrame(products)
# 筛选常用字段，优化Excel可读性
columns = [
"商品ID", "商品标题", "品牌", "核心型号", "关键属性", "基础单价", "含税状态",
"批量阶梯价", "库存状态", "最小起订量", "供应商名称", "是否授权供应商",
"供应商资质", "供应商地域", "核心认证", "质保期", "详情页URL"
]
df = df[columns].drop_duplicates(subset=["商品ID"]) # 按商品ID去重
df.to_excel(save_path, index=False, engine="openpyxl")
print(f"数据已保存至：{save_path}（共{len(df)}条记录）")
# 调用示例（工控采购选品场景）
if __name__ == "__main__":
# 搜索条件配置（西门子PLC现货采购）
SEARCH_KEYWORDS = "西门子 S7-1200 PLC 现货"
BRAND = "西门子"
CATEGORY_ID = "10102" # 工控设备>PLC
PRICE_START = 1000.0
PRICE_END = 5000.0
REGION = "广东省"
QUALIFICATION = "ISO9001,原厂授权"
STOCK_STATUS = "现货"
IS_AUTHORIZED = 1 # 仅筛选授权供应商
# 批量搜索商品
product_list = batch_search_industrial_items(
keywords=SEARCH_KEYWORDS,
brand=BRAND,
category_id=CATEGORY_ID,
price_start=PRICE_START,
price_end=PRICE_END,
region=REGION,
qualification=QUALIFICATION,
stock_status=STOCK_STATUS,
is_authorized=IS_AUTHORIZED
)
# 保存结果
if product_list:
save_search_results(product_list, SAVE_PATH)
else:
print("未获取到符合条件的工控商品数据")