python采集淘宝关键词搜索API接口系列，json数据返回

import requestsimport hashlibimport timeimport jsonimport loggingfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typeimport redis  # 可选，缓存用# ==================== 配置项 ====================# 1. 淘宝开放平台信息（替换为自己的）APP_KEY = "你的App Key"APP_SECRET = "你的App Secret"API_URL = "https://eco.taobao.com/router/rest"# 2. 日志配置（记录调用过程和异常）logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("taobao_api.log"), logging.StreamHandler()])logger = logging.getLogger(__name__)# 3. 缓存配置（可选，减少重复调用）REDIS_CLIENT = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)CACHE_EXPIRE = 30 * 60  # 缓存有效期30分钟# ==================== 核心工具函数 ====================def generate_sign(params: dict) -> str:
    """生成淘宝API签名（核心步骤，严格遵循规则）"""
    try:
        # 1. 按参数名ASCII升序排序，过滤空值
        sorted_params = sorted([(k, v) for k, v in params.items() if v is not None and v != ""], key=lambda x: x[0])
        # 2. 拼接参数字符串
        sign_str = "".join([f"{k}{v}" for k, v in sorted_params])
        # 3. 首尾拼接App Secret，MD5加密转大写
        sign_str = APP_SECRET + sign_str + APP_SECRET
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign    except Exception as e:
        logger.error(f"生成签名失败：{str(e)}")
        raise@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=5),  # 重试间隔：2s→4s→8s（最大5s）
    retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)),
    reraise=True  # 重试失败后抛出异常)def call_taobao_api(params: dict) -> dict:
    """调用淘宝API，包含重试机制"""
    try:
        # 添加签名
        params["sign"] = generate_sign(params)
        # 发起GET请求
        response = requests.get(
            API_URL,
            params=params,
            timeout=10,  # 超时时间10秒
            headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        )
        response.encoding = "utf-8"
        result = response.json()
        logger.info(f"API调用返回：{json.dumps(result, ensure_ascii=False)[:200]}")  # 日志只打印前200字符
        
        # 检查调用是否成功
        if result.get("code") != 0:
            error = result["resp_data"]["error_response"]
            error_msg = f"API调用失败：{error['msg']}（错误码：{error['sub_code']}）"
            logger.error(error_msg)
            raise Exception(error_msg)
        return result    except requests.exceptions.Timeout:
        logger.error("API调用超时，将重试")
        raise
    except requests.exceptions.ConnectionError:
        logger.error("API连接失败，将重试")
        raise
    except Exception as e:
        logger.error(f"API调用异常：{str(e)}")
        raisedef get_cache(key: str) -> dict | None:
    """从Redis获取缓存"""
    try:
        cache_data = REDIS_CLIENT.get(key)
        if cache_data:
            return json.loads(cache_data)
        return None
    except Exception as e:
        logger.warning(f"读取缓存失败：{str(e)}")
        return Nonedef set_cache(key: str, data: dict):
    """设置Redis缓存"""
    try:
        REDIS_CLIENT.setex(key, CACHE_EXPIRE, json.dumps(data, ensure_ascii=False))
    except Exception as e:
        logger.warning(f"设置缓存失败：{str(e)}")# ==================== 核心采集函数 ====================def collect_taobao_keyword(keyword: str, page_no: int = 1, page_size: int = 20, use_cache: bool = True) -> tuple[dict, list]:
    """
    采集淘宝关键词搜索数据
    :param keyword: 搜索关键词
    :param page_no: 页码
    :param page_size: 每页条数（最大100）
    :param use_cache: 是否使用缓存
    :return: (搜索元信息, 商品列表)
    """
    # 1. 构建缓存key
    cache_key = f"taobao_search:{keyword}:{page_no}:{page_size}"
    if use_cache:
        cache_data = get_cache(cache_key)
        if cache_data:
            logger.info(f"命中缓存：{cache_key}")
            return cache_data["meta"], cache_data["goods"]
    
    # 2. 构造API参数
    params = {
        "method": "taobao.tbk.item.search",
        "app_key": APP_KEY,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "format": "json",
        "v": "2.0",
        "sign_method": "md5",
        "q": keyword,
        "page_no": page_no,
        "page_size": page_size,
        "sort": "sales_desc"  # 按销量降序
    }
    
    # 3. 调用API
    result = call_taobao_api(params)
    
    # 4. 解析JSON数据
    resp_data = result["resp_data"]["tbk_item_search_response"]
    # 4.1 解析元信息
    meta = {
        "总商品数": resp_data.get("total_results", 0),
        "当前页码": resp_data.get("page_no", 1),
        "每页条数": resp_data.get("page_size", 20),
        "总页数": resp_data.get("max_page", 0)
    }
    # 4.2 解析商品列表
    goods_list = []
    for item in resp_data.get("results", []):
        goods_info = {
            "商品ID": item.get("num_iid"),
            "标题": item.get("title"),
            "主图链接": item.get("pict_url"),
            "券后价": float(item.get("zk_final_price", 0)),
            "原价": float(item.get("reserve_price", 0)),
            "销量": item.get("sales", 0),
            "店铺名称": item.get("shop_title"),
            "是否天猫": item.get("is_tmall", False),
            "佣金比例": float(item.get("tk_rate", 0)) / 100,
            "发货地": item.get("provcity"),
            "商品链接": item.get("item_url")
        }
        goods_list.append(goods_info)
    
    # 5. 设置缓存
    if use_cache:
        set_cache(cache_key, {"meta": meta, "goods": goods_list})
    
    logger.info(f"采集完成：关键词[{keyword}] 页码[{page_no}] 共{len(goods_list)}条商品")
    return meta, goods_listdef batch_collect(keyword: str, max_page: int = 5, page_size: int = 20):
    """批量采集多页数据"""
    all_goods = []
    try:
        # 先获取第一页，拿到总页数
        meta, first_page_goods = collect_taobao_keyword(keyword, page_no=1, page_size=page_size)
        all_goods.extend(first_page_goods)
        total_page = min(meta["总页数"], max_page)  # 最多采集max_page页
        
        # 采集后续页码
        for page in range(2, total_page + 1):
            time.sleep(1)  # 休眠1秒，避免限流
            _, goods = collect_taobao_keyword(keyword, page_no=page, page_size=page_size)
            all_goods.extend(goods)
        
        # 保存到本地JSON文件
        with open(f"taobao_{keyword}_data.json", "w", encoding="utf-8") as f:
            json.dump({
                "关键词": keyword,
                "采集时间": time.strftime("%Y-%m-%d %H:%M:%S"),
                "总条数": len(all_goods),
                "商品列表": all_goods            }, f, ensure_ascii=False, indent=2)
        logger.info(f"批量采集完成：共{len(all_goods)}条商品，已保存到文件")
        return all_goods    except Exception as e:
        logger.error(f"批量采集失败：{str(e)}")
        return []# ==================== 调用示例 ====================if __name__ == "__main__":
    # 单页采集
    # meta, goods = collect_taobao_keyword("夏季纯棉T恤", page_no=1, page_size=20)
    # print("元信息：", json.dumps(meta, ensure_ascii=False, indent=2))
    # print("商品列表（前2条）：", json.dumps(goods[:2], ensure_ascii=False, indent=2))
    
    # 批量采集（最多5页）
    batch_collect("夏季纯棉T恤", max_page=5, page_size=20)