前言 苏宁开放平台关键词搜索接口，多用于比价、商品库批量拉取场景。和普通搜索接口不一样，它存在两个典型痛点：分页到末尾时不会直接返回空数组，而是返回少量重复商品；同时促销价、划线价混杂在返回字段中，直接入库容易造成价格错乱。本文不只是简单调用接口，重点实现分页循环终止判断、价格字段分离、商品基础信息扁平化封装，适合批量商品检索业务。 前置准备 Python3，依赖 requests 库。需要在苏宁云台申请 AppKey、AppSecret，接口采用 MD5 签名鉴权。接口存在 QPS 限制，批量循环分页必须增加请求间隔，否则会触发接口限流返回业务错误码。 完整代码示例 核心逻辑解析 签名生成：参数按 ASCII 排序后拼接密钥生成 MD5 大写签名，严格匹配平台鉴权规则。 自动分页 + SKU 去重：自定义分页终止逻辑，用 sku 集合做去重，解决接口尾页返回重复商品的问题。 价格字段分离：区分售价与市场价，避免促销活动期间价格字段混淆。 分层异常处理：区分网络异常、业务接口报错，网络抖动自动重试。 线上踩坑要点 关键词编码：中文关键词直接传入，内部不需要额外 url 编码，编码错误会返回空列表。 分页上限：接口有最大返回页数限制，超过后不会报错，只会重复返回前面数据，必须做 SKU 去重判断。 商品状态过滤：下架商品依旧出现在搜索列表，入库前需要 status 字段过滤在售商品。 批量请求控制：多关键词循环查询时，建议使用队列，控制并发，高频调用会触发权限拦截。
import requests
import time
import hashlib
class SuningSearchClient:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.session = requests.Session()
self.retry_count = 2
self.sleep_interval = 1.2
def build_sign(self, params):
# 参数按key升序拼接，生成MD5大写签名
sorted_items = sorted(params.items(), key=lambda x: x[0])
raw_str = ''.join(f"{k}{v}" for k, v in sorted_items) + self.app_secret
return hashlib.md5(raw_str.encode('utf-8')).hexdigest().upper()
def search_item_list(self, keyword, page_index=1, page_size=10):
api_url = "https://open.suning.com/api/http/sopRequest"
req_params = {
"appKey": self.app_key,
"keyword": keyword,
"pageIndex": str(page_index),
"pageSize": str(page_size)
}
req_params["sign"] = self.build_sign(req_params)
for attempt in range(self.retry_count):
try:
resp = self.session.post(api_url, json=req_params, timeout=10)
time.sleep(self.sleep_interval)
if resp.status_code != 200:
return {"code": -2, "data": [], "msg": "HTTP请求异常"}
res_json = resp.json()
if res_json.get("sn_error"):
err = res_json["sn_error"]
return {"code": -1, "data": [], "msg": f"{err['error_code']}:{err['error_msg']}"}
list_raw = res_json.get("sn_body", {}).get("searchCommodity", [])
result_list = []
for item in list_raw:
item_data = {
"sku_id": item.get("skuId"),
"title": item.get("commodityName", "").strip(),
"price": item.get("salePrice"),
"market_price": item.get("marketPrice"),
"pic_url": item.get("commodityPic", ""),
"brand_name": item.get("brandName", ""),
"is_sale": int(item.get("status", 0)) == 1
}
result_list.append(item_data)
return {"code":0, "data":result_list, "msg":"success"}
except Exception as e:
if attempt < self.retry_count -1:
time.sleep(2)
continue
return {"code": -3, "data": [], "msg": str(e)}
def get_all_by_keyword(self, keyword, max_page=5):
"""自动分页拉取，通过sku去重，防止末尾重复数据"""
all_items = []
seen_skus = set()
for page in range(1, max_page+1):
res = self.search_item_list(keyword, page_index=page)
if res["code"] !=0:
print(f"第{page}页查询失败：{res['msg']}")
break
batch = res["data"]
if not batch:
break
new_items = []
for goods in batch:
if goods["sku_id"] and goods["sku_id"] not in seen_skus:
seen_skus.add(goods["sku_id"])
new_items.append(goods)
if not new_items:
# 没有新增商品，终止分页
break
all_items.extend(new_items)
return all_items
if __name__ == "__main__":
client = SuningSearchClient(app_key="your_appkey", app_secret="your_appsecret")
goods_list = client.get_all_by_keyword("蓝牙耳机", max_page=3)
for goods in goods_list:
print(goods["sku_id"], goods["title"], goods["price"])
