电子元件的搜索接口与普通消费品存在显著差异，其核心在于参数化筛选和技术指标匹配。电子元件（如电阻、电容、芯片、传感器等）的搜索不仅需要匹配关键词，更需要精确筛选电气参数、物理特性、认证标准等专业维度。本文将针对电子元件的行业特性，深入分析 item_search 接口的设计逻辑、参数体系及 Python 实现方案。
一、电子元件 item_search 接口核心特性分析
1. 接口定位与行业价值
2. 接口权限与调用限制
3. 核心参数解析（含电子元件专属参数）
必选参数
电子元件专属筛选参数（部分）
二、返回数据结构与行业专属字段解析
1. 数据结构总览
{
"sn_responseContent": {
"sn_body": {
"searchResult": {
"totalCount": 128, // 总商品数
"pageNo": 1, // 当前页码
"pageSize": 20, // 每页条数
"productList": [ // 商品列表
{
// 1. 通用信息层
"productCode": "ELEC20240600159",
"productName": "STMicroelectronics STM32F103C8T6 32位单片机",
"brandName": "STMicroelectronics",
"price": "18.50", // 单价
"stockStatus": "有库存",
// 2. 技术参数层（电子元件核心）
"techParams": {
"baseParams": {
"model": "STM32F103C8T6", // 原厂型号
"core": "ARM Cortex-M3", // 内核
"frequency": "72MHz", // 主频
"flash": "64KB", // 闪存
"ram": "20KB" // 内存
},
"physicalParams": {
"package": "LQFP48", // 封装
"pinCount": 48, // 引脚数
"dimensions": "7x7mm" // 尺寸
},
"environmentalParams": {
"operatingTemp": "-40℃ ~ 85℃", // 工作温度
"storageTemp": "-55℃ ~ 125℃" // 存储温度
},
"certifications": ["RoHS", "CE", "REACH"] // 认证
},
// 3. 供应链层
"supplyChain": {
"moq": 10, // 最小起订量
"leadTime": "3-5天", // 交期
"batchCount": 3, // 库存批次数量
"supplierType": "授权分销商" // 供应商类型
},
// 4. 替代型号信息
"alternativeModels": [
{"model": "STM32F103CBT6", "similarity": 95}, // 相似度95%
{"model": "GD32F103C8T6", "similarity": 90} // 国产替代
]
},
// 更多商品...
],
// 筛选条件统计（用于前端筛选项展示）
"filterStats": {
"brandStats": [{"brandId": "1002", "brandName": "ST", "count": 35}, ...],
"packageStats": [{"value": "LQFP48", "count": 28}, ...],
"priceRangeStats": [{"range": "10-20", "count": 56}, ...]
}
}
},
"sn_head": {"returnCode": "0000", "returnMessage": "success"}
}}2. 核心字段解读（行业专属）
三、Python 实现方案（适配电子元件搜索特性）
import requests
import time
import hashlib
import json
import logging
import pandas as pd
import matplotlib.pyplot as plt
import re
from datetime import datetime
from typing import Dict, Optional, List, Tuple
from collections import defaultdict
# 配置日志（工业场景需详细记录搜索参数与结果，便于追溯）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - [电子元件搜索] - %(message)s",
handlers=[logging.FileHandler("elec_search_api.log"), logging.StreamHandler()]
)
# 配置中文显示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams["axes.unicode_minus"] = False
class ElecItemSearch:
"""电子元件 item_search 接口封装类，支持参数化搜索与技术指标分析"""
def __init__(self, app_key: str, app_secret: str, auth_level: str = "enterprise"):
"""
初始化电子元件搜索客户端
:param app_key: 应用标识
:param app_secret: 应用密钥
:param auth_level: 权限等级（personal/enterprise/industrial）
"""
self.app_key = app_key
self.app_secret = app_secret
self.auth_level = auth_level
self.api_url = "https://open.elec-platform.com/api/http/elecSearch"
# 频率控制（按权限等级）
self.rate_limit = {
"personal": 3,
"enterprise": 20,
"industrial": 60
}.get(auth_level, 3) # 次/分钟
self.call_timestamps = []
# 支持的技术参数筛选类型
self.supported_params = {
"electrical": ["resistance", "capacitance", "voltage", "current", "frequency"],
"physical": ["package", "pinCount", "dimensions", "weight"],
"environmental": ["operatingTemp", "storageTemp", "humidity"]
}
# 支持的排序方式
self.supported_sorts = [
"price asc", "price desc", # 价格排序
"stock desc", # 库存排序
"precision desc", # 精度排序（电子元件专属）
"popularity desc", # popularity
"updateTime desc" # 更新时间
]
def _generate_sign(self, params: Dict) -> str:
"""生成签名（电子元件平台标准MD5签名）"""
# 参数按ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 拼接为key=value&key=value格式
param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 拼接appSecret并加密
sign_str = f"{param_str}&appSecret={self.app_secret}"
md5 = hashlib.md5()
md5.update(sign_str.encode('utf-8'))
return md5.hexdigest().upper()
def _check_rate_limit(self) -> None:
"""检查调用频率，确保不超限"""
current_time = time.time()
# 保留1分钟内的调用记录
self.call_timestamps = [t for t in self.call_timestamps if current_time - t < 60]
if len(self.call_timestamps) >= self.rate_limit:
# 计算需要等待的时间
oldest_time = self.call_timestamps[0]
sleep_time = 60 - (current_time - oldest_time) + 0.2
logging.warning(f"调用频率超限，等待 {sleep_time:.1f} 秒（权限等级：{self.auth_level}）")
time.sleep(sleep_time)
# 清理过期记录
self.call_timestamps = [t for t in self.call_timestamps if time.time() - t < 60]
# 记录本次调用时间
self.call_timestamps.append(current_time)
def _validate_filters(self, filters: Dict) -> Tuple[bool, str]:
"""验证筛选参数的合法性"""
# 验证排序方式
if "sort" in filters and filters["sort"] not in self.supported_sorts:
return False, f"不支持的排序方式: {filters['sort']}，支持: {', '.join(self.supported_sorts)}"
# 验证参数筛选格式
for param_type in ["paramFilter", "physicalFilter", "envFilter"]:
if param_type in filters and filters[param_type]:
try:
# 检查是否为合法JSON
filter_json = json.loads(filters[param_type])
# 检查参数是否在支持的列表中
param_category = param_type.replace("Filter", "")
if param_category in self.supported_params:
for key in filter_json:
if key not in self.supported_params[param_category]:
return False, f"不支持的{param_category}参数: {key}，支持: {', '.join(self.supported_params[param_category])}"
except json.JSONDecodeError:
return False, f"{param_type}格式错误，需为JSON字符串"
return True, "验证通过"
def search_elec_items(self, keyword: str,
page_no: int = 1, page_size: int = 20,
filters: Optional[Dict] = None) -> Optional[Dict]:
"""
搜索电子元件商品
:param keyword: 搜索关键词（型号、规格、功能描述）
:param page_no: 页码
:param page_size: 每页数量（10-50）
:param filters: 筛选条件（含技术参数筛选）
:return: 搜索结果
"""
# 基础参数验证
if not (2 <= len(keyword) <= 50):
logging.error(f"关键词长度必须在2-50之间，当前: {len(keyword)}")
return None
if not (10 <= page_size <= 50):
logging.error(f"page_size必须在10-50之间，当前: {page_size}")
return None
# 构建基础参数
base_params = {
"appKey": self.app_key,
"timestamp": datetime.now().strftime("%Y%m%d%H%M%S"),
"method": "elec.product.search",
"version": "v2.1",
"keyword": keyword,
"pageNo": str(page_no),
"pageSize": str(page_size)
}
# 处理筛选参数
if filters and isinstance(filters, Dict):
# 验证筛选参数
valid, msg = self._validate_filters(filters)
if not valid:
logging.error(f"筛选参数错误: {msg}")
return None
# 合并筛选参数（转换为字符串类型）
for k, v in filters.items():
if v is not None:
base_params[k] = json.dumps(v) if isinstance(v, (dict, list)) else str(v)
# 生成签名
sign = self._generate_sign(base_params)
base_params["sign"] = sign
# 检查频率限制
self._check_rate_limit()
try:
# 发送请求
response = requests.get(self.api_url, params=base_params, timeout=15)
response.raise_for_status()
# 解析响应
result = response.json()
# 处理错误响应
if "sn_responseContent" not in result:
logging.error(f"响应格式错误: {result}")
return None
response_content = result["sn_responseContent"]
if "sn_error" in response_content:
error = response_content["sn_error"]
logging.error(f"API错误: {error.get('error_msg')} (错误码: {error.get('error_code')})")
return None
# 提取搜索结果
search_data = response_content.get("sn_body", {}).get("searchResult", {})
if not search_data.get("productList"):
logging.info(f"关键词 '{keyword}' 未找到匹配商品")
return None
logging.info(f"搜索成功: 关键词 '{keyword}'，第{page_no}页，共{search_data.get('totalCount')}个商品")
return search_data
except requests.exceptions.RequestException as e:
logging.error(f"请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"响应解析失败: {response.text[:200]}...")
return None
def batch_search(self, keyword: str, max_pages: int = 5,
page_size: int = 50, filters: Optional[Dict] = None) -> Tuple[List[Dict], Dict]:
"""
批量获取多页搜索结果
:param keyword: 搜索关键词
:param max_pages: 最大页数
:param page_size: 每页数量
:param filters: 筛选条件
:return: 商品列表和元信息
"""
all_items = []
meta_info = {
"keyword": keyword,
"total_items": 0,
"total_pages": 0,
"filters": filters,
"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
page = 1
while page <= max_pages:
logging.info(f"获取关键词 '{keyword}' 第 {page}/{max_pages} 页...")
result = self.search_elec_items(keyword, page, page_size, filters)
if not result:
break
# 提取商品数据
items = result.get("productList", [])
if not items:
logging.info("当前页无数据，停止获取")
break
all_items.extend(items)
# 记录元信息（第一页）
if page == 1:
meta_info["total_items"] = result.get("totalCount", 0)
meta_info["total_pages"] = min(max_pages,
(meta_info["total_items"] + page_size - 1) // page_size)
meta_info["filter_stats"] = result.get("filterStats", {}) # 筛选条件统计
page += 1
logging.info(f"批量搜索完成，共获取 {len(all_items)} 个商品")
return all_items, meta_info
def analyze_technical_params(self, items: List[Dict]) -> Dict:
"""分析搜索结果中的技术参数分布（电子元件核心分析）"""
if not items:
return {"error": "无商品数据可分析"}
# 1. 封装类型分布
package_dist = defaultdict(int)
# 2. 品牌分布
brand_dist = defaultdict(int)
# 3. 工作温度范围分布（区分民用/工业/军工级）
temp_grade_dist = defaultdict(int)
temp_grade_map = {
"民用级": lambda t: "0℃~70℃" in t,
"工业级": lambda t: "-40℃~85℃" in t,
"军工级": lambda t: "-55℃~125℃" in t,
"其他": lambda t: True # 匹配所有未命中的情况
}
# 4. 价格分布（按元件类型区分，避免不同类型价格不可比）
price_by_category = defaultdict(list)
# 5. 最小起订量分布
moq_dist = defaultdict(int)
for item in items:
# 封装分布
package = item.get("techParams", {}).get("physicalParams", {}).get("package", "未知")
package_dist[package] += 1
# 品牌分布
brand = item.get("brandName", "未知品牌")
brand_dist[brand] += 1
# 工作温度等级分布
operating_temp = item.get("techParams", {}).get("environmentalParams", {}).get("operatingTemp", "")
for grade, check in temp_grade_map.items():
if check(operating_temp):
temp_grade_dist[grade] += 1
break
# 价格按类目分布
category = item.get("categoryName", "未知类目")
try:
price = float(item.get("price", 0))
price_by_category[category].append(price)
except (ValueError, TypeError):
pass
# 最小起订量分布
moq = item.get("supplyChain", {}).get("moq", 0)
try:
moq_val = int(moq)
if moq_val <= 10:
moq_dist["1-10"] += 1
elif moq_val <= 100:
moq_dist["11-100"] += 1
elif moq_val <= 1000:
moq_dist["101-1000"] += 1
else:
moq_dist[">1000"] += 1
except (ValueError, TypeError):
moq_dist["未知"] += 1
# 计算价格统计值
price_stats = {}
for category, prices in price_by_category.items():
if prices:
price_stats[category] = {
"count": len(prices),
"min": round(min(prices), 2),
"max": round(max(prices), 2),
"avg": round(sum(prices)/len(prices), 2)
}
return {
"package_distribution": dict(package_dist),
"brand_distribution": dict(brand_dist),
"temperature_grade_distribution": dict(temp_grade_dist),
"price_statistics_by_category": price_stats,
"moq_distribution": dict(moq_dist),
"analysis_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
def find_alternatives(self, items: List[Dict], target_model: str) -> List[Dict]:
"""
从搜索结果中查找目标型号的替代型号
:param items: 商品列表
:param target_model: 目标型号
:return: 替代型号列表（按相似度排序）
"""
alternatives = []
for item in items:
# 检查商品自身是否为目标型号
current_model = item.get("techParams", {}).get("baseParams", {}).get("model", "")
if current_model == target_model:
# 提取该商品的替代型号推荐
alt_models = item.get("alternativeModels", [])
for alt in alt_models:
# 补充替代型号的商品信息
alt_info = {
"model": alt.get("model"),
"similarity": alt.get("similarity", 0),
"brand": item.get("brandName"),
"price": item.get("price"),
"package": item.get("techParams", {}).get("physicalParams", {}).get("package")
}
alternatives.append(alt_info)
# 按相似度降序排序
return sorted(alternatives, key=lambda x: x["similarity"], reverse=True)
def visualize_analysis(self, tech_analysis: Dict, output_dir: str = ".") -> None:
"""可视化技术参数分析结果"""
if "error" in tech_analysis:
logging.warning(tech_analysis["error"])
return
# 1. 封装类型分布饼图
if tech_analysis["package_distribution"]:
plt.figure(figsize=(10, 8))
package_data = tech_analysis["package_distribution"]
# 只显示占比>3%的封装，其余归为"其他"
others = 0
filtered = {}
total = sum(package_data.values())
for pkg, count in package_data.items():
ratio = count / total
if ratio > 0.03:
filtered[pkg] = count
else:
others += count
if others > 0:
filtered["其他"] = others
labels = list(filtered.keys())
sizes = list(filtered.values())
plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)
plt.title('封装类型分布')
plt.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/elec_package_dist.png")
plt.close()
logging.info(f"封装分布图表已保存至 {output_dir}/elec_package_dist.png")
# 2. 品牌分布条形图（前10）
if tech_analysis["brand_distribution"]:
plt.figure(figsize=(12, 6))
brand_data = sorted(tech_analysis["brand_distribution"].items(),
key=lambda x: x[1], reverse=True)[:10]
brands, counts = zip(*brand_data)
plt.bar(brands, counts, color='skyblue')
plt.title('品牌分布（前10）')
plt.xlabel('品牌')
plt.ylabel('商品数量')
plt.xticks(rotation=45)
for i, v in enumerate(counts):
plt.text(i, v + 0.5, str(v), ha='center')
plt.tight_layout()
plt.savefig(f"{output_dir}/elec_brand_dist.png")
plt.close()
logging.info(f"品牌分布图表已保存至 {output_dir}/elec_brand_dist.png")
# 3. 工作温度等级分布饼图
if tech_analysis["temperature_grade_distribution"]:
plt.figure(figsize=(8, 8))
temp_data = tech_analysis["temperature_grade_distribution"]
labels = list(temp_data.keys())
sizes = list(temp_data.values())
plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)
plt.title('工作温度等级分布')
plt.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/elec_temp_grade_dist.png")
plt.close()
logging.info(f"温度等级分布图表已保存至 {output_dir}/elec_temp_grade_dist.png")
# 4. 类目价格统计柱状图
if tech_analysis["price_statistics_by_category"]:
plt.figure(figsize=(12, 6))
category_data = tech_analysis["price_statistics_by_category"]
# 只显示有足够数据的类目
valid_categories = {k: v for k, v in category_data.items() if v["count"] >= 5}
if valid_categories:
categories = list(valid_categories.keys())
avg_prices = [v["avg"] for v in valid_categories.values()]
plt.bar(categories, avg_prices, color='orange')
plt.title('各类目平均价格')
plt.xlabel('商品类目')
plt.ylabel('平均价格（元）')
plt.xticks(rotation=45)
for i, v in enumerate(avg_prices):
plt.text(i, v + 0.1, f"¥{v}", ha='center')
plt.tight_layout()
plt.savefig(f"{output_dir}/elec_category_price.png")
plt.close()
logging.info(f"类目价格图表已保存至 {output_dir}/elec_category_price.png")
def export_to_excel(self, items: List[Dict], tech_analysis: Dict,
meta_info: Dict, filename: str) -> None:
"""导出搜索结果与分析数据到Excel"""
if not items and not tech_analysis:
logging.warning("无数据可导出")
return
try:
with pd.ExcelWriter(filename) as writer:
# 1. 搜索元信息
pd.DataFrame([meta_info]).to_excel(writer, sheet_name='搜索信息', index=False)
# 2. 商品列表（含技术参数）
if items:
item_data = []
for item in items:
tech_params = item.get("techParams", {})
supply_chain = item.get("supplyChain", {})
data = {
"商品ID": item.get("productCode"),
"商品名称": item.get("productName"),
"品牌": item.get("brandName"),
"型号": tech_params.get("baseParams", {}).get("model", ""),
"封装": tech_params.get("physicalParams", {}).get("package", ""),
"工作温度": tech_params.get("environmentalParams", {}).get("operatingTemp", ""),
"认证": ",".join(tech_params.get("certifications", [])),
"单价(元)": item.get("price"),
"最小起订量": supply_chain.get("moq", ""),
"交期": supply_chain.get("leadTime", ""),
"库存状态": item.get("stockStatus")
}
item_data.append(data)
df_items = pd.DataFrame(item_data)
df_items.to_excel(writer, sheet_name='商品列表', index=False)
# 3. 技术参数分析结果
if tech_analysis and "error" not in tech_analysis:
# 封装分布
df_package = pd.DataFrame(
list(tech_analysis["package_distribution"].items()),
columns=["封装类型", "数量"]
)
df_package.to_excel(writer, sheet_name='封装分布', index=False)
# 品牌分布
df_brand = pd.DataFrame(
list(tech_analysis["brand_distribution"].items()),
columns=["品牌", "数量"]
)
df_brand.to_excel(writer, sheet_name='品牌分布', index=False)
# 温度等级分布
df_temp = pd.DataFrame(
list(tech_analysis["temperature_grade_distribution"].items()),
columns=["温度等级", "数量"]
)
df_temp.to_excel(writer, sheet_name='温度等级分布', index=False)
logging.info(f"数据已导出至 {filename}")
except Exception as e:
logging.error(f"导出Excel失败: {str(e)}")
# 示例调用
if __name__ == "__main__":
# 替换为实际参数
APP_KEY = "your_elec_appkey"
APP_SECRET = "your_elec_appsecret"
KEYWORD = "STM32F103 单片机" # 搜索关键词（型号+品类）
# 初始化客户端（企业级权限）
elec_search = ElecItemSearch(APP_KEY, APP_SECRET, auth_level="enterprise")
# 1. 定义技术筛选条件（搜索STM32F103系列单片机，LQFP48封装，工业级温度）
filters = {
"sort": "price asc", # 按价格升序
"physicalFilter": json.dumps({"package": "LQFP48", "pinCount": 48}), # 物理参数筛选
"envFilter": json.dumps({"operatingTemp": "-40~85℃"}), # 环境参数筛选
"certFilter": "RoHS,CE", # 认证筛选
"priceFrom": 10, # 价格下限
"priceTo": 50, # 价格上限
"stockFilter": json.dumps({"minQty": 100}) # 库存筛选（至少100个）
}
# 2. 批量搜索（获取前3页结果）
print("=== 电子元件搜索 ===")
items, meta_info = elec_search.batch_search(
keyword=KEYWORD,
max_pages=3,
page_size=50,
filters=filters
)
if items:
print(f"搜索关键词: {KEYWORD}")
print(f"获取商品数量: {len(items)}")
print(f"总商品数量: {meta_info['total_items']}")
print(f"主要封装类型: {', '.join([k for k, v in meta_info['filter_stats'].get('packageStats', [])[:3]])}")
# 3. 技术参数分析
print("\n=== 技术参数分析 ===")
if items:
tech_analysis = elec_search.analyze_technical_params(items)
print("封装类型分布（前3）:")
top_packages = sorted(tech_analysis["package_distribution"].items(),
key=lambda x: x[1], reverse=True)[:3]
for pkg, count in top_packages:
print(f" {pkg}: {count}个商品")
print("\n品牌分布（前3）:")
top_brands = sorted(tech_analysis["brand_distribution"].items(),
key=lambda x: x[1], reverse=True)[:3]
for brand, count in top_brands:
print(f" {brand}: {count}个商品")
print("\n温度等级分布:")
for grade, count in tech_analysis["temperature_grade_distribution"].items():
print(f" {grade}: {count}个商品")
print("\n最小起订量分布:")
for range_, count in tech_analysis["moq_distribution"].items():
print(f" {range_}个: {count}个商品")
# 4. 查找替代型号（以STM32F103C8T6为例）
print("\n=== 替代型号推荐 ===")
alternatives = elec_search.find_alternatives(items, "STM32F103C8T6")
if alternatives:
for i, alt in enumerate(alternatives[:5], 1):
print(f"{i}. 型号: {alt['model']}, 相似度: {alt['similarity']}%, 价格: ¥{alt['price']}")
else:
print("未找到替代型号")
# 5. 可视化分析结果
elec_search.visualize_analysis(tech_analysis)
# 6. 导出数据到Excel
elec_search.export_to_excel(
items, tech_analysis, meta_info,
f"电子元件搜索_{KEYWORD.replace(' ', '_')}_分析.xlsx"
)
else:
print("未获取到商品数据，无法进行分析")