×

深度分析苏宁API接口,用Python脚本实现

万邦科技Lex 万邦科技Lex 发表于2025-08-21 13:37:27 浏览19 评论0

抢沙发发表评论


 免费测试苏宁API接口

苏宁易购作为国内领先的综合电商平台,其 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 适合快速验证和个人项目。实际使用中需根据需求选择,并优先遵守平台规则。

群贤毕至

访客