# 安装依赖
# pip install requests pandas schedule sqlite3

import requests
import json
import pandas as pd
import sqlite3
import schedule
import time
from datetime import datetime

# Pangolinfo API 配置
API_ENDPOINT = "https://api.pangolinfo.com/scrape"
API_KEY = "your_pangolinfo_api_key"  # 在 https://tool.pangolinfo.com 获取

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

def scrape_category_top100(category_url: str, marketplace: str = "US") -> dict:
    """
    采集亚马逊单个类目的 Top 100 Best Sellers 数据
    
    Args:
        category_url: 亚马逊类目 Best Sellers 页面 URL
                      例如: https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics/
        marketplace: 站点代码，支持 US/UK/DE/JP/CA/FR/IT/ES
    
    Returns:
        包含 Top 100 商品数据的字典，字段包括:
        - rank: 排名 (1-100)
        - asin: 商品 ASIN
        - title: 商品标题
        - price: 当前价格
        - rating: 评分 (0-5.0)
        - review_count: 评论数量
        - brand: 品牌名称
        - is_prime: 是否 Prime
        - badge: 徽章类型 (Best Seller / Amazon's Choice / None)
        - subcategory_path: 完整类目路径
        - image_url: 主图 URL
        - scraped_at: 采集时间戳
    """
    payload = {
        "url": category_url,
        "marketplace": marketplace,
        "output_format": "json",          # 返回结构化 JSON
        "parse_template": "amazon_bestsellers",  # 使用亚马逊榜单专用解析模板
        "include_fields": [
            "rank", "asin", "title", "price", "rating",
            "review_count", "brand", "is_prime", "badge",
            "subcategory_path", "image_url", "customer_says"
        ]
    }
    
    try:
        response = requests.post(
            API_ENDPOINT,
            headers=HEADERS,
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        
        data = response.json()
        
        # 为每条记录添加采集时间戳
        timestamp = datetime.utcnow().isoformat()
        for item in data.get("products", []):
            item["scraped_at"] = timestamp
            item["category_url"] = category_url
            item["marketplace"] = marketplace
        
        return data
        
    except requests.exceptions.Timeout:
        print(f"[ERROR] 请求超时: {category_url}")
        return {"products": [], "error": "timeout"}
    except requests.exceptions.HTTPError as e:
        print(f"[ERROR] HTTP 错误 {e.response.status_code}: {category_url}")
        return {"products": [], "error": str(e)}

from concurrent.futures import ThreadPoolExecutor, as_completed

# 定义需要监控的类目列表（示例：20 个核心类目）
CATEGORY_URLS = [
    "https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics/",
    "https://www.amazon.com/Best-Sellers-Home-Kitchen/zgbs/kitchen/",
    "https://www.amazon.com/Best-Sellers-Sports-Outdoors/zgbs/sporting-goods/",
    "https://www.amazon.com/Best-Sellers-Toys-Games/zgbs/toys-and-games/",
    "https://www.amazon.com/Best-Sellers-Beauty/zgbs/beauty/",
    "https://www.amazon.com/Best-Sellers-Health-Personal-Care/zgbs/hpc/",
    "https://www.amazon.com/Best-Sellers-Pet-Supplies/zgbs/pet-supplies/",
    "https://www.amazon.com/Best-Sellers-Clothing/zgbs/apparel/",
    "https://www.amazon.com/Best-Sellers-Books/zgbs/books/",
    "https://www.amazon.com/Best-Sellers-Tools-Home-Improvement/zgbs/hi/",
    # ... 可继续扩展至数百个子类目
]

def batch_scrape_categories(category_urls: list, max_workers: int = 5) -> list:
    """
    并发批量采集多个类目的 Top 100 数据
    
    max_workers 建议值：5-10，过高可能触发 API 并发限制
    实际并发上限以 Pangolinfo 控制台的套餐配置为准
    """
    all_products = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_url = {
            executor.submit(scrape_category_top100, url): url
            for url in category_urls
        }
        
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                products = result.get("products", [])
                all_products.extend(products)
                print(f"[OK] 采集完成: {url} — 获取 {len(products)} 条记录")
            except Exception as e:
                print(f"[FAIL] {url}: {e}")
    
    return all_products

def setup_database(db_path: str = "amazon_top100.db"):
    """初始化 SQLite 数据库，创建榜单数据表"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS category_top100 (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            scraped_at TEXT NOT NULL,
            marketplace TEXT NOT NULL,
            category_url TEXT NOT NULL,
            rank INTEGER NOT NULL,
            asin TEXT NOT NULL,
            title TEXT,
            price REAL,
            rating REAL,
            review_count INTEGER,
            brand TEXT,
            is_prime INTEGER,
            badge TEXT,
            subcategory_path TEXT,
            image_url TEXT,
            customer_says TEXT,
            UNIQUE(scraped_at, asin, category_url)  -- 防重复写入
        )
    """)
    
    # 创建常用查询索引
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_asin ON category_top100(asin)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_scraped_at ON category_top100(scraped_at)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_category ON category_top100(category_url)")
    
    conn.commit()
    return conn

def save_products(conn: sqlite3.Connection, products: list):
    """将采集数据写入数据库"""
    cursor = conn.cursor()
    saved_count = 0
    
    for p in products:
        try:
            cursor.execute("""
                INSERT OR IGNORE INTO category_top100 
                (scraped_at, marketplace, category_url, rank, asin, title, price,
                 rating, review_count, brand, is_prime, badge, subcategory_path,
                 image_url, customer_says)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                p.get("scraped_at"), p.get("marketplace"), p.get("category_url"),
                p.get("rank"), p.get("asin"), p.get("title"), p.get("price"),
                p.get("rating"), p.get("review_count"), p.get("brand"),
                1 if p.get("is_prime") else 0, p.get("badge"),
                p.get("subcategory_path"), p.get("image_url"),
                p.get("customer_says")
            ))
            saved_count += 1
        except Exception as e:
            print(f"[DB ERROR] ASIN {p.get('asin')}: {e}")
    
    conn.commit()
    print(f"[DB] 成功写入 {saved_count}/{len(products)} 条记录")

def run_daily_collection():
    """每日定时采集任务主函数"""
    print(f"\n{'='*50}")
    print(f"开始采集任务: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*50}")
    
    # 1. 批量采集
    products = batch_scrape_categories(CATEGORY_URLS, max_workers=5)
    print(f"\n[汇总] 本次采集共获取 {len(products)} 条商品数据")
    
    # 2. 写入数据库
    conn = setup_database()
    save_products(conn, products)
    conn.close()
    
    # 3. 生成简报（可选：发送到 Slack / 企业微信）
    generate_daily_report(products)

def generate_daily_report(products: list):
    """生成每日榜单变化摘要报告"""
    df = pd.DataFrame(products)
    if df.empty:
        print("[WARN] 本次采集无数据，跳过报告生成")
        return
    
    # 按类目分组，统计每个类目的 Top 10
    report_lines = ["=== 每日类目 Top 10 速览 ===\n"]
    
    for category, group in df.groupby("category_url"):
        top10 = group.nsmallest(10, "rank")[["rank", "asin", "title", "price", "rating"]]
        category_name = category.split("/zgbs/")[-1].rstrip("/").replace("-", " ").title()
        report_lines.append(f"\n【{category_name}】")
        for _, row in top10.iterrows():
            report_lines.append(
                f"  #{int(row['rank'])} {row['title'][:40]}... | "
                f"${row['price']:.2f} | ⭐{row['rating']}"
            )
    
    report = "\n".join(report_lines)
    print(report)
    
    # 保存报告到文件
    report_filename = f"report_{datetime.now().strftime('%Y%m%d')}.txt"
    with open(report_filename, "w", encoding="utf-8") as f:
        f.write(report)

# 设置每日 UTC 08:00 运行（对应北京时间 16:00）
schedule.every().day.at("08:00").do(run_daily_collection)

# 也可以配置多个采集频率
# schedule.every(6).hours.do(run_daily_collection)  # 每 6 小时采集一次（适合热门类目）

if __name__ == "__main__":
    print("亚马逊类目 Top 100 数据采集任务已启动...")
    run_daily_collection()  # 立即执行一次
    
    while True:
        schedule.run_pending()
        time.sleep(60)

def analyze_rank_changes(db_path: str = "amazon_top100.db", days: int = 7):
    """
    分析过去 N 天内类目榜单的排名变化
    识别快速上升的商品（潜在机会）和快速下降的商品（竞品预警）
    """
    conn = sqlite3.connect(db_path)
    
    query = """
        WITH ranked_data AS (
            SELECT 
                asin, title, brand, price, rating, review_count,
                rank, category_url, scraped_at,
                ROW_NUMBER() OVER (
                    PARTITION BY asin, category_url 
                    ORDER BY scraped_at ASC
                ) as day_seq
            FROM category_top100
            WHERE scraped_at >= datetime('now', ? || ' days')
        ),
        first_last AS (
            SELECT 
                asin, title, brand, category_url,
                MAX(CASE WHEN day_seq = 1 THEN rank END) as rank_start,
                MIN(rank) as rank_best,
                MAX(CASE WHEN day_seq = (
                    SELECT MAX(day_seq) FROM ranked_data r2 
                    WHERE r2.asin = ranked_data.asin 
                    AND r2.category_url = ranked_data.category_url
                ) THEN rank END) as rank_latest,
                AVG(price) as avg_price,
                MAX(review_count) as latest_reviews
            FROM ranked_data
            GROUP BY asin, category_url
        )
        SELECT 
            *,
            (rank_start - rank_latest) as rank_improvement,
            CASE 
                WHEN (rank_start - rank_latest) > 20 THEN '🚀 快速上升'
                WHEN (rank_start - rank_latest) < -20 THEN '⚠️ 快速下降'
                ELSE '📊 相对稳定'
            END as trend_label
        FROM first_last
        ORDER BY rank_improvement DESC
    """
    
    df = pd.read_sql_query(query, conn, params=(f"-{days}",))
    conn.close()
    
    print(f"\n=== 过去 {days} 天排名变化分析 ===")
    print(f"\n🚀 上升最快的 10 个商品:")
    top_risers = df[df['rank_improvement'] > 0].head(10)
    for _, row in top_risers.iterrows():
        print(f"  ASIN: {row['asin']} | {row['title'][:35]}...")
        print(f"  排名: #{row['rank_start']} → #{row['rank_latest']} "
              f"(上升 {row['rank_improvement']} 位) | 均价: ${row['avg_price']:.2f}")
    
    return df