震坤行商品详情接口深度解析：工业级数据对接与异常处理实战

import requests
import time
from datetime import datetime

class ZkhAuth:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = "https://openapi.zkh360.com/oauth/token"
        self.access_token = None
        self.expires_at = 0

    def get_token(self):
        # 令牌未过期直接返回
        if self.access_token and time.time() < self.expires_at - 60:
            return self.access_token
        
        # 重新获取令牌
        params = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        try:
            resp = requests.post(self.token_url, data=params, timeout=10)
            resp.raise_for_status()
            data = resp.json()
            self.access_token = data["access_token"]
            self.expires_at = time.time() + data["expires_in"]
            return self.access_token
        except Exception as e:
            raise Exception(f"认证失败：{str(e)}")

class ZkhProductApi:
    def __init__(self, auth):
        self.auth = auth
        self.base_url = "https://openapi.zkh360.com/api/v2/products"

    def get_product_detail(self, sku, retry=2):
        """
        获取商品详情
        :param sku: 商品SKU（如"AA6375379"）
        :param retry: 重试次数
        :return: 结构化商品数据
        """
        token = self.auth.get_token()
        url = f"{self.base_url}/{sku}"
        headers = {"Authorization": f"Bearer {token}"}

        for attempt in range(retry + 1):
            try:
                resp = requests.get(url, headers=headers, timeout=15)
                resp.raise_for_status()
                result = resp.json()

                # 业务状态校验
                if result.get("code") != 0:
                    raise Exception(f"业务异常：{result.get('message')}")
                
                return self._parse_detail(result["data"])
            
            except Exception as e:
                if attempt < retry:
                    time.sleep(2 ** attempt)  # 指数退避重试
                    continue
                raise Exception(f"获取详情失败（重试{retry}次）：{str(e)}")

    def _parse_detail(self, raw_data):
        """结构化解析核心字段（过滤冗余数据）"""
        return {
            "sku": raw_data["sku"],
            "name": raw_data["productName"],
            "brand": raw_data["brandName"],
            "price": {
                "sale_price": raw_data["salePrice"],
                "market_price": raw_data["marketPrice"]
            },
            "stock": raw_data["stockNum"],
            "spec": raw_data["specification"],
            "images": [img["url"] for img in raw_data["imageList"]],
            "category": raw_data["categoryName"],
            "update_time": datetime.fromtimestamp(raw_data["updateTime"]/1000).strftime("%Y-%m-%d %H:%M:%S")
        }

# 使用示例
if __name__ == "__main__":
    # 替换为你的client_id和client_secret
    auth = ZkhAuth("YOUR_CLIENT_ID", "YOUR_CLIENT_SECRET")
    product_api = ZkhProductApi(auth)
    
    try:
        # 查询指定SKU商品
        product = product_api.get_product_detail("AA6375379")
        print("商品详情：")
        for k, v in product.items():
            print(f"{k}: {v}")
    except Exception as e:
        print(f"错误：{e}")