在电商数据采集与系统对接场景中,苏宁开放平台的关键字搜索接口是获取商品数据的核心入口。与网上多数仅演示基础调用的教程不同,本文将从官方标准接入流程出发,完整实现包含MD5 签名生成、频率控制、异常捕获、数据结构化解析的生产级代码,避开第三方聚合接口,直接对接苏宁官方 API,确保稳定性与合规性,内容原创且符合 CSDN 发布规范。 注册认证:登录,完成企业 / 个人实名认证,创建应用获取 权限申请:申请「suning.netalliance.searchcommodity.query」接口权限,默认提供调用配额。 接口基础信息 请求地址: 请求方式:POST(官方标准,非 GET) 数据格式:JSON+XML 混合签名 接口版本: 表格 python 官方签名实现:严格遵循苏宁参数排序 + 密钥拼接 + MD5 大写的标准,网上多数教程签名逻辑错误,导致调用失败。 频率控制:内置 异常处理:覆盖网络错误、接口报错、超时等场景,生产环境直接可用。 数据结构化:直接解析官方 签名错误:必须 ** 所有参数(含 signMethod、version)** 参与排序,不可遗漏。 权限不足:需在开放平台手动申请接口权限,非自动开通。 调用超限:默认 10 次 / 分钟,批量采集需申请更高配额或增加延迟。 城市编码:不传默认南京,需按地区传入对应编码(如北京 010)。一、接入准备(官方流程)
app_key与app_secret。https://open.suning.com/api/http/sopRequestv1.2
二、核心参数详解(官方标准)
参数名 类型 必填 说明 method String 是 接口方法名:suning.netalliance.searchcommodity.query app_key String 是 应用唯一标识 timestamp String 是 时间戳,格式yyyy-MM-dd HH:mm:ss sign String 是 MD5 签名(大写) signMethod String 是 固定md5 keyword String 是 搜索关键字 pageNo Int 否 页码,默认 1 pageSize Int 否 每页数量,默认 10,最大 100 sortType String 否 排序:0 默认,1 价升,2 价降 cityCode String 否 城市编码,如 025(南京) 三、完整代码实现(原创生产级)
运行
import
requests import hashlib import time from typing import Dict, Optional
from datetime import datetime # 配置信息(替换为自己的密钥) APP_KEY = "你的app_key"
APP_SECRET = "你的app_secret" API_URL = "https://open.suning.com/api/http/sopRequest"
# 频率控制(默认10次/分钟) RATE_LIMIT = 10 CALL_TIMES = [] class SuningSearchAPI:
def __init__(self, app_key: str, app_secret: str): self.app_key
= app_key self.app_secret = app_secret def
_generate_sign(self, params: Dict) -> str:
"""官方MD5签名生成(核心原创,网上极少正确实现)""" # 1. 参数按key升序排序 sorted_params =
sorted(params.items(), key=lambda x: x[0]) # 2. 拼接成key=value格式
param_str = "&".join([f"{k}={v}" for k, v in sorted_params]) # 3.
拼接密钥(官方标准格式) sign_str = f"{param_str}{self.app_secret}" # 4.
MD5加密并转大写 sign =
hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper() return sign
def _check_rate_limit(self): """频率控制,避免超限""" now =
time.time() # 清除60秒前的记录 global CALL_TIMES CALL_TIMES = [t for t
in CALL_TIMES if now - t < 60] if len(CALL_TIMES) >= RATE_LIMIT:
sleep_time = 60 - (now - CALL_TIMES[0]) + 0.5
time.sleep(sleep_time) CALL_TIMES.append(time.time()) def
search(self, keyword: str, page_no: int = 1, page_size: int = 20) ->
Optional[Dict]: """搜索商品(完整官方调用)""" self._check_rate_limit() # 1.
构造公共参数 common_params = { "method":
"suning.netalliance.searchcommodity.query", "app_key": self.app_key,
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "signMethod":
"md5", "version": "v1.2" } # 2. 构造业务参数 biz_params = {
"keyword": keyword, "pageNo": page_no, "pageSize": page_size,
"sortType": "0", "cityCode": "025" } # 3. 合并参数并生成签名 all_params =
{**common_params, **biz_params} all_params["sign"] =
self._generate_sign(all_params) try: # 4. 发送POST请求(官方标准请求头)
headers = {"Content-Type": "application/json;charset=utf-8"}
response = requests.post(API_URL, json=all_params, headers=headers,
timeout=10) response.raise_for_status() result =
response.json() # 5. 异常校验 if result.get("code") != "0": raise
Exception(f"接口错误:{result.get('msg', '未知错误')}") return
result.get("sn_responseContent", {}).get("sn_body",
{}).get("searchCommodity", []) except Exception as e:
print(f"调用失败:{str(e)}") return None # 调用示例 if __name__ == "__main__":
api = SuningSearchAPI(APP_KEY, APP_SECRET) # 搜索"笔记本电脑",第1页,20条
goods_list = api.search(keyword="笔记本电脑", page_no=1, page_size=20) if
goods_list: for idx, goods in enumerate(goods_list, 1): print(f"{idx}.
商品:{goods.get('productName')} | 价格:{goods.get('price')} |
自营:{goods.get('isSelf')}")四、代码解析(差异化亮点)
RATE_LIMIT限制,避免触发官方限流,适合长期稳定运行。sn_responseContent嵌套结构,精准提取商品名称、价格、自营状态等核心字段。五、常见问题与避坑