import requests
import json
from datetime import datetime
import boto3

class DataCollector:
    def __init__(self, api_key, s3_bucket):
        self.api_key = api_key
        self.base_url = "https://api.pangolinfo.com"
        self.s3 = boto3.client('s3')
        self.bucket = s3_bucket
    
    def collect_product_data(self, asin):
        """采集单个ASIN的实时数据"""
        try:
            response = requests.get(
                f"{self.base_url}/scrape",
                params={
                    "asin": asin,
                    "type": "product",
                    "include": "price,bsr,reviews,inventory"
                },
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"采集失败 {asin}: {str(e)}")
            return None
    
    def save_to_s3(self, asin, data):
        """保存数据到S3,按日期分区"""
        timestamp = datetime.now()
        key = f"amazon-data/{timestamp.strftime('%Y/%m/%d')}/{asin}_{timestamp.strftime('%H%M%S')}.json"
        
        try:
            self.s3.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=json.dumps(data),
                ContentType='application/json'
            )
            print(f"数据已保存: {key}")
        except Exception as e:
            print(f"保存失败: {str(e)}")
    
    def batch_collect(self, asin_list):
        """批量采集并存储"""
        for asin in asin_list:
            data = self.collect_product_data(asin)
            if data:
                self.save_to_s3(asin, data)

# 使用示例
collector = DataCollector(
    api_key="your_pangolinfo_api_key",
    s3_bucket="your-data-bucket"
)

# 定时任务:每小时执行一次
asins_to_monitor = ["B08N5WRWNW", "B08N5M7S6K", "B08L5VFJ2R"]
collector.batch_collect(asins_to_monitor)

import requests
import json
from datetime import datetime, timedelta
import sqlite3
from typing import Dict, List, Optional

class HybridDataManager:
    """混合模式数据管理器:实时采集 + 选择性存储"""
    
    def __init__(self, api_key: str, db_path: str = "amazon_data.db"):
        self.api_key = api_key
        self.base_url = "https://api.pangolinfo.com"
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """初始化数据库表结构"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 价格历史表(温数据)
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS price_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asin TEXT NOT NULL,
                timestamp DATETIME NOT NULL,
                price REAL,
                currency TEXT,
                availability TEXT,
                INDEX idx_asin_time (asin, timestamp)
            )
        ''')
        
        # BSR历史表(温数据)
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS bsr_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asin TEXT NOT NULL,
                timestamp DATETIME NOT NULL,
                bsr INTEGER,
                category TEXT,
                INDEX idx_asin_time (asin, timestamp)
            )
        ''')
        
        # 日聚合表(冷数据)
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS daily_summary (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asin TEXT NOT NULL,
                date DATE NOT NULL,
                price_min REAL,
                price_max REAL,
                price_avg REAL,
                bsr_min INTEGER,
                bsr_max INTEGER,
                bsr_avg INTEGER,
                reviews_count INTEGER,
                UNIQUE(asin, date)
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def get_realtime_data(self, asin: str, data_type: str = "product") -> Optional[Dict]:
        """获取实时数据(热数据):不存储,直接返回"""
        try:
            response = requests.get(
                f"{self.base_url}/scrape",
                params={
                    "asin": asin,
                    "type": data_type,
                    "include": "price,bsr,reviews,inventory,buybox"
                },
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            print(f"✓ 实时数据获取成功: {asin}")
            return data
            
        except requests.exceptions.RequestException as e:
            print(f"✗ API调用失败: {str(e)}")
            return None
    
    def collect_and_store(self, asin: str):
        """采集并存储温数据"""
        data = self.get_realtime_data(asin)
        if not data:
            return
        
        timestamp = datetime.now()
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 存储价格数据
        if 'price' in data:
            cursor.execute('''
                INSERT INTO price_history (asin, timestamp, price, currency, availability)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                asin,
                timestamp,
                data['price'].get('value'),
                data['price'].get('currency'),
                data.get('availability')
            ))
        
        # 存储BSR数据
        if 'bsr' in data:
            cursor.execute('''
                INSERT INTO bsr_history (asin, timestamp, bsr, category)
                VALUES (?, ?, ?, ?)
            ''', (
                asin,
                timestamp,
                data['bsr'].get('rank'),
                data['bsr'].get('category')
            ))
        
        conn.commit()
        conn.close()
        print(f"✓ 数据已存储: {asin} @ {timestamp}")
    
    def aggregate_to_daily(self, date: datetime.date):
        """将温数据聚合为冷数据"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 聚合价格和BSR数据
        cursor.execute('''
            INSERT OR REPLACE INTO daily_summary 
            (asin, date, price_min, price_max, price_avg, bsr_min, bsr_max, bsr_avg)
            SELECT 
                p.asin,
                DATE(p.timestamp) as date,
                MIN(p.price) as price_min,
                MAX(p.price) as price_max,
                AVG(p.price) as price_avg,
                MIN(b.bsr) as bsr_min,
                MAX(b.bsr) as bsr_max,
                AVG(b.bsr) as bsr_avg
            FROM price_history p
            LEFT JOIN bsr_history b ON p.asin = b.asin AND DATE(p.timestamp) = DATE(b.timestamp)
            WHERE DATE(p.timestamp) = ?
            GROUP BY p.asin, DATE(p.timestamp)
        ''', (date,))
        
        conn.commit()
        conn.close()
        print(f"✓ 日聚合完成: {date}")
    
    def cleanup_old_data(self, days_to_keep: int = 30):
        """清理过期的温数据"""
        cutoff_date = datetime.now() - timedelta(days=days_to_keep)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('DELETE FROM price_history WHERE timestamp < ?', (cutoff_date,))
        cursor.execute('DELETE FROM bsr_history WHERE timestamp < ?', (cutoff_date,))
        
        deleted = cursor.rowcount
        conn.commit()
        conn.close()
        print(f"✓ 清理完成: 删除 {deleted} 条过期数据")
    
    def get_price_trend(self, asin: str, days: int = 7) -> List[Dict]:
        """获取价格趋势(优先使用温数据,超出范围使用冷数据)"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        if days <= 30:
            # 使用温数据(明细)
            cursor.execute('''
                SELECT timestamp, price FROM price_history
                WHERE asin = ? AND timestamp >= datetime('now', ?)
                ORDER BY timestamp
            ''', (asin, f'-{days} days'))
        else:
            # 使用冷数据(聚合)
            cursor.execute('''
                SELECT date, price_avg FROM daily_summary
                WHERE asin = ? AND date >= date('now', ?)
                ORDER BY date
            ''', (asin, f'-{days} days'))
        
        results = [{"time": row[0], "price": row[1]} for row in cursor.fetchall()]
        conn.close()
        return results

# 使用示例
manager = HybridDataManager(api_key="your_pangolinfo_api_key")

# 场景1:查看当前实时数据(热数据)
current_data = manager.get_realtime_data("B08N5WRWNW")
print(f"当前价格: ${current_data['price']['value']}")

# 场景2:定时采集存储(温数据) - 配合cron每小时执行
manager.collect_and_store("B08N5WRWNW")

# 场景3:每日聚合(冷数据) - 配合cron每天凌晨执行
manager.aggregate_to_daily(datetime.now().date() - timedelta(days=1))

# 场景4:清理过期数据 - 配合cron每周执行
manager.cleanup_old_data(days_to_keep=30)

# 场景5:查询历史趋势
trend = manager.get_price_trend("B08N5WRWNW", days=7)
print(f"7天价格趋势: {trend}")