苏宁易购（综合家电，综合，聚焦家电、3C 等品类的综合电商平台）的商品详情数据（如实时价格、库存状态、促销活动、售后政策等）对价格监控、竞品分析、供应链管理等场景具有重要价值。由于平台无公开官方 API，开发者需通过页面解析或逆向工程实现商品详情（item_get）的获取。本文系统讲解接口逻辑、技术实现、家电 3C 场景适配及反爬应对，帮助构建稳定的苏宁商品详情获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（环境与 URL 结构）
三、接口调用流程（基于页面解析与动态接口）
四、代码实现示例（Python）
import requests
import time
import random
import re
import json
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import Dict, List
class SuningItemApi:
def __init__(self, proxy_pool: List[str] = None, cookie: str = "", city_id: str = "110100"):
self.base_url = "https://item.suning.com/{product_id}.html"
self.price_stock_api = "https://pas.suning.com/nspcsale_0_0000000{product_id}_0000000{product_id}_110_0000000000_1100259_12506_12506_0000000.html"
self.param_api = "https://product.suning.com/emall/wcsnum/findParameterNew.do"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表
self.cookie = cookie # 登录态Cookie，需包含cityId
self.city_id = city_id # 城市ID（默认北京110100）
def _get_headers(self) -> Dict[str, str]:
"""生成随机请求头，包含城市定位"""
headers = {
"User-Agent": self.ua.random,
"Referer": "https://www.suning.com/",
"Accept": "application/json, text/plain, */*",
"X-Requested-With": "XMLHttpRequest"
}
if self.cookie:
# 确保Cookie包含cityId（覆盖默认城市）
if "cityId=" not in self.cookie:
self.cookie += f"; cityId={self.city_id}"
headers["Cookie"] = self.cookie
return headers
def _get_proxy(self) -> Dict[str, str]:
"""随机获取代理"""
if self.proxy_pool and len(self.proxy_pool) > 0:
proxy = random.choice(self.proxy_pool)
return {"http": proxy, "https": proxy}
return None
def _pad_product_id(self, product_id: str) -> str:
"""补全商品ID至12位（苏宁接口要求）"""
return product_id.zfill(12)
def _clean_price(self, price_str: str) -> float:
"""清洗价格（去除¥、逗号等）"""
if not price_str:
return 0.0
price_str = re.sub(r"[^\d.]", "", price_str)
return float(price_str) if price_str else 0.0
def _parse_static_data(self, html: str) -> Dict:
"""解析主页面静态数据"""
soup = BeautifulSoup(html, "lxml")
# 提取规格选项（型号/颜色）
specs = []
for attr_item in soup.select("div.attr-item"):
spec_name = attr_item.select_one(".attr-title")?.text.strip() or "规格"
spec_values = [
{
"name": option.text.strip(),
"spec_id": option.get("data-spec") or "", # 规格ID（多规格时使用）
"selected": "selected" in option.get("class", [])
}
for option in attr_item.select(".attr-value a")
]
if spec_values:
specs.append({
"name": spec_name,
"values": spec_values
})
return {
"title": soup.select_one("h1.pro-name")?.text.strip() or "",
"images": {
"main": [img.get("src") for img in soup.select("div.img-list img") if img.get("src")],
"detail": [img.get("src") for img in soup.select("div.detail-img img") if img.get("src")]
},
"price": {
"sale_str": soup.select_one("div.mainprice .def-price")?.text.strip() or "",
"member_str": soup.select_one("div.member-price")?.text.strip() or ""
},
"brand": {
"name": soup.select_one("div.brand-name a")?.text.strip() or ""
},
"promotion": {
"tags": [tag.text.strip() for tag in soup.select("div.prom-tag")],
"gifts": [gift.text.strip() for gift in soup.select("div.gift-list span")]
},
"service": {
"policy": [s.text.strip() for s in soup.select("div.service-tags span")],
"price_protect": "7天价保" in soup.select_one("div.service-tags")?.text or False
},
"specs": specs # 规格选项（如内存、颜色）
}
def _fetch_price_stock(self, product_id: str, spec_id: str = "", headers: Dict[str, str], proxy: Dict[str, str]) -> Dict:
"""调用价格库存接口（支持多规格）"""
api_data = {"price": {}, "stock": {}, "delivery": {}}
try:
# 补全商品ID至12位
padded_id = self._pad_product_id(product_id)
# 多规格时使用spec_id，否则用product_id
target_id = spec_id if spec_id else padded_id
url = self.price_stock_api.format(product_id=target_id)
response = requests.get(
url,
headers=headers,
proxies=proxy,
timeout=10
)
# 苏宁接口响应为JSONP格式，需提取JSON部分
json_str = re.search(r"({.*})", response.text).group(1)
data = json.loads(json_str)
# 解析价格
price_info = data.get("price", {})
api_data["price"] = {
"netPrice": price_info.get("netPrice", 0),
"memberPrice": price_info.get("memberPrice", 0),
"promotionPrice": price_info.get("promotionPrice", 0)
}
# 解析库存
stock_info = data.get("stock", {})
api_data["stock"] = {
"stockDesc": stock_info.get("stockDesc", ""),
"canSaleFlag": stock_info.get("canSaleFlag", False),
"limitNum": stock_info.get("limitNum", 0)
}
# 解析配送
delivery_info = data.get("delivery", {})
api_data["delivery"] = {
"timeDesc": delivery_info.get("timeDesc", "")
}
except Exception as e:
print(f"价格库存接口失败: {str(e)}")
return api_data
def _fetch_parameters(self, product_id: str, headers: Dict[str, str], proxy: Dict[str, str]) -> Dict:
"""调用参数接口获取核心参数"""
params = {"partNumber": product_id}
try:
response = requests.get(
self.param_api,
params=params,
headers=headers,
proxies=proxy,
timeout=10
)
data = response.json()
# 提取核心参数（如手机的CPU、内存）
parameters = {}
for group in data.get("itemParamList", []):
for param in group.get("paramList", []):
parameters[param.get("paramName")] = param.get("paramValue")
return {"parameters": parameters}
except Exception as e:
print(f"参数接口失败: {str(e)}")
return {"parameters": {}}
def _merge_multi_specs(self, static_specs: List[Dict], product_id: str, headers: Dict[str, str], proxy: Dict[str, str]) -> List[Dict]:
"""合并多规格商品的价格和库存"""
merged_specs = []
for spec_group in static_specs:
spec_name = spec_group["name"]
merged_values = []
for spec in spec_group["values"]:
spec_id = spec["spec_id"]
# 调用规格对应的价格库存接口
spec_data = self._fetch_price_stock(product_id, spec_id, headers, proxy)
merged_values.append({
**spec,** spec_data["price"],
"stock": spec_data["stock"],
"delivery": spec_data["delivery"]
})
merged_specs.append({
"name": spec_name,
"values": merged_values
})
return merged_specs
def item_get(self, product_id: str, timeout: int = 10) -> Dict:
"""
获取苏宁商品详情
:param product_id: 商品ID（如0071570066）
:param timeout: 超时时间
:return: 标准化商品数据
"""
try:
# 1. 主页面请求
url = self.base_url.format(product_id=product_id)
headers = self._get_headers()
proxy = self._get_proxy()
# 随机延迟，避免反爬
time.sleep(random.uniform(2, 4))
response = requests.get(
url=url,
headers=headers,
proxies=proxy,
timeout=timeout
)
response.raise_for_status()
main_html = response.text
# 2. 解析主页面数据
static_data = self._parse_static_data(main_html)
if not static_data["title"]:
return {"success": False, "error_msg": "商品不存在或已下架"}
# 3. 获取价格库存数据（默认规格）
price_stock_data = self._fetch_price_stock(product_id, "", headers, proxy)
# 4. 获取核心参数
param_data = self._fetch_parameters(product_id, headers, proxy)
# 5. 处理多规格商品（若有）
merged_specs = static_data["specs"]
if static_data["specs"] and any(len(s["values"]) > 1 for s in static_data["specs"]):
merged_specs = self._merge_multi_specs(static_data["specs"], product_id, headers, proxy)
# 6. 整合结果
result = {
"success": True,
"data": {
"product_id": product_id,** static_data,
"price": {
**static_data["price"],** price_stock_data["price"]
},
"stock": price_stock_data["stock"],
"delivery": price_stock_data["delivery"],
"parameters": param_data["parameters"],
"specs": merged_specs, # 多规格已合并价格库存
"url": url,
"update_time": time.strftime("%Y-%m-%d %H:%M:%S"),
"city_id": self.city_id # 数据对应的城市
}
}
return result
except requests.exceptions.HTTPError as e:
if "403" in str(e):
return {"success": False, "error_msg": "触发反爬，建议更换代理或Cookie", "code": 403}
if "401" in str(e):
return {"success": False, "error_msg": "Cookie无效，请重新登录获取", "code": 401}
return {"success": False, "error_msg": f"HTTP错误: {str(e)}", "code": response.status_code}
except Exception as e:
return {"success": False, "error_msg": f"获取失败: {str(e)}", "code": -1}
# 使用示例
if __name__ == "__main__":
# 代理池（替换为有效代理）
PROXIES = [
"http://123.45.67.89:8888",
"http://98.76.54.32:8080"
]
# 登录态Cookie（从浏览器获取，需包含cityId）
COOKIE = "userid=xxx; sessionId=xxx; cityId=110100; areaId=110101" # 北京
# 初始化API客户端（指定城市为北京）
api = SuningItemApi(proxy_pool=PROXIES, cookie=COOKIE, city_id="110100")
# 获取商品详情（示例product_id）
product_id = "0071570066" # 小米14手机
result = api.item_get(product_id)
if result["success"]:
data = result["data"]
print(f"商品标题: {data['title']}")
print(f"价格信息: 售价¥{data['price']['netPrice']} | 会员价¥{data['price']['memberPrice']} | 促销价¥{data['price']['promotionPrice']}")
print(f"库存配送: {data['stock']['stockDesc']} | {data['delivery']['timeDesc']}")
print(f"促销活动: {', '.join(data['promotion']['tags'])}")
if data['promotion']['gifts']:
print(f"赠品: {', '.join(data['promotion']['gifts'])}")
print(f"售后服务: {', '.join(data['service']['policy'])} | {'支持价保' if data['service']['price_protect'] else '不支持价保'}")
print(f"核心参数: CPU={data['parameters'].get('CPU型号')} | 内存={data['parameters'].get('运行内存')} | 存储={data['parameters'].get('机身存储')}")
if data['specs']:
print(f"规格选项:")
for spec_group in data['specs'][:1]: # 第一个规格组（如内存）
print(f" {spec_group['name']}:")
for spec in spec_group['values'][:2]:
print(f" {spec['name']}: 价格¥{spec['netPrice']} | {spec['stock']['stockDesc']}")
print(f"详情页: {data['url']}")
else:
print(f"获取失败: {result['error_msg']}（错误码: {result.get('code')}）")