当当接口开发避坑指南：3 大痛点 + 签名模板，0 失败接入商品详情接口

import requests
import hashlib
import time
import secrets
import logging
# 配置日志（便于排查问题）
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger('dangdang-api')
class DangDangProductClient:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = "https://api.open.dangdang.com/product/detail"
        self.timeout = 8  # 电商接口建议超时设5-10秒
    def _generate_sign(self, params: dict) -> str:
        """生成当当接口签名（核心步骤）"""
        # 1. 移除空值参数，按ASCII升序排序
        sorted_params = sorted([(k, v) for k, v in params.items() if v is not None], key=lambda x: x[0])
        # 2. 拼接参数字符串
        param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
        # 3. 尾部拼接app_secret并加密
        sign_str = f"{param_str}{self.client_secret}"
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def get_product_detail(self, isbn: str = None, product_id: str = None, need_stock: bool = False, detail_level: int = 1) -> dict:
        """
        获取商品详情
        :param isbn: 国际标准书号（13位/10位）
        :param product_id: 平台商品编号（与isbn二选一）
        :param need_stock: 是否返回库存数据
        :param detail_level: 详情级别（1-基础/2-扩展/3-完整）
        """
        # 参数合法性校验
        if not (isbn or product_id):
            raise ValueError("isbn与product_id必须传入一个")
        if detail_level not in [1, 2, 3]:
            raise ValueError("detail_level只能为1、2、3")
        # 构建基础参数
        params = {
            "timestamp": str(int(time.time())),
            "nonce": secrets.token_hex(3),  # 6位随机字符串
            "detail_level": str(detail_level),
            "need_stock": str(need_stock).lower()  # 需转为小写布尔值
        }
        # 优先级处理：isbn优先于product_id
        if isbn:
            params["isbn"] = isbn
        else:
            params["product_id"] = product_id
        # 生成签名
        params["sign"] = self._generate_sign(params)
        # 发送请求
        headers = {"X-Client-Id": self.client_id}
        try:
            response = requests.get(self.base_url, params=params, headers=headers, timeout=self.timeout)
            response.raise_for_status()  # 触发HTTP错误
            result = response.json()
            # 处理纸书与电子书数据差异
            data = result.get("data", {})
            if data.get("ebook"):
                logger.info(f"获取电子书详情成功，ISBN：{isbn}")
                return self._process_ebook_data(data)
            else:
                logger.info(f"获取纸书详情成功，ISBN：{isbn}")
                return self._process_paperbook_data(data)
        except requests.exceptions.HTTPError as e:
            logger.error(f"HTTP错误：{e.response.status_code}，响应：{e.response.text}")
            raise
        except Exception as e:
            logger.error(f"接口调用失败：{str(e)}")
            raise
    def _process_paperbook_data(self, data: dict) -> dict:
        """处理纸书商品数据"""
        return {
            "商品编号": data["base_info"]["product_id"],
            "ISBN": data["base_info"]["isbn"],
            "书名": data["base_info"]["title"],
            "作者": data["base_info"]["author"],
            "出版社": data["base_info"]["press"],
            "定价": data["price_info"]["retail_price"],
            "当当价": data["price_info"]["dangdang_price"],
            "库存状态": self._parse_stock_status(data["price_info"]["stock_status"]),
            "封面图": data["media_info"]["cover_image"]
        }
    def _process_ebook_data(self, data: dict) -> dict:
        """处理电子书商品数据"""
        return {
            "商品编号": data["base_info"]["product_id"],
            "ISBN": data["base_info"]["isbn"],
            "书名": data["base_info"]["title"],
            "作者": data["base_info"]["author"],
            "定价": data["price_info"]["retail_price"],
            "当当价": data["price_info"]["dangdang_price"],
            "音频预览": data["media_info"].get("audio_preview", "无")
        }
    @staticmethod
    def _parse_stock_status(status_code: int) -> str:
        """解析库存状态码"""
        status_map = {1: "充足", 2: "紧张", 3: "预售"}
        return status_map.get(status_code, "未知")
# 示例调用
if __name__ == "__main__":
    # 替换为自己的ClientId和ClientSecret
    client = DangDangProductClient(
        client_id="your_client_id",
        client_secret="your_client_secret"
    )
    # 获取纸书详情（以《中国历代政治得失》为例）
    book_detail = client.get_product_detail(isbn="9787108009821", need_stock=True, detail_level=2)
    print(book_detail)