京东商品搜索列表API开发实录：Python 签名、分页与异常处理

import hashlib
import requests
import time
from typing import Optional, Dict, List

class JdUnionSearchApi:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.gateway = "https://api.jd.com/routerjson"

    @staticmethod
    def get_sign(params: Dict[str, str], secret: str) -> str:
        # 按key升序排序
        sorted_pairs = sorted(params.items(), key=lambda x: x[0])
        raw_str = "".join([f"{k}{v}" for k, v in sorted_pairs]) + secret
        md5_val = hashlib.md5(raw_str.encode("utf-8")).hexdigest()
        return md5_val.upper()

    def search_goods(self,
                     keyword: str,
                     page_index: int = 1,
                     page_size: int = 20,
                     sort: Optional[str] = None,
                     cid: Optional[str] = None,
                     price_start: Optional[int] = None,
                     price_end: Optional[int] = None
                     ) -> dict:
        """单次搜索一页商品"""
        params = {
            "method": "jd.union.open.goods.search",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "keyword": keyword,
            "pageIndex": str(page_index),
            "pageSize": str(page_size)
        }
        # 可选参数，不为空才加入
        if sort:
            params["sort"] = sort
        if cid:
            params["cid"] = cid
        if price_start is not None:
            params["priceStart"] = str(price_start)
        if price_end is not None:
            params["priceEnd"] = str(price_end)

        params["sign"] = self.get_sign(params, self.app_secret)
        try:
            resp = requests.get(self.gateway, params=params, timeout=10)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.Timeout:
            return {"code": -1, "msg": "请求超时"}
        except requests.exceptions.HTTPError as e:
            return {"code": -2, "msg": f"HTTP异常：{str(e)}"}
        except Exception as e:
            return {"code": -99, "msg": f"未知异常：{str(e)}"}

    def fetch_all_pages(self, keyword: str, max_page: int = 5, sleep_sec: float = 1.2, **kwargs) -> List[dict]:
        """循环分页拉取，内置限流休眠，防止被封"""
        all_goods = []
        for page in range(1, max_page + 1):
            res = self.search_goods(keyword=keyword, page_index=page, **kwargs)
            if res.get("code", 0) != 0:
                print(f"第{page}页请求失败：{res.get('msg')}")
                break
            # 根据京东返回结构提取商品数组，这里根据实际返回调整key
            data = res.get("jd_union_open_goods_search_response", {})
            result = data.get("result", {})
            goods_list = result.get("goodsList", [])
            if not goods_list:
                print(f"第{page}页无商品，停止翻页")
                break
            all_goods.extend(goods_list)
            print(f"成功获取第{page}页，本页{len(goods_list)}条")
            time.sleep(sleep_sec)
        return all_goods


if __name__ == "__main__":
    # 替换为自己的密钥
    ak = "YOUR_APP_KEY"
    sk = "YOUR_APP_SECRET"
    api = JdUnionSearchApi(ak, sk)
    goods = api.fetch_all_pages(keyword="无线充电器", max_page=3, page_size=20)
    print(f"一共拿到商品：{len(goods)} 条")