import requestsimport randomimport timefrom typing import Dict, Optional# 1. 配置请求头（模拟浏览器，避免被识别为爬虫）def get_headers() -> Dict:
    user_agents = [        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/115.0"
    ]    return {        "User-Agent": random.choice(user_agents),        "Cookie": "你的淘宝登录Cookie（从浏览器复制）",  # 需替换为真实Cookie
        "Referer": "https://item.taobao.com/"
    }# 2. 爬取单个商品详情def crawl_taobao_item(item_id: str) -> Optional[Dict]:    # 淘宝商品详情接口（示例，实际需抓包确认最新接口）
    url = f"https://h5api.m.taobao.com/h5/mtop.taobao.detail.getdetail/6.0/?id={item_id}"
    headers = get_headers()    
    try:        # 加随机延时（1-3秒），降低反爬风险
        time.sleep(random.uniform(1, 3))
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()  # 触发HTTP错误（如403、500）
        
        # 解析JSON数据，提取核心字段（字段名需按实际接口调整）
        data = response.json()
        item_info = data["data"]["itemInfoModel"]["itemInfo"]        
        # 结构化返回结果（只保留客户需要的字段）
        return {            "item_id": item_id,            "title": item_info["title"],            "price": item_info["price"],            "stock": item_info["totalStock"],            "sales_count": item_info["sellCount"],            "shop_name": data["data"]["sellerInfoModel"]["sellerName"],            "detail_images": [img["url"] for img in data["data"]["itemImageModel"]["largeImages"]]
        }    
    except Exception as e:        print(f"爬取商品{item_id}失败：{str(e)}")        return None

import sqlite3# 初始化数据库表def init_db():
    conn = sqlite3.connect("taobao_items.db")
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS items (
            item_id TEXT PRIMARY KEY,
            title TEXT,
            price TEXT,
            stock INTEGER,
            sales_count INTEGER,
            shop_name TEXT,
            detail_images TEXT,
            crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    conn.close()# 存储商品数据到数据库def save_to_db(item_data: Dict):
    conn = sqlite3.connect("taobao_items.db")
    cursor = conn.cursor()    # 将图片列表转为字符串存储（取出时需用json.loads解析）
    detail_images_str = ",".join(item_data["detail_images"])
    cursor.execute('''
        INSERT OR REPLACE INTO items 
        (item_id, title, price, stock, sales_count, shop_name, detail_images)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', (
        item_data["item_id"],
        item_data["title"],
        item_data["price"],
        item_data["stock"],
        item_data["sales_count"],
        item_data["shop_name"],
        detail_images_str
    ))
    conn.commit()
    conn.close()

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModel  # 用于参数校验import json# 初始化FastAPI应用app = FastAPI(title="淘宝商品详情API", version="1.0")# 初始化数据库（若使用存储功能）init_db()# 定义请求参数模型（校验商品ID格式，可选）class ItemRequest(BaseModel):
    item_id: str# 定义API接口：获取商品详情@app.post("/taobao/item/detail", summary="根据商品ID获取淘宝详情")def get_item_detail(request: ItemRequest):
    item_id = request.item_id    # 1. 先查数据库，有缓存则直接返回（减少爬虫请求）
    conn = sqlite3.connect("taobao_items.db")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM items WHERE item_id = ?", (item_id,))
    db_data = cursor.fetchone()
    conn.close()    
    if db_data:        # 数据库数据转为JSON格式（处理图片列表）
        return {            "item_id": db_data[0],            "title": db_data[1],            "price": db_data[2],            "stock": db_data[3],            "sales_count": db_data[4],            "shop_name": db_data[5],            "detail_images": db_data[6].split(","),            "source": "数据库缓存"
        }    
    # 2. 数据库无缓存，调用爬虫获取数据
    crawl_data = crawl_taobao_item(item_id)    if not crawl_data:        raise HTTPException(status_code=400, detail=f"商品ID {item_id} 爬取失败，请检查ID有效性或稍后重试")    
    # 3. 存储到数据库（更新缓存）
    save_to_db(crawl_data)
    crawl_data["source"] = "实时爬取"
    return crawl_data# 启动API服务（需安装uvicorn：pip install uvicorn）if __name__ == "__main__":    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)