苏宁的item_search接口是按关键字搜索商品列表的核心接口，为开发者提供了通过关键词、分类、价格区间等条件筛选商品的能力。该接口广泛应用于电商数据分析、竞品研究、价格监控等场景，能够高效获取精准的商品列表及关键信息。
一、苏宁 item_search 接口核心特性分析
- 接口定位与核心价值
苏宁item_search接口作为平台商品搜索功能，其核心价值体现在：
支持多维度筛选，精准定位目标商品
提供商品基础信息、价格、销量、评价等核心数据
支持分页获取，可获取大规模商品数据
可结合促销标签筛选优惠商品
提供多种排序功能，快速定位热门或优质商品
支持按地区筛选，获取区域化商品信息 - 接口权限与调用限制
使用苏宁搜索接口需遵守以下规范：
限制类型 具体规则 说明
权限要求 需注册苏宁开发者账号，创建应用并通过审核 个人与企业开发者权限不同，企业开发者可获取更完整数据
调用频率 基础权限：10 次 / 分钟；高级权限：60 次 / 分钟 按 appKey 和 IP 双重限制
数据返回 单次最多返回 50 条商品，最多支持 100 页数据 搜索结果有 5 分钟缓存
字段限制 基础字段免费，部分高级字段需申请开通 敏感信息受平台保护
搜索限制 关键词长度限制为 2-30 个字符 支持中英文及部分特殊符号 - 核心参数解析
必选参数
参数名 类型 说明 示例
appKey String 应用唯一标识 "suning_appkey_12345"
sign String 签名，按平台算法生成 见下文签名逻辑
timestamp String 时间戳（yyyyMMddHHmmss） "20230701120000"
keyword String 搜索关键词 "智能手机"
可选参数
参数名 类型 说明 示例
pageNo Integer 页码，默认 1 1
pageSize Integer 每页条数，10-50 50
sort String 排序方式 "price asc"（价格升序）、"saleCount desc"（销量降序）等
priceFrom Float 最低价格 1000
priceTo Float 最高价格 5000
categoryId String 类目 ID 筛选 "700501003"
brandId String 品牌 ID 筛选 "1000000567"
area String 地区编码 "110100"（北京）
hasPromotion String 是否只看促销商品 "1"（是）、"0"（否）
isSelf String 是否只看自营商品 "1"（是）、"0"（否）
二、签名生成与返回数据结构 - 签名生成逻辑
苏宁开放平台采用 MD5 签名算法，步骤如下：
收集所有请求参数（不包含sign），按参数名 ASCII 码升序排序
拼接为key=value格式的字符串，并用&连接（如"appKey=test&keyword=手机"）
在拼接字符串末尾加上&appSecret=你的appSecret
对整个字符串进行 MD5 加密，得到 32 位大写签名值 - 返回数据结构解析
接口返回 JSON 格式数据，核心结构包括：
响应状态信息：请求状态码、消息、请求 ID
搜索结果统计：总商品数、总页数、当前页
商品列表：符合搜索条件的商品集合
筛选条件：可用的筛选选项（价格区间、品牌等）
商品列表中的关键数据字段：
productCode：商品唯一标识
productName：商品标题
picUrl：商品主图 URL
price：当前价格
marketPrice：市场价（原价）
commentCount：评价数量
averageScore：平均评分
brandName：品牌名称
categoryName：商品类目
salesCount：销售数量
isSelf：是否自营（1 是 0 否）
promotionFlag：是否有促销（1 是 0 否）
stockStatus：库存状态
三、Python 实现方案
以下是苏宁item_search接口的完整 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"
)
配置中文显示
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
plt.rcParams["axes.unicode_minus"] = False
class SuningItemSearch:
"""苏宁item_search接口封装类，用于根据关键词搜索商品并分析结果"""
def __init__(self, app_key: str, app_secret: str):
"""
初始化苏宁API客户端
:param app_key: 应用的appKey
:param app_secret: 应用的appSecret
"""
self.app_key = app_key
self.app_secret = app_secret
self.api_url = "https://open.suning.com/api/http/sopRequest"
# 频率控制
self.rate_limit = 10 # 默认基础权限，高级权限可修改为60次/分钟
self.call_timestamps = [] # 存储调用时间戳（秒级）
# 支持的排序方式
self.supported_sorts = [
"0", # 默认排序
"price asc", # 价格升序
"price desc", # 价格降序
"salesCount desc", # 销量降序
"averageScore desc"# 评分降序
]
def set_rate_limit(self, limit: int) -> None:
"""设置调用频率限制（次/分钟）"""
if 10 <= limit <= 60:
self.rate_limit = limit
logging.info(f"已设置调用频率限制为 {limit} 次/分钟")
else:
logging.warning("频率限制必须在10-60之间，未修改")
def _generate_sign(self, params: Dict) -> str:
"""生成签名（MD5算法）"""
# 1. 按参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为"key=value&key=value"格式
param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 拼接appSecret并进行MD5加密
sign_str = f"{param_str}&appSecret={self.app_secret}"
md5 = hashlib.md5()
md5.update(sign_str.encode('utf-8'))
sign = md5.hexdigest().upper() # 转为大写
return sign
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.1 # 额外加0.1秒保险
logging.warning(f"调用频率超限，等待 {sleep_time:.1f} 秒")
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 search_items(self, keyword: str,
page_no: int = 1, page_size: int = 50,
filters: Optional[Dict] = None) -> Optional[Dict]:
"""
按关键词搜索商品
:param keyword: 搜索关键词
:param page_no: 页码
:param page_size: 每页数量
:param filters: 筛选参数
:return: 搜索结果数据
"""
# 验证关键词
if not (2 <= len(keyword) <= 30):
logging.error(f"关键词长度必须在2-30个字符之间，当前长度: {len(keyword)}")
return None
# 验证排序方式
if filters and "sort" in filters and filters["sort"] not in self.supported_sorts:
logging.error(f"不支持的排序方式: {filters['sort']}，支持的方式: {', '.join(self.supported_sorts)}")
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"), # 时间戳格式：yyyyMMddHHmmss
"method": "suning.search.sale.query", # 苏宁商品搜索接口方法名
"version": "v1.2", # 接口版本
"keyword": keyword,
"pageNo": str(page_no),
"pageSize": str(page_size)
}
# 合并筛选参数
if filters and isinstance(filters, Dict):
# 过滤空值参数
valid_filters = {k: v for k, v in filters.items() if v is not None}
# 转换为字符串类型
for k, v in valid_filters.items():
base_params[k] = str(v).lower() if isinstance(v, bool) 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=10)
response.raise_for_status()
# 解析响应
result = response.json()
# 处理错误
if "sn_responseContent" not in result:
logging.error(f"API响应格式错误: {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.warning("未获取到搜索结果")
return None
logging.info(f"成功搜索关键词 '{keyword}'，第 {page_no} 页，找到 {search_data.get('totalCount', 0)} 个商品")
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_items(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_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)
# 计算总页数，不超过max_pages
meta_info["total_pages"] = min(max_pages,
(meta_info["total_items"] + page_size - 1) // page_size)
meta_info["available_brands"] = result.get("brandList", [])
meta_info["available_categories"] = result.get("categoryList", [])
page += 1
logging.info(f"批量搜索完成，共获取 {len(all_items)} 个商品数据")
return all_items, meta_info
def analyze_search_results(self, items: List[Dict], meta_info: Dict) -> Dict:
"""分析搜索结果数据"""
if not items:
return {"error": "没有商品数据可分析"}
keyword = meta_info.get("keyword", "未知关键词")
# 1. 价格分析
price_info = {
"min_price": None,
"max_price": None,
"avg_price": None,
"median_price": None,
"price_distribution": defaultdict(int),
"avg_discount": 0,
"discount_distribution": defaultdict(int)
}
prices = []
discounts = []
for item in items:
try:
price = float(item.get("price", 0))
prices.append(price)
# 价格区间分布
if price < 100:
price_info["price_distribution"]["<100"] += 1
elif price < 500:
price_info["price_distribution"]["100-500"] += 1
elif price < 1000:
price_info["price_distribution"]["500-1000"] += 1
elif price < 3000:
price_info["price_distribution"]["1000-3000"] += 1
else:
price_info["price_distribution"][">3000"] += 1
# 折扣分析
market_price = float(item.get("marketPrice", price))
if market_price > 0 and price > 0:
discount = round(price / market_price, 2)
discounts.append(discount)
# 折扣区间分布
if discount < 0.3:
price_info["discount_distribution"]["<3折"] += 1
elif discount < 0.5:
price_info["discount_distribution"]["3-5折"] += 1
elif discount < 0.7:
price_info["discount_distribution"]["5-7折"] += 1
elif discount < 0.9:
price_info["discount_distribution"]["7-9折"] += 1
else:
price_info["discount_distribution"]["9折以上"] += 1
except (ValueError, TypeError):
continue
if prices:
price_info["min_price"] = round(min(prices), 2)
price_info["max_price"] = round(max(prices), 2)
price_info["avg_price"] = round(sum(prices)/len(prices), 2)
# 计算中位数
sorted_prices = sorted(prices)
n = len(sorted_prices)
if n % 2 == 1:
price_info["median_price"] = round(sorted_prices[n//2], 2)
else:
price_info["median_price"] = round((sorted_prices[n//2-1] + sorted_prices[n//2])/2, 2)
if discounts:
price_info["avg_discount"] = round(sum(discounts)/len(discounts), 2)
price_info["price_distribution"] = dict(price_info["price_distribution"])
price_info["discount_distribution"] = dict(price_info["discount_distribution"])
# 2. 品牌分析
brand_info = {
"total_brands": 0,
"brand_distribution": defaultdict(int),
"top_brands": []
}
for item in items:
brand_name = item.get("brandName", "未知品牌")
brand_info["brand_distribution"][brand_name] += 1
brand_info["total_brands"] = len(brand_info["brand_distribution"])
# 获取前10个品牌
brand_info["top_brands"] = sorted(
brand_info["brand_distribution"].items(),
key=lambda x: x[1],
reverse=True
)[:10]
brand_info["brand_distribution"] = dict(brand_info["brand_distribution"])
# 3. 类目分析
category_info = {
"total_categories": 0,
"category_distribution": defaultdict(int),
"top_categories": []
}
for item in items:
category = item.get("categoryName", "未知类目")
# 只取末级类目
if ">" in category:
category = category.split(">")[-1]
category_info["category_distribution"][category] += 1
category_info["total_categories"] = len(category_info["category_distribution"])
# 获取前10个类目
category_info["top_categories"] = sorted(
category_info["category_distribution"].items(),
key=lambda x: x[1],
reverse=True
)[:10]
category_info["category_distribution"] = dict(category_info["category_distribution"])
# 4. 销售与评价分析
sales_info = {
"total_sales": 0,
"avg_sales": 0,
"sales_distribution": defaultdict(int),
"avg_rating": 0,
"rating_distribution": defaultdict(int),
"total_comments": 0
}
sales_counts = []
ratings = []
for item in items:
try:
# 销量分析
sales = int(item.get("salesCount", 0))
sales_counts.append(sales)
sales_info["total_sales"] += sales
# 销量区间分布
if sales == 0:
sales_info["sales_distribution"]["0"] += 1
elif sales < 10:
sales_info["sales_distribution"]["1-9"] += 1
elif sales < 50:
sales_info["sales_distribution"]["10-49"] += 1
elif sales < 100:
sales_info["sales_distribution"]["50-99"] += 1
elif sales < 500:
sales_info["sales_distribution"]["100-499"] += 1
else:
sales_info["sales_distribution"][">=500"] += 1
# 评价分析
rating = float(item.get("averageScore", 0))
if rating > 0:
ratings.append(rating)
# 评分区间分布
if rating < 3:
sales_info["rating_distribution"]["<3分"] += 1
elif rating < 4:
sales_info["rating_distribution"]["3-4分"] += 1
elif rating < 4.5:
sales_info["rating_distribution"]["4-4.5分"] += 1
else:
sales_info["rating_distribution"][">=4.5分"] += 1
# 总评价数
sales_info["total_comments"] += int(item.get("commentCount", 0))
except (ValueError, TypeError):
continue
if sales_counts:
sales_info["avg_sales"] = round(sum(sales_counts)/len(sales_counts), 1)
if ratings:
sales_info["avg_rating"] = round(sum(ratings)/len(ratings), 1)
sales_info["sales_distribution"] = dict(sales_info["sales_distribution"])
sales_info["rating_distribution"] = dict(sales_info["rating_distribution"])
# 5. 促销与自营分析
promotion_info = {
"promotion_rate": 0, # 促销商品占比
"self_rate": 0, # 自营商品占比
"promotion_self_rate": 0 # 自营且促销商品占比
}
promotion_count = 0
self_count = 0
promotion_self_count = 0
for item in items:
is_promotion = item.get("promotionFlag", "0") == "1"
is_self = item.get("isSelf", "0") == "1"
if is_promotion:
promotion_count += 1
if is_self:
self_count += 1
if is_promotion and is_self:
promotion_self_count += 1
# 计算比例
total = len(items)
if total > 0:
promotion_info["promotion_rate"] = round(promotion_count / total, 2)
promotion_info["self_rate"] = round(self_count / total, 2)
promotion_info["promotion_self_rate"] = round(promotion_self_count / total, 2)
return {
"search_keyword": keyword,
"total_items": len(items),
"price_analysis": price_info,
"brand_analysis": brand_info,
"category_analysis": category_info,
"sales_analysis": sales_info,
"promotion_analysis": promotion_info,
"analysis_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
def get_top_items(self, items: List[Dict], by: str = "sales", top_n: int = 10) -> List[Dict]:
"""
获取特定维度排名靠前的商品
:param items: 商品列表
:param by: 排序维度：sales(销量), price(价格), rating(评分), discount(折扣)
:param top_n: 取前N名
:return: 排序后的商品列表
"""
if not items:
return []
valid_dimensions = ["sales", "price", "rating", "discount"]
if by not in valid_dimensions:
logging.warning(f"无效的排序维度: {by}，使用默认值 'sales'")
by = "sales"
# 准备排序数据
ranked_items = []
for item in items:
try:
if by == "sales":
# 销量排序是降序
value = int(item.get("salesCount", 0))
sort_value = -value
elif by == "price":
# 价格排序是升序
value = float(item.get("price", 0))
sort_value = value
elif by == "rating":
# 评分排序是降序
value = float(item.get("averageScore", 0))
sort_value = -value
elif by == "discount":
# 折扣排序是升序（折扣越低越优惠）
price = float(item.get("price", 0))
market_price = float(item.get("marketPrice", price))
if market_price == 0:
value = 1.0 # 无原价视为无折扣
else:
value = round(price / market_price, 2)
sort_value = value
ranked_items.append((item, sort_value, value))
except (ValueError, TypeError):
continue
# 排序并返回前N名
ranked_items.sort(key=lambda x: x[1])
return [{"item": p[0], "value": p[2]} for p in ranked_items[:top_n]]
def visualize_analysis(self, analysis: Dict, output_dir: str = ".") -> None:
"""可视化分析结果"""
if "error" in analysis:
logging.warning(analysis["error"])
return
keyword = analysis.get("search_keyword", "未知关键词")
# 1. 价格分布直方图
if "price_analysis" in analysis and analysis["price_analysis"]["price_distribution"]:
plt.figure(figsize=(10, 6))
price_bins = sorted(analysis["price_analysis"]["price_distribution"].keys(),
key=lambda x: float(re.findall(r'\d+', x)[0]) if re.findall(r'\d+', x) else 0)
counts = [analysis["price_analysis"]["price_distribution"][b] for b in price_bins]
plt.bar(price_bins, counts, color='lightblue')
plt.title(f'关键词 "{keyword}" 价格分布')
plt.xlabel('价格区间 (元)')
plt.ylabel('商品数量')
for i, v in enumerate(counts):
plt.text(i, v + 0.5, str(v), ha='center')
plt.tight_layout()
plt.savefig(f"{output_dir}/suning_price_distribution.png")
plt.close()
logging.info(f"价格分布图表已保存至 {output_dir}/suning_price_distribution.png")
# 2. 折扣分布饼图
if "price_analysis" in analysis and analysis["price_analysis"]["discount_distribution"]:
plt.figure(figsize=(10, 8))
discount_data = analysis["price_analysis"]["discount_distribution"]
labels = list(discount_data.keys())
sizes = list(discount_data.values())
plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)
plt.title(f'关键词 "{keyword}" 折扣分布')
plt.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/suning_discount_distribution.png")
plt.close()
logging.info(f"折扣分布图表已保存至 {output_dir}/suning_discount_distribution.png")
# 3. 品牌分布条形图（前10名）
if "brand_analysis" in analysis and analysis["brand_analysis"]["top_brands"]:
plt.figure(figsize=(12, 6))
brands, counts = zip(*analysis["brand_analysis"]["top_brands"])
plt.bar(brands, counts, color='purple')
plt.title(f'关键词 "{keyword}" 品牌分布（前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}/suning_brand_distribution.png")
plt.close()
logging.info(f"品牌分布图表已保存至 {output_dir}/suning_brand_distribution.png")
# 4. 销量分布条形图
if "sales_analysis" in analysis and analysis["sales_analysis"]["sales_distribution"]:
plt.figure(figsize=(10, 6))
sales_bins = sorted(analysis["sales_analysis"]["sales_distribution"].keys(),
key=lambda x: int(re.findall(r'\d+', x)[0]) if re.findall(r'\d+', x) else -1)
counts = [analysis["sales_analysis"]["sales_distribution"][b] for b in sales_bins]
plt.bar(sales_bins, counts, color='orange')
plt.title(f'关键词 "{keyword}" 销量分布')
plt.xlabel('销量区间')
plt.ylabel('商品数量')
for i, v in enumerate(counts):
plt.text(i, v + 0.5, str(v), ha='center')
plt.tight_layout()
plt.savefig(f"{output_dir}/suning_sales_distribution.png")
plt.close()
logging.info(f"销量分布图表已保存至 {output_dir}/suning_sales_distribution.png")
# 5. 评分分布饼图
if "sales_analysis" in analysis and analysis["sales_analysis"]["rating_distribution"]:
plt.figure(figsize=(8, 8))
rating_data = analysis["sales_analysis"]["rating_distribution"]
labels = list(rating_data.keys())
sizes = list(rating_data.values())
plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)
plt.title(f'关键词 "{keyword}" 评分分布')
plt.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/suning_rating_distribution.png")
plt.close()
logging.info(f"评分分布图表已保存至 {output_dir}/suning_rating_distribution.png")
# 6. 促销与自营商品占比饼图
if "promotion_analysis" in analysis:
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
# 促销商品占比
total = analysis["total_items"]
has_promo = round(analysis["promotion_analysis"]["promotion_rate"] * total)
no_promo = total - has_promo
labels1 = ['促销商品', '非促销商品']
sizes1 = [has_promo, no_promo]
ax1.pie(sizes1, labels=labels1, autopct='%1.1f%%', startangle=90)
ax1.set_title(f'促销商品占比')
ax1.axis('equal')
# 自营商品占比
has_self = round(analysis["promotion_analysis"]["self_rate"] * total)
no_self = total - has_self
labels2 = ['自营商品', '第三方商品']
sizes2 = [has_self, no_self]
ax2.pie(sizes2, labels=labels2, autopct='%1.1f%%', startangle=90)
ax2.set_title(f'自营商品占比')
ax2.axis('equal')
plt.tight_layout()
plt.savefig(f"{output_dir}/suning_promotion_self_ratio.png")
plt.close()
logging.info(f"促销与自营占比图表已保存至 {output_dir}/suning_promotion_self_ratio.png")
def export_to_excel(self, items: List[Dict], analysis: Dict, meta_info: Dict,
filename: str) -> None:
"""导出搜索结果和分析到Excel"""
if not items and not analysis:
logging.warning("没有数据可导出")
return
try:
with pd.ExcelWriter(filename) as writer:
# 搜索信息
pd.DataFrame([meta_info]).to_excel(writer, sheet_name='搜索信息', index=False)
# 商品列表
if items:
item_list = []
for item in items:
# 计算折扣
try:
price = float(item.get("price", 0))
market_price = float(item.get("marketPrice", price))
discount = round(price / market_price, 2) if market_price > 0 else 1.0
except (ValueError, TypeError):
discount = 1.0
item_data = {
"商品ID": item.get("productCode"),
"标题": item.get("productName"),
"类目": item.get("categoryName"),
"品牌": item.get("brandName"),
"当前价(元)": item.get("price"),
"市场价(元)": item.get("marketPrice"),
"折扣": discount,
"销量": item.get("salesCount"),
"评价数": item.get("commentCount"),
"评分": item.get("averageScore"),
"是否自营": "是" if item.get("isSelf", "0") == "1" else "否",
"是否促销": "是" if item.get("promotionFlag", "0") == "1" else "否"
}
item_list.append(item_data)
df_items = pd.DataFrame(item_list)
df_items.to_excel(writer, sheet_name='商品列表', index=False)
# 分析结果
if analysis and "error" not in analysis:
# 价格分析
df_price = pd.DataFrame([analysis["price_analysis"]])
df_price.to_excel(writer, sheet_name='价格分析', index=False)
# 折扣分布
df_discount = pd.DataFrame(list(analysis["price_analysis"]["discount_distribution"].items()),
columns=['折扣区间', '数量'])
df_discount.to_excel(writer, sheet_name='折扣分布', index=False)
# 品牌分析
df_brand = pd.DataFrame(list(analysis["brand_analysis"]["brand_distribution"].items()),
columns=['品牌', '数量'])
df_brand.to_excel(writer, sheet_name='品牌分析', index=False)
# 类目分析
df_category = pd.DataFrame(list(analysis["category_analysis"]["category_distribution"].items()),
columns=['类目', '数量'])
df_category.to_excel(writer, sheet_name='类目分析', index=False)
# 销量分析
df_sales = pd.DataFrame(list(analysis["sales_analysis"]["sales_distribution"].items()),
columns=['销量区间', '数量'])
df_sales.to_excel(writer, sheet_name='销量分析', index=False)
logging.info(f"数据已导出至 {filename}")
except Exception as e:
logging.error(f"导出Excel失败: {e}")
示例调用
if name == "main":
# 替换为实际的参数（从苏宁开放平台获取）
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
KEYWORD = "智能手机" # 搜索关键词
# 初始化API客户端
suning_search = SuningItemSearch(APP_KEY, APP_SECRET)
# 若为高级权限，设置更高的频率限制
# suning_search.set_rate_limit(60)
# 1. 设置搜索筛选条件
filters = {
"sort": "salesCount desc", # 按销量排序
"priceFrom": 1000, # 最低价格
"priceTo": 5000, # 最高价格
"hasPromotion": "1", # 只看促销商品
"isSelf": "1", # 只看自营商品
"area": "110100" # 北京地区
}
# 2. 批量获取搜索结果
print("=== 搜索商品 ===")
items, meta_info = suning_search.batch_search_items(
keyword=KEYWORD,
max_pages=3, # 获取前3页
page_size=50,
filters=filters
)
if items:
print(f"搜索关键词: {KEYWORD}")
print(f"获取商品数量: {len(items)}")
print(f"总商品数量: {meta_info['total_items']}")
print(f"总页数: {meta_info['total_pages']}")
# 3. 分析搜索结果
print("\n=== 搜索结果分析 ===")
if items:
analysis = suning_search.analyze_search_results(items, meta_info)
print("价格分析:")
print(f" 价格范围: {analysis['price_analysis']['min_price']} - {analysis['price_analysis']['max_price']} 元")
print(f" 平均价格: {analysis['price_analysis']['avg_price']} 元")
print(f" 平均折扣: {analysis['price_analysis']['avg_discount']} (约{round(analysis['price_analysis']['avg_discount']*10, 1)}折)")
print("\n品牌分析:")
print(f" 品牌总数: {analysis['brand_analysis']['total_brands']}")
print(" 主要品牌:")
for brand, count in analysis["brand_analysis"]["top_brands"][:5]:
print(f" {brand}: {count}个商品")
print("\n类目分析:")
print(f" 类目总数: {analysis['category_analysis']['total_categories']}")
print(" 主要类目:")
for category, count in analysis["category_analysis"]["top_categories"][:5]:
print(f" {category}: {count}个商品")
print("\n销售分析:")
print(f" 总销量: {analysis['sales_analysis']['total_sales']}")
print(f" 平均销量: {analysis['sales_analysis']['avg_sales']}")
print(f" 平均评分: {analysis['sales_analysis']['avg_rating']}")
print(f" 总评价数: {analysis['sales_analysis']['total_comments']}")
print("\n促销与自营分析:")
print(f" 促销商品占比: {analysis['promotion_analysis']['promotion_rate']:.0%}")
print(f" 自营商品占比: {analysis['promotion_analysis']['self_rate']:.0%}")
print(f" 自营且促销商品占比: {analysis['promotion_analysis']['promotion_self_rate']:.0%}")
# 4. 获取各维度TOP商品
print("\n=== 各维度TOP3商品 ===")
top_sales = suning_search.get_top_items(items, by="sales", top_n=3)
print("销量TOP3:")
for i, item_info in enumerate(top_sales, 1):
item = item_info["item"]
print(f"{i}. {item['productName'][:30]}... 销量: {item_info['value']}")
top_rating = suning_search.get_top_items(items, by="rating", top_n=3)
print("\n评分TOP3:")
for i, item_info in enumerate(top_rating, 1):
item = item_info["item"]
print(f"{i}. {item['productName'][:30]}... 评分: {item_info['value']}")
top_cheap = suning_search.get_top_items(items, by="price", top_n=3)
print("\n价格最低TOP3:")
for i, item_info in enumerate(top_cheap, 1):
item = item_info["item"]
print(f"{i}. {item['productName'][:30]}... 价格: {item_info['value']} 元")
top_discount = suning_search.get_top_items(items, by="discount", top_n=3)
print("\n折扣最低TOP3:")
for i, item_info in enumerate(top_discount, 1):
item = item_info["item"]
print(f"{i}. {item['productName'][:30]}... 折扣: {item_info['value']} ({round(item_info['value']*10, 1)}折)")
# 5. 可视化分析结果
suning_search.visualize_analysis(analysis)
# 6. 导出数据到Excel
suning_search.export_to_excel(items, analysis, meta_info, f"苏宁搜索_{KEYWORD.replace(' ', '_')}_分析.xlsx")
else:
print("未获取到搜索结果，无法进行分析")
四、接口调用注意事项
- 常见错误及解决方案
错误码 说明 解决方案
1001 无效的 appKey 检查 appKey 是否正确，应用是否已通过审核
1002 签名错误 检查签名算法实现，确保使用 MD5 和正确的 appSecret
1003 权限不足 申请更高权限或检查是否已开通 item_search 接口权限
1004 调用频率超限 降低调用频率，或申请提高权限等级
2001 关键词无效 检查关键词长度是否在 2-30 个字符之间，是否包含非法字符
2002 参数错误 检查 pageSize 是否在 10-50 之间，sort 是否为支持的类型
2003 类目 ID 无效 验证 categoryId 是否正确，是否存在该类目
2004 品牌 ID 无效 验证 brandId 是否正确，是否存在这些品牌
3001 服务暂时不可用 稍后重试，可能是平台维护或临时故障
4001 账号被限制 检查账号状态，是否有违规行为导致限制 - 性能优化建议
精准筛选：使用价格区间、品牌、类目等筛选条件减少返回数据量
合理设置分页：使用最大 pageSize（50）减少请求次数，降低频率限制影响
缓存策略：对相同关键词和筛选条件的搜索结果设置 5-10 分钟缓存
批量处理控制：多关键词搜索时，使用队列控制并发，避免触发频率限制
关键词优化：使用精准关键词减少无关结果，提高数据质量
地区选择：根据业务需求选择合适的 region 参数，避免获取无关地区数据
增量更新：定期搜索时只获取新增或有变化的商品数据
五、应用场景与扩展建议
典型应用场景
市场趋势分析：通过关键词搜索分析特定品类商品的市场状况和趋势
竞品价格监控：跟踪同类商品的价格和促销情况，优化自身定价策略
品牌市场份额分析：了解特定品类中各品牌的市场表现和份额
促销活动效果评估：分析促销商品的特征和销售情况，评估促销效果
新品市场调研：通过搜索了解新品的市场接受度和竞争情况
用户需求分析：基于搜索结果分析用户偏好和需求特征
扩展建议
多关键词对比：同时分析多个相关关键词的搜索结果，全面了解市场
趋势追踪：定期执行相同搜索，跟踪市场变化和趋势
价格带分析：深入分析不同价格带商品的特征和市场表现
季节性分析：结合时间维度，分析不同季节的商品特征变化
关键词扩展：基于初始搜索结果，自动生成相关关键词并进行扩展搜索
智能推荐：基于搜索分析结果，自动推荐有潜力的商品或品牌
促销策略优化：分析促销商品特征，优化促销策略和选品
通过item_search接口获取的苏宁商品搜索数据，能够帮助电商从业者全面了解特定关键词相关的商品市场状况，为产品选型、定价策略和促销活动提供数据支持。在实际应用中，需结合苏宁平台的自营与第三方混合模式特点，制定针对性的经营策略，同时严格遵守平台的数据使用规范，确保合规使用 API 接口。
