pip install requests pymongo tenacity ratelimit python-dotenv

import hmac
import hashlib
from urllib.parse import urlencode
import time
from dotenv import load_dotenv
import os

# 加载环境变量（避免硬编码App Key/Secret）
load_dotenv()
APP_KEY = os.getenv("TAOBAO_APP_KEY")
APP_SECRET = os.getenv("TAOBAO_APP_SECRET")
API_GATEWAY = "http://gw.api.taobao.com/router/rest"  # 淘宝API网关


def generate_taobao_sign(params: dict) -> str:
    """生成淘宝API签名"""
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接参数字符串（key1value1key2value2）
    param_str = "".join([f"{k}{v}" for k, v in sorted_params])
    # 3. 前后添加App Secret，计算HMAC-SHA1签名
    sign_str = APP_SECRET + param_str + APP_SECRET
    sign = hmac.new(
        APP_SECRET.encode("utf-8"),
        sign_str.encode("utf-8"),
        hashlib.sha1
    ).hexdigest().upper()
    return sign


def build_taobao_request(method: str, biz_params: dict) -> dict:
    """构建淘宝API请求参数（含公共参数+业务参数）"""
    # 公共参数（所有API必传）
    public_params = {
        "app_key": APP_KEY,
        "method": method,
        "format": "json",  # 返回格式
        "v": "2.0",        # API版本
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "sign_method": "hmac",  # 签名方式
    }
    # 合并公共参数与业务参数
    all_params = {**public_params, **biz_params}
    # 生成签名并添加
    all_params["sign"] = generate_taobao_sign(all_params)
    return all_params

import requests
from pymongo import MongoClient
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from ratelimit import limits, sleep_and_retry
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("taobao_spider")

# 初始化MongoDB连接（存储商品数据）
mongo_client = MongoClient("mongodb://localhost:27017/")
db = mongo_client["taobao_data"]
product_collection = db["products"]


# 1. 限流装饰器：控制API调用频率（淘宝免费版QPS=5，即1秒5次）
@sleep_and_retry
@limits(calls=5, period=1)
def send_taobao_request(params: dict) -> dict:
    """发送淘宝API请求并返回解析后的JSON数据"""
    try:
        response = requests.get(
            url=API_GATEWAY,
            params=params,
            timeout=10  # 超时时间10秒，避免长期阻塞
        )
        response.raise_for_status()  # 触发HTTP错误（如403、500）
        result = response.json()
        # 检查API返回错误（如权限不足、参数错误）
        if "error_response" in result:
            error = result["error_response"]
            logger.error(f"API Error: {error['msg']} (code: {error['code']})")
            raise Exception(f"API Error: {error['msg']}")
        return result
    except requests.exceptions.RequestException as e:
        logger.error(f"HTTP Request Error: {str(e)}")
        raise e


# 2. 重试装饰器：临时失败自动重试（如网络波动、网关超时）
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 重试间隔：2s→4s→8s
    retry=retry_if_exception_type((requests.exceptions.RequestException, Exception)),
    after=lambda retry_state: logger.info(f"Retry {retry_state.attempt_number} times")
)
def fetch_taobao_products(keyword: str, page_no: int = 1, page_size: int = 40) -> dict:
    """抓取淘宝商品列表（按关键词搜索）"""
    # 业务参数（根据接口文档定义）
    biz_params = {
        "q": keyword,          # 搜索关键词（如“连衣裙”）
        "page_no": page_no,    # 页码
        "page_size": page_size,# 每页条数（最大40）
        "sort": "sales_desc",  # 排序方式：sales_desc=销量降序
        "price_start": 0,      # 最低价格（可选）
        "price_end": 1000      # 最高价格（可选）
    }
    # 构建完整请求参数
    request_params = build_taobao_request(
        method="taobao.item.search",
        biz_params=biz_params
    )
    # 发送请求并获取结果
    result = send_taobao_request(request_params)
    # 提取商品列表数据
    product_list = result["item_search_response"]["items"]["item"]
    logger.info(f"Fetched {len(product_list)} products for keyword: {keyword}, page: {page_no}")
    return product_list


# 3. 数据存储到MongoDB
def save_products_to_mongo(products: list):
    """将商品数据批量插入MongoDB（去重：按商品ID）"""
    if not products:
        return
    # 为每条数据添加抓取时间
    for product in products:
        product["crawl_time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    # 按num_iid（商品唯一ID）去重，存在则更新，不存在则插入
    bulk_operations = []
    for product in products:
        bulk_operations.append(
            {
                "update_one": {
                    "filter": {"num_iid": product["num_iid"]},  # 商品唯一标识
                    "update": {"$set": product},
                    "upsert": True  # 不存在则插入
                }
            }
        )
    if bulk_operations:
        product_collection.bulk_write(bulk_operations)
        logger.info(f"Saved {len(bulk_operations)} products to MongoDB")


# 4. 全量抓取（多页数据）
def crawl_taobao_product_by_keyword(keyword: str, max_page: int = 5):
    """按关键词全量抓取多页商品数据"""
    for page_no in range(1, max_page + 1):
        try:
            products = fetch_taobao_products(keyword, page_no)
            save_products_to_mongo(products)
            # 避免触发反爬：非限流情况下额外添加0.5秒间隔
            time.sleep(0.5)
        except Exception as e:
            logger.error(f"Failed to crawl page {page_no}: {str(e)}")
            continue


# 执行抓取（示例：抓取“秋季连衣裙”前5页数据）
if __name__ == "__main__":
    crawl_taobao_product_by_keyword(keyword="秋季连衣裙", max_page=5)

import redis

# 初始化Redis连接
redis_client = redis.Redis(host="localhost", port=6379, db=0)

def get_product_detail(num_iid: str) -> dict:
    """获取商品详情（优先从缓存读取）"""
    # 1. 先查缓存
    cache_key = f"taobao:product:{num_iid}"
    cached_data = redis_client.get(cache_key)
    if cached_data:
        logger.info(f"Get product {num_iid} from cache")
        return eval(cached_data.decode("utf-8"))  # 实际项目建议用json.loads
    
    # 2. 缓存未命中，调用API
    biz_params = {"num_iid": num_iid}
    request_params = build_taobao_request("taobao.item.get", biz_params)
    result = send_taobao_request(request_params)
    product_detail = result["item_get_response"]["item"]
    
    # 3. 存入缓存（过期时间3600秒）
    redis_client.setex(cache_key, 3600, str(product_detail))
    return product_detail