【实战解析】淘宝店铺商品全量爬取：item_search_shop API 深度指南

import requests
def fetch_shop_products(shop_id, page=1):    params = {        "key": "your_key",        "shop_id": shop_id,        "page": page,        "sort": "sales-desc",  # 按销量降序        "format": "jsonu"  # 中文友好格式    }    response = requests.get("https://api-gw.onebound.cn/taobao/item_search_shop", params=params)    return response.json()
# 分页获取全量商品def get_all_products(shop_id):    products = []    page = 1    while True:        data = fetch_shop_products(shop_id, page)        if not data["items"]["item"]:            break        products.extend(data["items"]["item"])        page += 1    return products

{  "items": {    "shop_id": "440688975",  // 店铺ID    "total_results": 240,  // 商品总数    "item": [      {        "num_iid": "583975606695",  // 商品ID        "title": "jifeng季风雪茄盒",  // 标题        "price": 276.00,  // 原价        "promotion_price": 138.00,  // 促销价        "sales": 4,  // 销量        "pic_url": "https://img.alicdn.com/xxx.jpg"  // 主图      }    ]  }}

# 定时任务：每日0点更新竞品数据import scheduleimport time
def update_competitor_data():    competitor_shops = ["440688975", "55123456"]    for shop_id in competitor_shops:        products = get_all_products(shop_id)        for p in products:            # 保存到数据库（示例）            db.execute("""                INSERT INTO products (num_iid, title, price, sales)                VALUES (%s, %s, %s, %s)                ON DUPLICATE KEY UPDATE price=VALUES(price), sales=VALUES(sales)            """, (p["num_iid"], p["title"], p["price"], p["sales"]))
schedule.every().day.at("00:00").do(update_competitor_data)while True:    schedule.run_pending()    time.sleep(1)

def select_top_products(products):    return [p for p in products             if p["sales"] > 1000             and p["promotion_price"] < 200             and "夏季新款" in p["category"]]

# 分布式请求队列（示例）from concurrent.futures import ThreadPoolExecutor
def process_shop(shop_id):    try:        return get_all_products(shop_id)    except Exception as e:        print(f"处理失败：{shop_id} - {str(e)}")
# 并行处理5个店铺with ThreadPoolExecutor(max_workers=5) as executor:    results = executor.map(process_shop, ["440688975", "55123456", ...])

# 随机延迟（0.5-2秒）time.sleep(random.uniform(0.5, 2.0))
# 轮换User-Agentheaders = {"User-Agent": fake_useragent.UserAgent().random}

from requests_rotating_proxy import RotatingProxyManagerproxy_manager = RotatingProxyManager([    "http://user:pass@proxy1.com:8080",    "http://user:pass@proxy2.com:8080"])