义乌购关键词商品搜索接口实战：双签名兼容 + 批发多维度筛选 + 分页限流防护（Python 合规生产版）

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

class YiWuKeywordSearchClient:
    def __init__(self, app_key, app_secret, is_new_key=True):
        self.app_key = app_key
        self.app_secret = app_secret
        self.is_new_key = is_new_key  # True=2024后密钥SHA1，False=老密钥MD5
        self.api_url = "https://api.yiwugo.com/search/goods"
        self.session = requests.Session()

    def build_signature(self, params):
        """双算法兼容标准签名，过滤空参数并编码中文"""
        valid_params = {k: v for k, v in params.items() if v is not None and str(v).strip() != ""}
        sorted_kv = sorted(valid_params.items(), key=lambda x: x[0])
        sign_parts = []
        for k, v in sorted_kv:
            val_encode = quote(str(v), encoding="utf-8")
            sign_parts.append(f"{k}={val_encode}")
        sign_raw_str = "&".join(sign_parts) + f"&secret={self.app_secret}"
        if self.is_new_key:
            sign_res = hashlib.sha1(sign_raw_str.encode("utf-8"))
        else:
            sign_res = hashlib.md5(sign_raw_str.encode("utf-8"))
        return sign_res.hexdigest().upper()

    def single_page_search(self, keyword, page=1, page_size=30, min_price=None, max_price=None, min_buy=None, sort="sales_desc"):
        """单页关键词搜索，支持批发多条件筛选"""
        timestamp = int(time.time())
        base_params = {
            "app_key": self.app_key,
            "keyword": keyword,
            "timestamp": timestamp,
            "page": page,
            "page_size": min(page_size, 50),
            "sort": sort
        }
        if min_price:
            base_params["min_price"] = min_price
        if max_price:
            base_params["max_price"] = max_price
        if min_buy:
            base_params["min_buy"] = min_buy
        base_params["sign"] = self.build_signature(base_params)
        try:
            resp = self.session.get(self.api_url, params=base_params, timeout=15)
            raw_data = resp.json()
            # 429限流指数退避重试
            if raw_data.get("code") == 429:
                time.sleep(3)
                return self.single_page_search(keyword, page, page_size, min_price, max_price, min_buy, sort)
            if raw_data.get("code") != 200:
                return {"code": -1, "msg": raw_data.get("msg", "接口调用失败"), "list": []}
            data_body = raw_data.get("data", {})
            goods_raw = data_body.get("goods_list", [])
            goods_clean = []
            # 批发商品标准化清洗
            for item in goods_raw:
                goods_clean.append({
                    "goods_id": item.get("goods_id"),
                    "title": item.get("title"),
                    "market_shop": item.get("market_shop", ""),
                    "wholesale_price": item.get("price"),
                    "min_buy": item.get("min_buy"),
                    "stock": item.get("stock"),
                    "warning_stock": item.get("warning_stock", 0),
                    "sales_volume": item.get("sales"),
                    "main_img": item.get("main_image"),
                    "supplier_name": item.get("supplier_name")
                })
            # 免费版强制1秒间隔，规避风控
            time.sleep(1)
            return {
                "code": 200,
                "total": data_body.get("total", 0),
                "total_page": data_body.get("total_page", 0),
                "current_page": page,
                "goods_list": goods_clean
            }
        except Exception as e:
            return {"code": -2, "msg": f"网络异常：{str(e)}", "list": []}

    def get_all_keyword_goods(self, keyword, min_price=None, max_price=None, min_buy=None, sort="sales_desc"):
        """自动循环拉取关键词全部货源，无需手动分页"""
        all_goods = []
        current_page = 1
        while True:
            page_result = self.single_page_search(keyword, current_page, 30, min_price, max_price, min_buy, sort)
            if page_result["code"] != 200 or len(page_result["goods_list"]) == 0:
                break
            all_goods.extend(page_result["goods_list"])
            if current_page >= page_result["total_page"]:
                break
            current_page += 1
        return {"keyword": keyword, "total_matched": len(all_goods), "goods_all": all_goods}

# 调用示例
if __name__ == "__main__":
    client = YiWuKeywordSearchClient(
        app_key="开发者后台申请的AppKey",
        app_secret="开发者后台密钥AppSecret",
        is_new_key=True
    )
    # 搜索饰品，价格5-50元，起批≥10件，按销量排序
    result = client.get_all_keyword_goods(keyword="饰品", min_price=5, max_price=50, min_buy=10)
    print(json.dumps(result, ensure_ascii=False, indent=2))