苏宁开放平台商品详情接口实战：多维度数据获取与结构化处理

import hashlibimport timeimport jsonfrom urllib.parse import urlencodeclass SuningAuthUtil:    """苏宁接口签名与时间戳工具类（避坑版）"""    @staticmethod    def generate_sign(params, app_secret):        """        生成苏宁签名（关键步骤：空值过滤+ASCII排序）        :param params: 参数字典（含公共参数+业务参数）        :param app_secret: 应用密钥（开放平台获取）        :return: 签名字符串（大写）        """        try:            # 避坑1：过滤空值/空字符串参数（苏宁会因空参数导致签名失败）            valid_params = {k: v for k, v in params.items() if v is not None and v != ""}            # 避坑2：严格按参数名ASCII升序排序（不能自定义顺序）            sorted_params = sorted(valid_params.items(), key=lambda x: x[0])            # 避坑3：用urlencode拼接（自动处理特殊字符编码，如中文）            param_str = urlencode(sorted_params)            # 拼接密钥并SHA256加密            sign_str = f"{param_str}{app_secret}"            return hashlib.sha256(sign_str.encode('utf-8')).hexdigest().upper()        except Exception as e:            print(f"签名生成失败（常见原因：参数类型错误/密钥为空）：{str(e)}")            return None    @staticmethod    def get_timestamp():        """生成符合苏宁格式的时间戳（避坑：精确到秒，与服务器时间差≤5分钟）"""        return time.strftime("%Y%m%d%H%M%S")

import requestsimport timefrom threading import Lockfrom SuningAuthUtil import SuningAuthUtil  # 引入上文签名工具类class SuningProductClient:    """苏宁商品详情接口客户端（含QPS控制）"""    def __init__(self, app_key, app_secret):        self.app_key = app_key        self.app_secret = app_secret        self.base_url = "https://open.suning.com/api/mpp"        self.version = "v1.3.0"        self.timeout = 15  # 超时时间（避免卡请求）        self.qps_limit = 5  # 苏宁QPS限制        self.last_request_time = 0        self.request_lock = Lock()  # 线程锁控制并发    def _check_qps(self):        """避坑：控制QPS，避免超限被临时限制IP"""        with self.request_lock:            current_time = time.time()            min_interval = 1.0 / self.qps_limit  # 每次请求最小间隔            elapsed = current_time - self.last_request_time            if elapsed < min_interval:                time.sleep(min_interval - elapsed)  # 不足间隔则等待            self.last_request_time = time.time()    def get_single_product(self, product_code, fields=None):        """获取单个商品详情"""        self._check_qps()        # 1. 构造请求URL与参数        url = f"{self.base_url}/{self.version}/product/get"        biz_params = {"productCode": product_code}        if fields:            biz_params["fields"] = fields  # 按需筛选字段，减少数据量        # 2. 组装公共参数        common_params = {            "appKey": self.app_key,            "version": self.version,            "timestamp": SuningAuthUtil.get_timestamp(),            "format": "json",            "paramJson": json.dumps(biz_params, ensure_ascii=False)  # 业务参数转JSON        }        # 3. 生成签名        common_params["sign"] = SuningAuthUtil.generate_sign(common_params, self.app_secret)        # 4. 发送请求        try:            response = requests.post(                url,                data=common_params,                headers={"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"},                timeout=self.timeout            )            response.raise_for_status()  # 捕获4xx/5xx错误            result = response.json()            # 5. 处理响应            if result.get("code") == "0000":                return self._parse_response(result["result"])  # 结构化解析            else:                raise Exception(f"接口错误：{result.get('msg')}（错误码：{result.get('code')}）")        except Exception as e:            print(f"单商品查询失败（商品编码：{product_code}）：{str(e)}")            return None    def get_batch_products(self, product_codes, fields=None):        """批量获取商品详情（避坑：最多30个编码/次）"""        if len(product_codes) > 30:            raise ValueError("批量查询最多支持30个商品编码，需分批处理")        self._check_qps()        # 1. 构造参数（类似单商品，业务参数为productCodes）        url = f"{self.base_url}/{self.version}/product/batchGet"        biz_params = {"productCodes": ",".join(product_codes)}        if fields:            biz_params["fields"] = fields        # 2. 组装公共参数+签名（同单商品逻辑）        common_params = {            "appKey": self.app_key,            "version": self.version,            "timestamp": SuningAuthUtil.get_timestamp(),            "format": "json",            "paramJson": json.dumps(biz_params, ensure_ascii=False)        }        common_params["sign"] = SuningAuthUtil.generate_sign(common_params, self.app_secret)        # 3. 发送请求并解析        try:            response = requests.post(url, data=common_params, timeout=self.timeout)            response.raise_for_status()            result = response.json()            if result.get("code") == "0000":                product_list = result["result"].get("productList", [])                return [self._parse_response(p) for p in product_list]  # 批量解析            else:                raise Exception(f"批量查询错误：{result.get('msg')}（错误码：{result.get('code')}）")        except Exception as e:            print(f"批量查询失败（编码列表：{product_codes[:3]}...）：{str(e)}")            return None    def _parse_response(self, raw_data):        """将原始响应解析为结构化数据（方便前端/数据库使用）"""        if not raw_data:            return None        # 1. 价格信息（转float，避免字符串计算错误）        price_info = {            "original_price": float(raw_data.get("price", 0)),            "promotion_price": float(raw_data.get("promotionPrice", 0)),            "member_price": float(raw_data.get("memberPrice", 0))        }        # 2. 库存信息（结构化判断是否可购）        stock_info = {            "stock_flag": raw_data.get("stockFlag"),            "stock_desc": raw_data.get("stockDesc"),            "can_buy": raw_data.get("stockFlag") in ["1", "3"],  # 1=有货，3=预售可购            "limit_buy": int(raw_data.get("limitBuyNum", 0)) > 0        }        # 3. 媒体资源（补全图片URL协议头）        media_info = {            "main_images": [f"https:{url}" if url.startswith("//") else url                            for url in raw_data.get("picUrls", [])],            "detail_images": [f"https:{m['content']}" for m in raw_data.get("detailModule", [])                               if m.get("type") == "img"],            "video_url": raw_data.get("videoUrl")        }        # 4. 服务信息（提取关键服务名）        service_info = [s["serviceName"] for s in raw_data.get("serviceList", [])]        # 5. 整合返回        return {            "product_code": raw_data.get("productCode"),            "product_name": raw_data.get("productName"),            "brand": raw_data.get("brandName"),            "shop_name": raw_data.get("shopName"),            "price": price_info,            "stock": stock_info,            "media": media_info,            "services": service_info,            "sales": int(raw_data.get("salesVolume", 0)),            "score": float(raw_data.get("averageScore", 0)),            "update_time": raw_data.get("updateTime")        }

import osimport jsonimport sqlite3from datetime import datetime, timedeltafrom SuningProductClient import SuningProductClientclass SuningProductCache:    """苏宁商品详情缓存管理器（减少接口调用次数）"""    def __init__(self, app_key, app_secret, cache_dir="./suning_cache"):        self.client = SuningProductClient(app_key, app_secret)        self.cache_dir = cache_dir        self.db_path = os.path.join(cache_dir, "product_cache.db")        self._init_db()  # 初始化缓存数据库    def _init_db(self):        """创建缓存表（首次使用自动初始化）"""        if not os.path.exists(self.cache_dir):            os.makedirs(self.cache_dir)        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()        cursor.execute('''            CREATE TABLE IF NOT EXISTS product (                product_code TEXT PRIMARY KEY,                data TEXT,                cache_time TEXT,                expire_time TEXT            )        ''')        conn.commit()        conn.close()    def get_product(self, product_code, fields=None, cache_ttl=3600):        """获取商品（优先读缓存，过期则调用接口）"""        # 1. 尝试读缓存        cached_data = self._get_cached(product_code, cache_ttl)        if cached_data:            return cached_data        # 2. 缓存过期，调用接口        fresh_data = self.client.get_single_product(product_code, fields)        if fresh_data:            self._save_cache(product_code, fresh_data, cache_ttl)  # 保存新缓存        return fresh_data    def _get_cached(self, product_code, cache_ttl):        """从缓存获取数据（判断是否过期）"""        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()        cursor.execute(            "SELECT data, cache_time FROM product WHERE product_code = ?",            (product_code,)        )        result = cursor.fetchone()        conn.close()        if not result:            return None        # 判断缓存是否过期        data_str, cache_time = result        cache_time_obj = datetime.strptime(cache_time, "%Y-%m-%d %H:%M:%S")        if (datetime.now() - cache_time_obj).total_seconds() > cache_ttl:            return None  # 过期返回空        return json.loads(data_str)    def _save_cache(self, product_code, data, cache_ttl):        """保存数据到缓存"""        data_str = json.dumps(data, ensure_ascii=False)        cache_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")        expire_time = (datetime.now() + timedelta(seconds=cache_ttl)).strftime("%Y-%m-%d %H:%M:%S")        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()        # 插入或更新缓存（避免重复数据）        cursor.execute('''            INSERT OR REPLACE INTO product             (product_code, data, cache_time, expire_time)            VALUES (?, ?, ?, ?)        ''', (product_code, data_str, cache_time, expire_time))        conn.commit()        conn.close()    def clean_expired_cache(self, max_age=86400):        """清理过期缓存（默认保留24小时内数据）"""        expire_time = (datetime.now() - timedelta(seconds=max_age)).strftime("%Y-%m-%d %H:%M:%S")        conn = sqlite3.connect(self.db_path)        cursor = conn.cursor()        cursor.execute("DELETE FROM product WHERE cache_time < ?", (expire_time,))        deleted_count = cursor.rowcount        conn.commit()        conn.close()        print(f"清理过期缓存：共删除{deleted_count}条记录")        return deleted_count

def single_product_demo():    # 替换为你的苏宁开放平台应用信息    APP_KEY = "your_app_key"    APP_SECRET = "your_app_secret"    # 初始化缓存管理器（兼顾效率与实时性）    cache_manager = SuningProductCache(APP_KEY, APP_SECRET)    # 要查询的商品编码（从苏宁商品页URL提取，如https://product.suning.com/0000000000/1000123456.html中的1000123456）    product_code = "1000123456"    # 按需筛选字段（只获取需要的，减少传输量）    fields = "productCode,productName,price,promotionPrice,stockFlag,stockDesc,picUrls,detailModule,serviceList"        # 获取商品详情（缓存1小时）    product = cache_manager.get_product(product_code, fields=fields, cache_ttl=3600)    if product:        print(f"===== 商品详情：{product['product_name']} =====")        print(f"商品编码：{product['product_code']}")        print(f"品牌：{product['brand']}")        print(f"价格：原价¥{product['price']['original_price']} | 促销价¥{product['price']['promotion_price']}")        print(f"库存：{product['stock']['stock_desc']}（可购：{'是' if product['stock']['can_buy'] else '否'}）")        print(f"服务保障：{'; '.join(product['services'])}")        print(f"主图数量：{len(product['media']['main_images'])} | 详情图数量：{len(product['media']['detail_images'])}")        # 清理24小时前的过期缓存    cache_manager.clean_expired_cache()if __name__ == "__main__":    single_product_demo()

def batch_product_compare():    APP_KEY = "your_app_key"    APP_SECRET = "your_app_secret"    client = SuningProductClient(APP_KEY, APP_SECRET)    # 要对比的商品编码列表（不超过30个）    product_codes = ["1000123456", "1000123457", "1000123458", "1000123459"]        # 批量获取商品详情    products = client.get_batch_products(product_codes)    if not products:        print("批量查询失败")        return        # 对比核心维度（价格、销量、服务）    print("===== 商品批量对比结果 =====")    for idx, p in enumerate(products, 1):        if not p:            continue        print(f"\n{idx}. 商品：{p['product_name']}（编码：{p['product_code']}）")        print(f"   价格：¥{p['price']['promotion_price']}（原价¥{p['price']['original_price']}）")        print(f"   销量：30天{p['sales']}件 | 评分：{p['score']}分")        print(f"   核心服务：{'; '.join(p['services'][:3])}")  # 只显示前3个服务if __name__ == "__main__":    batch_product_compare()