苏宁易购作为国内领先的综合电商平台,其 API 体系分为**官方开放平台接口**(需认证)和**非官方网页 / 移动端接口**(公开信息抓取)。
以下从接口特性、认证机制、核心功能展开分析,并提供两种场景的 Python 实现方案。
### 一、苏宁 API 核心特性分析
#### 1. 官方开放平台 API
苏宁开发者平台提供规范化接口,适用于商业合作,核心特点:
- **功能覆盖**:商品管理、订单处理、库存查询、营销活动等全链路电商能力。
- **认证机制**:基于`appKey + appSecret + 签名`的认证体系,签名通过 MD5 加密生成(规则:参数升序拼接 +`appSecret`后加密)。
- **接口风格**:RESTful 风格,支持 GET/POST,响应格式为 JSON,网关地址统一为`https://open.suning.com/api/http/sopRequest`。
- **典型接口**:
- 商品详情:(获取价格、库存、优惠券);
- 库存查询:;
- 订单创建:(需权限申请)。
#### 2. 非官方接口(网页 / 移动端)
通过抓包分析的公开信息接口,适用于信息查询,特点:
- **功能限制**:仅支持商品搜索、详情查看等公开信息,无订单 / 库存操作权限。
- **反爬机制**:
- 动态参数:搜索接口含时间戳(`_`)和随机回调函数(`callback`);
- User-Agent 检测:需模拟浏览器 / 移动端请求;
- 频率限制:高频请求触发 IP 封禁或验证码。
- **数据格式**:搜索结果为 JSONP(需剥离回调函数),详情页数据嵌在 HTML 或页面脚本中。
### 二、Python 脚本实现
#### 方案 1:官方 API 调用(需开发者账号)
适用于需要稳定数据和商业用途的场景,需先在苏宁开放平台注册并获取`appKey`和`appSecret`。
import requests
import json
import time
import hashlib
import logging
from typing import Dict, Optional
from requests.exceptions import RequestException
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class SuningOfficialAPI:
def __init__(self, app_key: str, app_secret: str):
"""初始化苏宁官方API客户端"""
self.app_key = app_key
self.app_secret = app_secret
self.base_url = "https://open.suning.com/api/http/sopRequest"
self.session = requests.Session()
def _generate_sign(self, params: Dict) -> str:
"""生成签名(苏宁官方规则)"""
# 1. 参数按ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 拼接appSecret并MD5加密(大写)
sign_str += self.app_secret
return hashlib.md5(sign_str.encode()).hexdigest().upper()
def _get_timestamp(self) -> str:
"""生成时间戳(格式:yyyyMMddHHmmss)"""
return time.strftime("%Y%m%d%H%M%S", time.localtime())
def call_api(self, method: str, biz_params: Dict) -> Optional[Dict]:
"""通用API调用方法"""
# 公共参数
public_params = {
"appKey": self.app_key,
"method": method,
"timestamp": self._get_timestamp(),
"format": "json",
"version": "v1.2"
}
# 合并参数并生成签名
all_params = {**public_params,** biz_params}
all_params["sign"] = self._generate_sign(all_params)
try:
response = self.session.get(
self.base_url,
params=all_params,
timeout=10
)
response.raise_for_status()
result = response.json()
# 处理错误响应
if result.get("code") != "0":
logging.error(f"API错误:{result.get('msg')}(错误码:{result.get('code')})")
return None
return result.get("body", {})
except RequestException as e:
logging.error(f"请求失败:{str(e)}")
return None
def get_product_detail(self, product_code: str) -> Optional[Dict]:
"""获取商品详情(含价格、优惠券)"""
method = "com.suning.govbus.product.getItem"
params = {"productCode": product_code}
result = self.call_api(method, params)
if not result or "productInfo" not in result:
return None
product = result["productInfo"]
# 解析价格和优惠
price = float(product.get("price", 0))
promotion_price = float(product.get("promotionPrice", price))
# 解析优惠券
coupons = []
for coupon in product.get("couponList", []):
coupons.append({
"id": coupon.get("couponId"),
"discount": float(coupon.get("discount", 0)), # 优惠金额
"min_amount": float(coupon.get("minAmount", 0)), # 使用门槛
"valid_time": coupon.get("validTime")
})
return {
"product_code": product_code,
"name": product.get("productName"),
"price": price,
"promotion_price": promotion_price,
"stock": product.get("stockFlag"), # 库存状态(有货/无货)
"coupons": coupons,
"shop_name": product.get("shopInfo", {}).get("shopName"),
"brand": product.get("brandName")
}
# 示例调用
if __name__ == "__main__":
# 替换为实际参数(从苏宁开放平台获取)
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
PRODUCT_CODE = "1000123456" # 商品编码(从商品URL提取)
api = SuningOfficialAPI(APP_KEY, APP_SECRET)
detail = api.get_product_detail(PRODUCT_CODE)
if detail:
print(f"商品名称:{detail['name']}")
print(f"原价:{detail['price']}元 | 促销价:{detail['promotion_price']}元")
print(f"库存:{detail['stock']} | 店铺:{detail['shop_name']}")
if detail["coupons"]:
print("可用优惠券:")
for coupon in detail["coupons"]:
print(f"- 满{coupon['min_amount']}元减{coupon['discount']}元({coupon['valid_time']})")
方案 2:非官方接口(网页抓取)
import requests
import json
import time
import hashlib
import logging
from typing import Dict, Optional
from requests.exceptions import RequestException
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class SuningOfficialAPI:
def __init__(self, app_key: str, app_secret: str):
"""初始化苏宁官方API客户端"""
self.app_key = app_key
self.app_secret = app_secret
self.base_url = "https://open.suning.com/api/http/sopRequest"
self.session = requests.Session()
def _generate_sign(self, params: Dict) -> str:
"""生成签名(苏宁官方规则)"""
# 1. 参数按ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
# 3. 拼接appSecret并MD5加密(大写)
sign_str += self.app_secret
return hashlib.md5(sign_str.encode()).hexdigest().upper()
def _get_timestamp(self) -> str:
"""生成时间戳(格式:yyyyMMddHHmmss)"""
return time.strftime("%Y%m%d%H%M%S", time.localtime())
def call_api(self, method: str, biz_params: Dict) -> Optional[Dict]:
"""通用API调用方法"""
# 公共参数
public_params = {
"appKey": self.app_key,
"method": method,
"timestamp": self._get_timestamp(),
"format": "json",
"version": "v1.2"
}
# 合并参数并生成签名
all_params = {**public_params,** biz_params}
all_params["sign"] = self._generate_sign(all_params)
try:
response = self.session.get(
self.base_url,
params=all_params,
timeout=10
)
response.raise_for_status()
result = response.json()
# 处理错误响应
if result.get("code") != "0":
logging.error(f"API错误:{result.get('msg')}(错误码:{result.get('code')})")
return None
return result.get("body", {})
except RequestException as e:
logging.error(f"请求失败:{str(e)}")
return None
def get_product_detail(self, product_code: str) -> Optional[Dict]:
"""获取商品详情(含价格、优惠券)"""
method = "com.suning.govbus.product.getItem"
params = {"productCode": product_code}
result = self.call_api(method, params)
if not result or "productInfo" not in result:
return None
product = result["productInfo"]
# 解析价格和优惠
price = float(product.get("price", 0))
promotion_price = float(product.get("promotionPrice", price))
# 解析优惠券
coupons = []
for coupon in product.get("couponList", []):
coupons.append({
"id": coupon.get("couponId"),
"discount": float(coupon.get("discount", 0)), # 优惠金额
"min_amount": float(coupon.get("minAmount", 0)), # 使用门槛
"valid_time": coupon.get("validTime")
})
return {
"product_code": product_code,
"name": product.get("productName"),
"price": price,
"promotion_price": promotion_price,
"stock": product.get("stockFlag"), # 库存状态(有货/无货)
"coupons": coupons,
"shop_name": product.get("shopInfo", {}).get("shopName"),
"brand": product.get("brandName")
}
# 示例调用
if __name__ == "__main__":
# 替换为实际参数(从苏宁开放平台获取)
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
PRODUCT_CODE = "1000123456" # 商品编码(从商品URL提取)
api = SuningOfficialAPI(APP_KEY, APP_SECRET)
detail = api.get_product_detail(PRODUCT_CODE)
if detail:
print(f"商品名称:{detail['name']}")
print(f"原价:{detail['price']}元 | 促销价:{detail['promotion_price']}元")
print(f"库存:{detail['stock']} | 店铺:{detail['shop_name']}")
if detail["coupons"]:
print("可用优惠券:")
for coupon in detail["coupons"]:
print(f"- 满{coupon['min_amount']}元减{coupon['discount']}元({coupon['valid_time']})") ### 三、关键技术解析
#### 1. 官方 API 核心点
- **签名生成**:严格按照参数 ASCII 升序拼接,确保与苏宁服务器签名验证一致;
- **错误处理**:通过响应`code`字段判断请求状态(`0`为成功),常见错误如`1001`(签名错误)、`1002`(参数缺失);
- **权限控制**:部分接口(如订单操作)需单独申请权限,个人开发者需完成企业认证。
#### 2. 非官方 API 反爬应对
- **动态参数**:搜索接口的`callback`和`_`(时间戳)需动态生成,避免固定值被识别;
- **请求头轮换**:使用 User-Agent 池模拟不同设备,降低被标记为爬虫的概率;
- **数据解析**:JSONP 响应需剥离回调函数,详情页优惠券信息嵌在 JavaScript 中,需通过正则提取。
### 四、注意事项
1. **合规性**:
- 官方 API 需遵守苏宁开放平台协议,商业使用需申请授权;
- 非官方接口抓取仅限个人学习,避免高频请求或大规模爬取(违反`robots.txt`)。
1. **稳定性**:
- 官方 API 更新频率低,兼容性强;非官方接口依赖页面结构,需定期维护解析规则。
1. **扩展建议**:
- 官方 API 可扩展订单管理、库存预警等功能;
- 非官方接口可添加代理池(如`requests-proxies`)规避 IP 封禁。
两种方案各有适用场景:官方 API 适合商业应用,非官方 API 适合快速验证和个人项目。实际使用中需根据需求选择,并优先遵守平台规则。