京东联盟关键词搜索接口实战：多条件组合筛选 + 券佣结构化 + 请求重试防风控（Python 落地）

import requests
import hashlib
import time
import json
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class JdKeywordSearch:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_gateway = "https://api.jd.com/routerjson"
        # 配置重试策略
        self.session = requests.Session()
        retry_rule = Retry(total=3, backoff_factor=0.4, status_forcelist=[429,500,503])
        self.session.mount("https://", HTTPAdapter(max_retries=retry_rule))

    def get_md5_sign(self, params):
        """京东标准MD5签名算法"""
        sorted_kv = sorted(params.items(), key=lambda x:x[0])
        sign_str = self.app_secret
        for k,v in sorted_kv:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        return hashlib.md5(sign_str.encode("utf8")).hexdigest().upper()

    def calc_hot_score(self, sale_num, comm_rate):
        """爆款潜力打分：销量权重60%+佣金40%"""
        s_score = min(sale_num/5000*60,60)
        c_score = min(comm_rate*4,40)
        return round(s_score+c_score,2)

    def search_goods(self, keyword, page=1, page_size=20, min_price=None,max_price=None,
                     is_self=0,has_coupon=0,min_comm=0,category_id=None):
        """
        关键词搜品，多条件可选入参
        :param keyword:搜索词；is_self:1仅自营；has_coupon:1仅带券
        """
        biz_param = {"keyword":keyword,"pageIndex":page,"pageSize":page_size}
        # 动态拼接筛选参数，空值不传入
        if min_price:biz_param["priceFrom"]=min_price
        if max_price:biz_param["priceTo"]=max_price
        if is_self==1:biz_param["isSelf"]=1
        if has_coupon==1:biz_param["hasCoupon"]=1
        if min_comm>0:biz_param["commissionShareStart"]=min_comm
        if category_id:biz_param["cid3"]=category_id

        public_params = {
            "app_key":self.app_key,"method":"jd.union.open.goods.search",
            "timestamp":time.strftime("%Y-%m-%d %H:%M:%S"),"format":"json","v":"2.0",
            "param_json":json.dumps(biz_param,ensure_ascii=False)
        }
        public_params["sign"] = self.get_md5_sign(public_params)
        try:
            resp = self.session.post(self.api_gateway,data=public_params,timeout=12)
            res = resp.json()
            if "error_response" in res:
                return {"code":-1,"msg":res["error_response"]["msg"]}
            resp_data = res["jd_union_open_goods_search_response"]["result"]
            goods_raw = resp_data.get("goodsInfoList",[])
            result_list = []
            for item in goods_raw:
                coupon_info = item.get("couponInfo",{})
                comm_info = item.get("commissionInfo",{})
                sale = int(item.get("inOrderCount30Days",0))
                comm_rate = float(comm_info.get("commissionRate",0))
                result_list.append({
                    "skuId":item.get("skuId"),"title":item.get("goodsName"),
                    "market_price":item.get("price"),"coupon_end_price":coupon_info.get("couponEndPrice"),
                    "commission_rate":comm_rate,"commission_amt":comm_info.get("commission"),
                    "is_jd_self":item.get("isJdSelf"),"month_sale":sale,
                    "hot_score":self.calc_hot_score(sale,comm_rate),"main_img":item.get("imgUrl")
                })
            time.sleep(1) # 防风控休眠
            return {"code":200,"total":resp_data.get("total"),"data":result_list}
        except Exception as e:
            return {"code":-2,"msg":f"请求异常：{str(e)}"}

# 调用示例
if __name__ == "__main__":
    ak = "替换你的AppKey"
    as_sec = "替换你的AppSecret"
    api = JdKeywordSearch(ak,as_sec)
    # 筛选：蓝牙耳机、100-300元、自营+带券、佣金≥5%
    ret = api.search_goods(keyword="蓝牙耳机",min_price=100,max_price=300,is_self=1,has_coupon=1,min_comm=5)
    print(json.dumps(ret,ensure_ascii=False,indent=2))