淘宝 API 关键词搜索接口深度解析：请求参数、签名机制与性能优化

app_key=2356789012&format=json&method=taobao.items.search&page_no=1&page_size=20&q=手机&timestamp=2025-12-03+14%3A30%3A00&v=2.0

import hashlib
import urllib.parse

def generate_sign(params, app_secret):
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接为key=value格式，URL编码value
    encoded_params = []
    for key, value in sorted_params:
        encoded_value = urllib.parse.quote(str(value), safe='')
        encoded_params.append(f"{key}={encoded_value}")
    sign_str = '&'.join(encoded_params)
    # 3. 首尾拼接AppSecret
    sign_str = f"{app_secret}{sign_str}{app_secret}"
    # 4. MD5加密
    md5 = hashlib.md5()
    md5.update(sign_str.encode('utf-8'))
    return md5.hexdigest().lower()

# 测试
if __name__ == "__main__":
    app_secret = "your_app_secret"
    params = {
        "app_key": "2356789012",
        "method": "taobao.items.search",
        "format": "json",
        "timestamp": "2025-12-03 14:30:00",
        "v": "2.0",
        "q": "手机",
        "page_no": 1,
        "page_size": 20
    }
    sign = generate_sign(params, app_secret)
    print("生成的签名：", sign)

pip install requests

import requests
import hashlib
import urllib.parse
from datetime import datetime

class TaobaoSearchAPI:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "http://gw.api.taobao.com/router/rest"  # 正式环境URL
        # 测试环境URL：http://gw.api.tbsandbox.com/router/rest

    def generate_sign(self, params):
        """生成签名（复用3.3节实现）"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        encoded_params = []
        for key, value in sorted_params:
            encoded_value = urllib.parse.quote(str(value), safe='')
            encoded_params.append(f"{key}={encoded_value}")
        sign_str = '&'.join(encoded_params)
        sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"
        md5 = hashlib.md5()
        md5.update(sign_str.encode('utf-8'))
        return md5.hexdigest().lower()

    def search_items(self, keyword, page_no=1, page_size=20, **kwargs):
        """
        关键词搜索商品
        :param keyword: 搜索关键词
        :param page_no: 页码
        :param page_size: 每页条数（1-40）
        :param kwargs: 其他业务参数（如cat、price_min、price_max、sort等）
        :return: 解析后的商品数据
        """
        # 1. 构建公共参数
        params = {
            "app_key": self.app_key,
            "method": "taobao.items.search",
            "format": "json",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "page_no": page_no,
            "page_size": min(page_size, 40),  # 限制最大条数
            "q": keyword
        }

        # 2. 添加额外业务参数
        params.update(kwargs)

        # 3. 生成签名
        sign = self.generate_sign(params)
        params["sign"] = sign

        # 4. 发送HTTP请求
        try:
            response = requests.get(self.base_url, params=params, timeout=10)
            response.raise_for_status()  # 抛出HTTP错误
            result = response.json()

            # 5. 解析响应
            if "error_response" in result:
                error = result["error_response"]
                raise Exception(f"接口调用失败：{error['msg']}（错误码：{error['code']}）")
            
            # 提取商品列表
            items = result["items_search_response"]["items"]["item"]
            total = result["items_search_response"]["total_results"]
            return {
                "total": total,
                "page_no": page_no,
                "page_size": page_size,
                "items": items
            }
        except Exception as e:
            print(f"搜索失败：{str(e)}")
            return None

# 示例调用
if __name__ == "__main__":
    # 替换为你的AppKey和AppSecret
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"

    # 初始化API客户端
    taobao_api = TaobaoSearchAPI(APP_KEY, APP_SECRET)

    # 搜索关键词"手机"，筛选价格1000-5000元，销量优先排序
    result = taobao_api.search_items(
        keyword="手机",
        page_no=1,
        page_size=30,
        price_min=1000,
        price_max=5000,
        sort="sales_desc"
    )

    # 打印结果
    if result:
        print(f"总商品数：{result['total']}")
        print(f"当前页商品数：{len(result['items'])}")
        for item in result["items"][:3]:  # 打印前3个商品
            print(f"商品标题：{item['title']}")
            print(f"价格：{item['price']}元")
            print(f"销量：{item['sale_count']}")
            print(f"商品链接：{item['detail_url']}\n")

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=1, max=5),  # 指数退避
    retry=retry_if_exception_type((requests.exceptions.RequestException, Exception))
)
def search_items_with_retry(self, keyword, **kwargs):
    return self.search_items(keyword, **kwargs)