当当开放平台关键词商品搜索接口实战：专属MD5签名+图书字段解析+风控分页查询（Python合规生产版）

import requests
import hashlib
import time
import json
from urllib.parse import quote

class DangdangSearchClient:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.dangdang.com/v2/searchItems.php"
        self.version = "v2.0"
        self.session = requests.Session()

    def build_dangdang_sign(self, params):
        """当当开放平台专属MD5签名生成"""
        valid_params = {k: v for k, v in params.items() if v}
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{quote(str(v), safe='')}"
        sign_str += self.app_secret
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    def single_page_search(self, keyword, page=1, page_size=20, sort="sort_default"):
        """单页关键词商品搜索，支持排序筛选"""
        if len(keyword) < 1 or len(keyword) > 30:
            return {"code": -1, "msg": "关键词长度1-30字符", "goods_list": []}
        timestamp = str(int(time.time()))
        params = {
            "app_key": self.app_key,
            "version": self.version,
            "keyword": keyword,
            "page": page,
            "page_size": page_size,
            "sort": sort,
            "timestamp": timestamp
        }
        params["sign"] = self.build_dangdang_sign(params)
        headers = {"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"}
        try:
            resp = self.session.post(self.base_url, data=params, headers=headers, timeout=15)
            raw_data = resp.json()
            # 限流自动重试
            if raw_data.get("code") == 429:
                time.sleep(2)
                return self.single_page_search(keyword, page, page_size, sort)
            if raw_data.get("code") != 0:
                return {"code": -1, "msg": raw_data.get("msg", "查询失败"), "goods_list": []}
            data = raw_data.get("data", {})
            goods_raw = data.get("items", [])
            goods_clean = []
            # 图书数据结构化清洗
            for item in goods_raw:
                goods_clean.append({
                    "goods_id": item.get("itemId"),
                    "title": item.get("itemName"),
                    "author": item.get("author", ""),
                    "publisher": item.get("publisher", ""),
                    "isbn": item.get("isbn", ""),
                    "publish_time": item.get("publishTime", ""),
                    "format": item.get("format", ""),  # 图书开本
                    "price": float(item.get("price", 0)),
                    "market_price": float(item.get("marketPrice", 0)),
                    "stock": item.get("stock", 0),
                    "sales": item.get("sales", 0)
                })
            time.sleep(1)
            return {
                "code": 200,
                "total": data.get("total", 0),
                "total_page": data.get("totalPage", 0),
                "current_page": page,
                "goods_list": goods_clean
            }
        except Exception as e:
            return {"code": -2, "msg": f"网络异常：{str(e)}", "goods_list": []}

    def get_all_goods(self, keyword, sort="sort_default"):
        """全自动分页拉取全部商品"""
        all_goods = []
        page = 1
        while True:
            res = self.single_page_search(keyword, page, 20, sort)
            if res["code"] != 200 or not res["goods_list"]:
                break
            all_goods.extend(res["goods_list"])
            if page >= res["total_page"]:
                break
            page += 1
        return {"keyword": keyword, "total": len(all_goods), "data": all_goods}

# 调用示例
if __name__ == "__main__":
    client = DangdangSearchClient(
        app_key="你的开放平台app_key",
        app_secret="你的开放平台app_secret"
    )
    # 搜索Python相关图书，按销量排序
    result = client.get_all_goods("Python编程", sort="sort_sales")
    print(json.dumps(result, ensure_ascii=False, indent=2))