孔夫子平台关键词已售商品接口实战：HMAC-SHA1签名+旧书品相解析+全自动分页（Python合规生产版）

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

class KongfzSoldSearchClient:
    def __init__(self, api_key, api_secret):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = "https://api.kongfz.com/v2/books/sold/search"
        self.session = requests.Session()

    def build_hmac_sign(self, params):
        """孔夫子平台官方HMAC-SHA1签名生成"""
        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 = "&".join([f"{k}={quote(str(v), safe='')}" for k, v in sorted_params])
        sign = hmac.new(self.api_secret.encode("utf-8"), sign_str.encode("utf-8"), hashlib.sha1).hexdigest()
        return sign

    def single_page_search(self, keyword, page=1, page_size=20, grade=""):
        """单页关键词已售商品搜索，支持品相筛选"""
        if len(keyword) < 1 or len(keyword) > 30:
            return {"code": -1, "msg": "关键词长度1-30字符", "goods_list": []}
        timestamp = str(int(time.time()))
        params = {
            "api_key": self.api_key,
            "keyword": keyword,
            "page": page,
            "page_size": page_size,
            "timestamp": timestamp
        }
        if grade:
            params["grade"] = grade  # 可选：九五新、八五新、七品等
        sign = self.build_hmac_sign(params)
        headers = {
            "Content-Type": "application/json;charset=utf-8",
            "Sign": sign
        }
        try:
            resp = self.session.post(self.base_url, headers=headers, json=params, timeout=15)
            raw_data = resp.json()
            # 限流自动重试
            if raw_data.get("code") == 429:
                time.sleep(2)
                return self.single_page_search(keyword, page, page_size, grade)
            if raw_data.get("code") != 200:
                return {"code": -1, "msg": raw_data.get("msg", "查询失败"), "goods_list": []}
            data = raw_data.get("data", {})
            goods_raw = data.get("list", [])
            goods_clean = []
            # 旧书数据结构化清洗
            for item in goods_raw:
                goods_clean.append({
                    "goods_id": item.get("bookId"),
                    "title": item.get("bookName"),
                    "author": item.get("author", ""),
                    "publisher": item.get("publisher", ""),
                    "publish_year": item.get("publishYear", ""),
                    "grade": item.get("grade", ""),  # 书籍品相
                    "binding": item.get("binding", ""),  # 装帧方式
                    "sold_price": float(item.get("soldPrice", 0)),
                    "publish_time": item.get("publishTime", ""),
                    "seller_credit": item.get("sellerCredit", 0),  # 卖家信用
                    "sold_time": item.get("soldTime", "")
                })
            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_sold_goods(self, keyword, grade=""):
        """全自动分页拉取全部已售商品"""
        all_goods = []
        page = 1
        while True:
            res = self.single_page_search(keyword, page, 20, grade)
            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 = KongfzSoldSearchClient(
        api_key="你的开放平台api_key",
        api_secret="你的开放平台api_secret"
    )
    # 搜索已售《红楼梦》，筛选九五新品相
    result = client.get_all_sold_goods("红楼梦", grade="九五新")
    print(json.dumps(result, ensure_ascii=False, indent=2))