https://zhuanlan.zhihu.com/p/2023449548563948435

import hashlibimport urllib.parsedef generate_sign(params: dict, app_secret: str) -> str:
    """
    生成京东API签名
    :param params: 请求参数字典（不含sign）
    :param app_secret: 应用密钥
    :return: 大写MD5签名
    """
    # 1. 过滤空值并排序
    sorted_params = sorted(
        [(k, v) for k, v in params.items() if v is not None and v != ""],
        key=lambda x: x[0]
    )
    
    # 2. 拼接字符串（注意：值需要URL编码）
    sign_str = app_secret    for key, value in sorted_params:
        # 对值进行URL编码，确保特殊字符正确处理
        encoded_value = urllib.parse.quote(str(value), safe='')
        sign_str += f"{key}{encoded_value}"
    sign_str += app_secret    
    # 3. MD5加密并转大写
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    return sign# 使用示例params = {
    "method": "jd.union.open.goods.search",
    "app_key": "your_app_key",
    "access_token": "your_access_token",
    "timestamp": "2025-04-03 17:30:00",
    "v": "1.0",
    "format": "json",
    "360buy_param_json": '{"keyword":"手机","pageIndex":1}'  # 业务参数需JSON序列化}sign = generate_sign(params, "your_app_secret")print(f"生成的签名: {sign}")

import requestsimport timeimport jsonimport hashlibimport urllib.parseimport loggingfrom typing import Dict, List, Optional, Tuplefrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# 配置日志logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)class JDUnionSearchAPI:
    """
    京东联盟商品搜索API客户端
    """
    
    def __init__(self, app_key: str, app_secret: str, access_token: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.api_url = "https://api.jd.com/routerjson"
        
        # 配置会话和重试机制（应对限流和网络波动）
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503]
        )
        self.session.mount('https://', HTTPAdapter(max_retries=retry))
        
        # 核心返回字段配置（按需调整）
        self.core_fields = (
            "skuId,spuId,productName,price,lowestPrice,"
            "commissionRate,commission,couponInfo,"
            "goodComments,totalComments,score,"
            "shopId,shopName,isSelf,volume,"
            "imgUrl,materialUrl,hotScore"
        )
        
        # 排序方式映射
        self.sort_mapping = {
            "price_asc": ("price", "asc"),
            "price_desc": ("price", "desc"),
            "sales_desc": ("volume", "desc"),
            "commission_desc": ("commissionRate", "desc"),
            "rating_desc": ("score", "desc")
        }
    
    def _generate_sign(self, params: dict) -> str:
        """生成请求签名"""
        sorted_params = sorted(
            [(k, v) for k, v in params.items() if v is not None],
            key=lambda x: x[0]
        )
        
        sign_str = self.app_secret        for key, value in sorted_params:
            encoded_value = urllib.parse.quote(str(value), safe='')
            sign_str += f"{key}{encoded_value}"
        sign_str += self.app_secret        
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def search_goods(
        self, 
        keyword: str,
        page: int = 1,
        page_size: int = 20,
        price_range: Optional[Tuple[float, float]] = None,
        sort: Optional[str] = "sales_desc",
        has_coupon: bool = False,
        is_self: Optional[bool] = None,
        min_good_rate: Optional[int] = None
    ) -> Dict:
        """
        搜索商品
        
        :param keyword: 搜索关键词（必填）
        :param page: 页码，默认1
        :param page_size: 每页数量（1-50）
        :param price_range: 价格范围元组 (min, max)
        :param sort: 排序方式（price_asc/price_desc/sales_desc/commission_desc）
        :param has_coupon: 是否只返回有优惠券的商品
        :param is_self: 是否只返回自营商品（None表示不筛选）
        :param min_good_rate: 最低好评率（0-100）
        :return: 包含商品列表和总数的字典
        """
        
        # 构建业务参数（需JSON序列化）
        search_params = {
            "keyword": keyword,
            "pageIndex": page,
            "pageSize": min(max(page_size, 1), 50),  # 限制1-50
            "fields": self.core_fields        }
        
        # 价格范围筛选
        if price_range and len(price_range) == 2:
            search_params["priceFrom"] = price_range[0]
            search_params["priceTo"] = price_range[1]
        
        # 排序设置
        if sort in self.sort_mapping:
            search_params["sortName"], search_params["sort"] = self.sort_mapping[sort]
        
        # 优惠券筛选
        if has_coupon:
            search_params["hasCoupon"] = 1
        
        # 自营筛选
        if is_self is not None:
            search_params["isSelf"] = 1 if is_self else 0
        
        # 好评率筛选
        if min_good_rate and 0 <= min_good_rate <= 100:
            search_params["goodRate"] = min_good_rate        
        # 构建完整请求参数
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        params = {
            "method": "jd.union.open.goods.search",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": timestamp,
            "format": "json",
            "v": "1.0",
            "sign_method": "md5",
            "360buy_param_json": json.dumps(search_params, ensure_ascii=False)
        }
        
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            # 发送请求
            response = self.session.get(
                self.api_url,
                params=params,
                timeout=(10, 30)
            )
            response.raise_for_status()
            result = response.json()
            
            # 处理错误响应
            if "error_response" in result:
                error = result["error_response"]
                logger.error(f"接口错误: {error.get('msg')} (代码: {error.get('code')})")
                return {"success": False, "error": error.get('msg')}
            
            # 解析成功响应
            data = result.get("jd_union_open_goods_search_response", {})
            search_result = data.get("result", {})
            goods_list = search_result.get("data", [])
            total_count = search_result.get("totalCount", 0)
            
            # 数据清洗和格式化
            cleaned_items = self._clean_goods_data(goods_list)
            
            # 控制请求频率（避免触发QPS限制）
            time.sleep(1.5)
            
            return {
                "success": True,
                "total": total_count,
                "page": page,
                "items": cleaned_items            }
            
        except requests.exceptions.RequestException as e:
            logger.error(f"网络请求异常: {str(e)}")
            return {"success": False, "error": f"网络错误: {str(e)}"}
        except json.JSONDecodeError:
            logger.error("JSON解析失败")
            return {"success": False, "error": "响应数据解析失败"}
        except Exception as e:
            logger.error(f"未知异常: {str(e)}")
            return {"success": False, "error": str(e)}
    
    def _clean_goods_data(self, goods_list: List[Dict]) -> List[Dict]:
        """清洗和格式化商品数据"""
        cleaned = []
        for item in goods_list:
            try:
                cleaned_item = {
                    "sku_id": item.get("skuId"),
                    "spu_id": item.get("spuId"),
                    "title": item.get("productName", ""),
                    "price": float(item.get("price", 0)),
                    "lowest_price": float(item.get("lowestPrice", 0)),
                    "commission_rate": f"{item.get('commissionRate', 0)}%",
                    "commission_amount": item.get("commission"),
                    "has_coupon": bool(item.get("couponInfo")),
                    "coupon_info": item.get("couponInfo"),
                    "shop_name": item.get("shopName"),
                    "is_self": item.get("isSelf") == 1,
                    "sales_volume": item.get("volume"),
                    "good_comments": item.get("goodComments"),
                    "total_comments": item.get("totalComments"),
                    "good_rate": item.get("score"),
                    "image_url": item.get("imgUrl"),
                    "product_url": f"https://item.jd.com/{item.get('skuId')}.html",
                    "material_url": item.get("materialUrl")  # 推广链接
                }
                cleaned.append(cleaned_item)
            except Exception as e:
                logger.warning(f"数据清洗异常: {e}")
                continue
        return cleaned# ==================== 使用示例 ====================if __name__ == "__main__":
    # 初始化API客户端（替换为你的真实凭证）
    jd_api = JDUnionSearchAPI(
        app_key="your_app_key_here",
        app_secret="your_app_secret_here",
        access_token="your_access_token_here"
    )
    
    # 场景1：搜索"保温杯"，价格10-50元，有优惠券，按销量排序
    result = jd_api.search_goods(
        keyword="保温杯",
        page=1,
        page_size=20,
        price_range=(10, 50),
        has_coupon=True,
        sort="sales_desc"
    )
    
    if result["success"]:
        print(f"✅ 搜索成功，共找到 {result['total']} 件商品")
        for item in result["items"][:5]:  # 只展示前5条
            print(f"\n【{item['title'][:20]}...】")
            print(f"   价格: ¥{item['lowest_price']} (原价¥{item['price']})")
            print(f"   佣金: {item['commission_rate']}")
            print(f"   销量: {item['sales_volume']} | 好评率: {item['good_rate']}")
            print(f"   链接: {item['product_url']}")
    else:
        print(f"❌ 搜索失败: {result['error']}")

{
  "jd_union_open_goods_search_response": {
    "result": {
      "totalCount": 1580,
      "data": [
        {
          "skuId": "100012043978",
          "productName": "富光保温杯 304不锈钢 500ml",
          "price": 59.90,
          "lowestPrice": 39.90,
          "commissionRate": 15.00,
          "commission": 5.99,
          "couponInfo": {
            "couponAmount": 10,
            "couponCondition": "满30可用"
          },
          "volume": 50000,
          "isSelf": 1,
          "shopName": "富光京东自营旗舰店",
          "score": 98,
          "imgUrl": "https://img10.360buyimg.com/...",
          "materialUrl": "https://union-click.jd.com/..."  // 推广链接
        }
      ]
    }
  }}