当前位置：首页 > news >正文

苏宁item_search - 按关键字搜索商品接口深度分析及 Python 实现

news 2025/9/28 9:22:50

苏宁的item_search接口是按关键字搜索商品列表的核心接口，为开发者提供了通过关键词、分类、价格区间等条件筛选商品的能力。该接口广泛应用于电商数据分析、竞品研究、价格监控等场景，能够高效获取精准的商品列表及关键信息。

一、苏宁 item_search 接口核心特性分析

1. 接口定位与核心价值

苏宁item_search接口作为平台商品搜索功能，其核心价值体现在：

支持多维度筛选，精准定位目标商品
提供商品基础信息、价格、销量、评价等核心数据
支持分页获取，可获取大规模商品数据
可结合促销标签筛选优惠商品
提供多种排序功能，快速定位热门或优质商品
支持按地区筛选，获取区域化商品信息

2. 接口权限与调用限制

使用苏宁搜索接口需遵守以下规范：

限制类型	具体规则	说明
权限要求	需注册苏宁开发者账号，创建应用并通过审核	个人与企业开发者权限不同，企业开发者可获取更完整数据
调用频率	基础权限：10 次 / 分钟；高级权限：60 次 / 分钟	按 appKey 和 IP 双重限制
数据返回	单次最多返回 50 条商品，最多支持 100 页数据	搜索结果有 5 分钟缓存
字段限制	基础字段免费，部分高级字段需申请开通	敏感信息受平台保护
搜索限制	关键词长度限制为 2-30 个字符	支持中英文及部分特殊符号

3. 核心参数解析

必选参数

参数名	类型	说明	示例
`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"`（否）

二、签名生成与返回数据结构

1. 签名生成逻辑

苏宁开放平台采用 MD5 签名算法，步骤如下：

收集所有请求参数（不包含sign），按参数名 ASCII 码升序排序
拼接为key=value格式的字符串，并用&连接（如"appKey=test&keyword=手机"）
在拼接字符串末尾加上&appSecret=你的appSecret
对整个字符串进行 MD5 加密，得到 32 位大写签名值

2. 返回数据结构解析

接口返回 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
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 配置日志
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"] = Falseclass 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_keyself.app_secret = app_secretself.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 = limitlogging.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 signdef _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 Noneresponse_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 Nonelogging.info(f"成功搜索关键词 '{keyword}'，第 {page_no} 页，找到 {search_data.get('totalCount', 0)} 个商品")return search_dataexcept requests.exceptions.RequestException as e:logging.error(f"请求异常: {str(e)}")return Noneexcept json.JSONDecodeError:logging.error(f"响应解析失败: {response.text[:200]}...")return Nonedef 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 = 1while 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("当前页无商品数据，停止获取")breakall_items.extend(items)# 保存元信息（第一页）if page == 1:meta_info["total_items"] = result.get("totalCount", 0)# 计算总页数，不超过max_pagesmeta_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 += 1logging.info(f"批量搜索完成，共获取 {len(all_items)} 个商品数据")return all_items, meta_infodef 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"] += 1elif price < 500:price_info["price_distribution"]["100-500"] += 1elif price < 1000:price_info["price_distribution"]["500-1000"] += 1elif price < 3000:price_info["price_distribution"]["1000-3000"] += 1else: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折"] += 1elif discount < 0.5:price_info["discount_distribution"]["3-5折"] += 1elif discount < 0.7:price_info["discount_distribution"]["5-7折"] += 1elif discount < 0.9:price_info["discount_distribution"]["7-9折"] += 1else:price_info["discount_distribution"]["9折以上"] += 1except (ValueError, TypeError):continueif 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] += 1brand_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] += 1category_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"] += 1elif sales < 10:sales_info["sales_distribution"]["1-9"] += 1elif sales < 50:sales_info["sales_distribution"]["10-49"] += 1elif sales < 100:sales_info["sales_distribution"]["50-99"] += 1elif sales < 500:sales_info["sales_distribution"]["100-499"] += 1else: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分"] += 1elif rating < 4:sales_info["rating_distribution"]["3-4分"] += 1elif rating < 4.5:sales_info["rating_distribution"]["4-4.5分"] += 1else:sales_info["rating_distribution"][">=4.5分"] += 1# 总评价数sales_info["total_comments"] += int(item.get("commentCount", 0))except (ValueError, TypeError):continueif 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 = 0self_count = 0promotion_self_count = 0for item in items:is_promotion = item.get("promotionFlag", "0") == "1"is_self = item.get("isSelf", "0") == "1"if is_promotion:promotion_count += 1if is_self:self_count += 1if 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 = -valueelif by == "price":# 价格排序是升序value = float(item.get("price", 0))sort_value = valueelif by == "rating":# 评分排序是降序value = float(item.get("averageScore", 0))sort_value = -valueelif 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 = valueranked_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"])returnkeyword = 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_promolabels1 = ['促销商品', '非促销商品']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_selflabels2 = ['自营商品', '第三方商品']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("没有数据可导出")returntry: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.0except (ValueError, TypeError):discount = 1.0item_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}")# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 示例调用
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. 导出数据到Excelsuning_search.export_to_excel(items, analysis, meta_info, f"苏宁搜索_{KEYWORD.replace(' ', '_')}_分析.xlsx")else:print("未获取到搜索结果，无法进行分析")

四、接口调用注意事项

1. 常见错误及解决方案

错误码	说明	解决方案
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	账号被限制	检查账号状态，是否有违规行为导致限制