#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品价格数据同步脚本（修正清空表顺序）

功能：
1. 请求buff接口获取饰品价格数据（最低售价、最高求购价）
2. 正确处理steam_url生成
3. 使用高性能批量插入技术处理数据
4. 维护近60天的价格数据
5. 批量计算近7天和30天的平均价格
6. 在插入数据前先清空目标表

使用方法：
    python sync_price.py                    # 手动执行一次
    python sync_price.py --daemon           # 守护进程模式（每10分钟执行一次）
    python sync_price.py --install-cron     # 安装到系统定时任务（每10分钟执行）

作者: 游铛铛技术数字员工
创建时间: 2026-01-28
"""

import os
import sys
import time
import argparse
import logging
import json
import requests
import pymysql
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple
import schedule
import urllib.parse

# ============================================================
# 配置信息
# ============================================================

# BUFF API配置
BUFF_API_URL = "https://buff.163.com/api/market/enterprise/items"
BUFF_API_TOKEN = "2ce4f8fd95194fe090d53b785056de0b"

# 数据库配置（从环境变量读取）
DB_HOST = os.getenv("DB_HOST", "125.122.32.114")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "Buff123456")
DB_NAME = os.getenv("DB_NAME", "buff")

# Steam社区市场基础URL
STEAM_MARKET_BASE_URL = "https://steamcommunity.com/market/listings/730/"

# 请求配置
REQUEST_TIMEOUT = 60
BATCH_SIZE = 5000  # 大幅提高批量大小
MAX_RETRIES = 3
RETRY_DELAY = 5

# 数据保留天数
PRICE_RETENTION_DAYS = 60

# 执行频率配置（分钟）
SYNC_INTERVAL_MINUTES = 10

# 平均价格计算批次大小
AVG_PRICE_BATCH_SIZE = 10000  # 增加批次大小

# ============================================================
# 配置日志
# ============================================================

def setup_logging():
    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, "logs", f"sync_price_{datetime.now().strftime('%Y%m%d')}.log")
    os.makedirs(os.path.dirname(log_file), exist_ok=True)
    
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler(log_file, encoding='utf-8')
        ]
    )
    return logging.getLogger(__name__)

logger = setup_logging()

# ============================================================
# 数据库操作类（修正清空表逻辑）
# ============================================================

class PriceDatabase:
    """价格数据库操作类（修正清空表逻辑）"""

    def __init__(self):
        self.connection = None
        self._connect()
    
    def _connect(self):
        """建立数据库连接，优化连接参数"""
        try:
            self.connection = pymysql.connect(
                host=DB_HOST,
                port=DB_PORT,
                user=DB_USER,
                password=DB_PASSWORD,
                database=DB_NAME,
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor,
                autocommit=False,
                # 优化连接参数
                connect_timeout=10,
                read_timeout=30,
                write_timeout=30
            )
            logger.info(f"数据库连接成功: {DB_HOST}:{DB_PORT}/{DB_NAME}")
        except Exception as e:
            logger.error(f"数据库连接失败: {e}")
            raise
    
    def close(self):
        if self.connection and self.connection.open:
            self.connection.close()
            logger.info("数据库连接已关闭")
    
    def clear_table(self, table_name: str):
        """清空表（使用TRUNCATE优化）"""
        try:
            with self.connection.cursor() as cursor:
                # 使用TRUNCATE比DELETE更快
                cursor.execute(f"TRUNCATE TABLE {table_name}")
            self.connection.commit()
            logger.info(f"已清空 {table_name} 表")
        except Exception as e:
            # 如果TRUNCATE失败，回退到DELETE
            logger.warning(f"TRUNCATE失败，使用DELETE: {e}")
            try:
                with self.connection.cursor() as cursor:
                    cursor.execute(f"DELETE FROM {table_name}")
                self.connection.commit()
                logger.info(f"已清空 {table_name} 表")
            except Exception as e2:
                logger.error(f"清空表失败: {e2}")
                raise
    
    def bulk_insert(self, table_name: str, data_list: List[Dict[str, Any]], 
                    columns: List[str]) -> int:
        """高性能批量插入"""
        if not data_list:
            return 0
        
        # 构建INSERT SQL
        column_list = ', '.join([f'`{col}`' for col in columns])
        placeholders = ', '.join(['%s'] * len(columns))
        sql = f"INSERT INTO `{table_name}` ({column_list}) VALUES ({placeholders})"
        
        total_inserted = 0
        
        with self.connection.cursor() as cursor:
            # 临时优化插入性能
            cursor.execute("SET SESSION autocommit = 0")
            cursor.execute("SET SESSION unique_checks = 0")
            cursor.execute("SET SESSION foreign_key_checks = 0")
            
            try:
                for i in range(0, len(data_list), BATCH_SIZE):
                    batch = data_list[i:i + BATCH_SIZE]
                    values = []
                    
                    for item in batch:
                        row_values = []
                        for col in columns:
                            value = item.get(col)
                            if value is None:
                                row_values.append(None)
                            elif isinstance(value, datetime):
                                row_values.append(value.strftime('%Y-%m-%d %H:%M:%S'))
                            elif isinstance(value, (int, float)):
                                row_values.append(value)
                            else:
                                # 字符串类型，确保编码正确
                                row_values.append(str(value))
                        values.append(tuple(row_values))
                    
                    # 使用executemany批量插入
                    cursor.executemany(sql, values)
                    total_inserted += len(batch)
                    
                    # 每插入3批提交一次，平衡性能和内存使用
                    if i > 0 and i % (BATCH_SIZE * 3) == 0:
                        self.connection.commit()
                        logger.info(f"批量插入进度: {total_inserted}/{len(data_list)}")
                
                # 最终提交
                self.connection.commit()
                
            finally:
                # 恢复会话设置
                cursor.execute("SET SESSION autocommit = 1")
                cursor.execute("SET SESSION unique_checks = 1")
                cursor.execute("SET SESSION foreign_key_checks = 1")
        
        return total_inserted
    
    def insert_daily_price_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到buff_goods_daily_price_middle表"""
        if not data_list:
            return 0
        
        # 定义列和对应的数据字段
        columns = ['goods_id', 'name', 'min_price', 'buy_max_price', 'stat_time', 'create_time']
        
        # 准备数据
        now = datetime.now()
        processed_data = []
        
        logger.info(f"开始准备 {len(data_list)} 条无tag价格数据...")
        start_time = time.time()
        
        for item in data_list:
            processed_data.append({
                'goods_id': str(item['goods_id']),
                'name': item['name'],
                'min_price': float(item['min_price']) if item['min_price'] else 0.00,
                'buy_max_price': float(item['buy_max_price']) if item['buy_max_price'] else 0.00,
                'stat_time': datetime.fromtimestamp(item['stat_time']),
                'create_time': now
            })
        
        prep_time = time.time() - start_time
        logger.info(f"数据准备完成，耗时: {prep_time:.2f}秒")
        logger.info(f"开始插入 {len(processed_data)} 条数据到buff_goods_daily_price_middle...")
        
        insert_start = time.time()
        result = self.bulk_insert('buff_goods_daily_price_middle', processed_data, columns)
        insert_time = time.time() - insert_start
        
        logger.info(f"插入完成: {result} 条，耗时: {insert_time:.2f}秒")
        return result
    
    def copy_from_middle_to_daily_price(self) -> int:
        """将中间表数据插入到buff_goods_daily_price表"""
        try:
            with self.connection.cursor() as cursor:
                # 使用INSERT IGNORE避免重复（如果有唯一索引冲突）
                sql = """
                INSERT IGNORE INTO buff_goods_daily_price 
                (goods_id, name, min_price, buy_max_price, stat_time, create_time, update_time)
                SELECT goods_id, name, min_price, buy_max_price, stat_time, NOW(), NOW()
                FROM buff_goods_daily_price_middle
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            
            self.connection.commit()
            return affected_rows
        except Exception as e:
            logger.error(f"复制数据失败: {e}")
            self.connection.rollback()
            return 0
    
    def cleanup_old_price_data(self):
        """清理超过60天的价格数据（分批删除，避免锁表）"""
        cutoff_date = datetime.now() - timedelta(days=PRICE_RETENTION_DAYS)
        
        try:
            with self.connection.cursor() as cursor:
                # 分批删除，每次10000条
                total_deleted = 0
                while True:
                    sql = """
                    DELETE FROM buff_goods_daily_price 
                    WHERE create_time < %s
                    LIMIT 10000
                    """
                    cursor.execute(sql, (cutoff_date,))
                    batch_deleted = cursor.connection.affected_rows()
                    total_deleted += batch_deleted
                    
                    if batch_deleted == 0:
                        break
                    
                    self.connection.commit()
                    time.sleep(0.05)  # 短暂休眠，避免锁表
                
                logger.info(f"已清理 {total_deleted} 条超过{PRICE_RETENTION_DAYS}天的旧价格数据")
        except Exception as e:
            logger.error(f"清理旧数据失败: {e}")
            self.connection.rollback()
    
    def insert_goods_info_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到buff_goods_info_middle表"""
        if not data_list:
            return 0
        
        # 定义列和对应的数据字段
        columns = ['goods_id', 'name', 'min_price', 'buy_max_price', 
                  'd7_avg_min_price', 'd30_avg_min_price', 
                  'buff_url', 'steam_url', 'stat_time', 'create_time']
        
        # 准备数据
        now = datetime.now()
        processed_data = []
        
        logger.info(f"开始准备 {len(data_list)} 条商品信息数据...")
        start_time = time.time()
        
        for item in data_list:
            processed_data.append({
                'goods_id': str(item['goods_id']),
                'name': item['name'],
                'min_price': float(item['min_price']) if item['min_price'] else 0.00,
                'buy_max_price': float(item['buy_max_price']) if item['buy_max_price'] else 0.00,
                'd7_avg_min_price': float(item['avg_7d_min_price']) if item['avg_7d_min_price'] else 0.00,
                'd30_avg_min_price': float(item['avg_30d_min_price']) if item['avg_30d_min_price'] else 0.00,
                'buff_url': item['buff_url'],
                'steam_url': item['steam_url'],
                'stat_time': datetime.fromtimestamp(item['stat_time']),
                'create_time': now
            })
        
        prep_time = time.time() - start_time
        logger.info(f"商品信息数据准备完成，耗时: {prep_time:.2f}秒")
        logger.info(f"开始插入 {len(processed_data)} 条数据到buff_goods_info_middle...")
        
        insert_start = time.time()
        result = self.bulk_insert('buff_goods_info_middle', processed_data, columns)
        insert_time = time.time() - insert_start
        
        logger.info(f"插入完成: {result} 条，耗时: {insert_time:.2f}秒")
        return result
    
    def copy_from_middle_to_goods_info(self) -> int:
        """将中间表数据插入到buff_goods_info表（先清空再插入）"""
        try:
            # 先清空目标表
            self.clear_table("buff_goods_info")
            
            with self.connection.cursor() as cursor:
                # 使用INSERT INTO（因为已经清空了，不需要REPLACE）
                sql = """
                INSERT INTO buff_goods_info 
                (goods_id, name, min_price, buy_max_price, d7_avg_min_price, d30_avg_min_price, 
                 buff_url, steam_url, stat_time, create_time, update_time)
                SELECT goods_id, name, min_price, buy_max_price, d7_avg_min_price, d30_avg_min_price, 
                       buff_url, steam_url, stat_time, NOW(), NOW()
                FROM buff_goods_info_middle
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            
            self.connection.commit()
            return affected_rows
        except Exception as e:
            logger.error(f"复制商品信息失败: {e}")
            self.connection.rollback()
            return 0
    
    def insert_daily_tag_price_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到buff_goods_daily_tag_price_middle表"""
        if not data_list:
            return 0
        
        # 定义列和对应的数据字段
        columns = ['goods_id', 'name', 'min_price', 'buy_max_price', 
                  'tag_id', 'tag_name', 'stat_time', 'create_time']
        
        # 准备数据
        now = datetime.now()
        processed_data = []
        
        logger.info(f"开始准备 {len(data_list)} 条有tag价格数据...")
        start_time = time.time()
        
        for item in data_list:
            processed_data.append({
                'goods_id': str(item['goods_id']),
                'name': item['name'],
                'min_price': float(item['min_price']) if item['min_price'] else 0.00,
                'buy_max_price': float(item['buy_max_price']) if item['buy_max_price'] else 0.00,
                'tag_id': str(item['tag_id']) if item['tag_id'] else None,
                'tag_name': item['tag_name'],
                'stat_time': datetime.fromtimestamp(item['stat_time']),
                'create_time': now
            })
        
        prep_time = time.time() - start_time
        logger.info(f"有tag价格数据准备完成，耗时: {prep_time:.2f}秒")
        logger.info(f"开始插入 {len(processed_data)} 条数据到buff_goods_daily_tag_price_middle...")
        
        insert_start = time.time()
        result = self.bulk_insert('buff_goods_daily_tag_price_middle', processed_data, columns)
        insert_time = time.time() - insert_start
        
        logger.info(f"插入完成: {result} 条，耗时: {insert_time:.2f}秒")
        return result
    
    def copy_from_middle_to_daily_tag_price(self) -> int:
        """将中间表数据插入到buff_goods_daily_tag_price表（先清空再插入）"""
        try:
            # 先清空目标表
            self.clear_table("buff_goods_daily_tag_price")
            
            with self.connection.cursor() as cursor:
                # 使用INSERT INTO（因为已经清空了，不需要INSERT IGNORE）
                sql = """
                INSERT INTO buff_goods_daily_tag_price 
                (goods_id, name, min_price, buy_max_price, tag_id, tag_name, stat_time, create_time, update_time)
                SELECT goods_id, name, min_price, buy_max_price, tag_id, tag_name, stat_time, NOW(), NOW()
                FROM buff_goods_daily_tag_price_middle
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            
            self.connection.commit()
            return affected_rows
        except Exception as e:
            logger.error(f"复制有tag价格失败: {e}")
            self.connection.rollback()
            return 0
    
    def calculate_average_prices_batch(self, goods_ids: List[str]) -> Dict[str, Tuple[float, float]]:
        """批量计算平均价格，优化性能"""
        if not goods_ids:
            return {}
        
        results = {}
        
        try:
            with self.connection.cursor() as cursor:
                # 创建临时表优化查询
                cursor.execute("""
                CREATE TEMPORARY TABLE IF NOT EXISTS temp_goods_ids (
                    goods_id VARCHAR(64) PRIMARY KEY
                ) ENGINE=MEMORY
                """)
                
                # 批量插入goods_id到临时表
                for i in range(0, len(goods_ids), 10000):
                    batch = goods_ids[i:i + 10000]
                    # 使用VALUES列表批量插入
                    values_placeholders = ','.join(['(%s)'] * len(batch))
                    cursor.execute(f"INSERT IGNORE INTO temp_goods_ids (goods_id) VALUES {values_placeholders}", batch)
                
                # 使用临时表进行JOIN查询，大幅提高性能
                cursor.execute("""
                SELECT 
                    t.goods_id,
                    COALESCE(ROUND(AVG(CASE 
                        WHEN dp.create_time >= DATE_SUB(NOW(), INTERVAL 7 DAY) 
                        AND dp.min_price > 0 
                        THEN dp.min_price 
                    END), 2), 0.00) as avg_7d,
                    COALESCE(ROUND(AVG(CASE 
                        WHEN dp.create_time >= DATE_SUB(NOW(), INTERVAL 30 DAY) 
                        AND dp.min_price > 0 
                        THEN dp.min_price 
                    END), 2), 0.00) as avg_30d
                FROM temp_goods_ids t
                LEFT JOIN buff_goods_daily_price dp ON t.goods_id = dp.goods_id
                WHERE dp.create_time >= DATE_SUB(NOW(), INTERVAL 30 DAY)
                GROUP BY t.goods_id
                """)
                
                rows = cursor.fetchall()
                for row in rows:
                    results[row['goods_id']] = (
                        float(row['avg_7d']),
                        float(row['avg_30d'])
                    )
                
                # 清理临时表
                cursor.execute("DROP TEMPORARY TABLE IF EXISTS temp_goods_ids")
                
        except Exception as e:
            logger.error(f"批量计算平均价格失败: {e}")
            # 回退到原方法
            return self._fallback_calculate_avg_prices(goods_ids)
        
        return results
    
    def _fallback_calculate_avg_prices(self, goods_ids: List[str]) -> Dict[str, Tuple[float, float]]:
        """后备的平均价格计算方法"""
        results = {}
        
        try:
            with self.connection.cursor() as cursor:
                for i in range(0, len(goods_ids), AVG_PRICE_BATCH_SIZE):
                    batch = goods_ids[i:i + AVG_PRICE_BATCH_SIZE]
                    placeholders = ','.join(['%s'] * len(batch))
                    
                    cursor.execute(f"""
                    SELECT 
                        goods_id,
                        COALESCE(ROUND(AVG(CASE 
                            WHEN create_time >= DATE_SUB(NOW(), INTERVAL 7 DAY) 
                            AND min_price > 0 
                            THEN min_price 
                        END), 2), 0.00) as avg_7d,
                        COALESCE(ROUND(AVG(CASE 
                            WHEN create_time >= DATE_SUB(NOW(), INTERVAL 30 DAY) 
                            AND min_price > 0 
                            THEN min_price 
                        END), 2), 0.00) as avg_30d
                    FROM buff_goods_daily_price 
                    WHERE goods_id IN ({placeholders})
                    AND create_time >= DATE_SUB(NOW(), INTERVAL 30 DAY)
                    GROUP BY goods_id
                    """, batch)
                    
                    rows = cursor.fetchall()
                    for row in rows:
                        results[row['goods_id']] = (
                            float(row['avg_7d']),
                            float(row['avg_30d'])
                        )
                    
                    if i % (AVG_PRICE_BATCH_SIZE * 5) == 0:
                        logger.info(f"平均价格计算进度: {len(results)}/{len(goods_ids)}")
                
        except Exception as e:
            logger.error(f"后备平均价格计算失败: {e}")
        
        return results
    
    def get_last_sync_time(self) -> Optional[datetime]:
        """获取上次同步时间"""
        try:
            with self.connection.cursor() as cursor:
                cursor.execute("""
                SELECT MAX(stat_time) as last_sync
                FROM buff_goods_daily_price 
                WHERE stat_time > DATE_SUB(NOW(), INTERVAL 1 HOUR)
                LIMIT 1
                """)
                result = cursor.fetchone()
                return result['last_sync'] if result['last_sync'] else None
        except Exception as e:
            logger.error(f"获取上次同步时间失败: {e}")
            return None


# ============================================================
# BUFF API 操作类（保持不变）
# ============================================================

class BuffPriceAPI:
    """BUFF价格API操作类"""
    
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {BUFF_API_TOKEN}",
            "Accept": "application/json",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })
    
    def fetch_price_data(self, game: str = "cs2", include_sticker: int = 1) -> Dict[str, Any]:
        """获取价格数据"""
        params = {"game": game, "include_sticker": include_sticker}
        
        for retry in range(MAX_RETRIES):
            try:
                logger.info(f"请求BUFF API (第{retry + 1}次尝试)...")
                response = self.session.get(BUFF_API_URL, params=params, timeout=REQUEST_TIMEOUT)
                response.raise_for_status()
                
                data = response.json()

                logger.info("=" * 60)
                logger.info("🔍 API返回数据结构:")
                logger.info(f"  顶层keys: {list(data.keys())}")
                
                # 获取数据统计信息
                stat_time = data.get("stat_time", 0)
                remaining = data.get("req_remaining", 0)
                info_data = data.get("info", [])
                
                logger.info(f"  统计时间: {datetime.fromtimestamp(stat_time)}")
                logger.info(f"  剩余调用次数: {remaining}")
                logger.info(f"  数据条数: {len(info_data)}")
                
                # 检查第一条数据是否有market_hash_name字段
                if info_data and len(info_data) > 0:
                    first_item = info_data[0]
                    logger.info(f"  第一条数据字段检查:")
                    logger.info(f"    goods_id: {first_item.get('goods_id')}")
                    logger.info(f"    name: {first_item.get('name')}")
                    logger.info(f"    market_hash_name: {first_item.get('market_hash_name', '不存在')}")
                
                logger.info("=" * 60)

                # 验证数据格式
                if "info" not in data:
                    logger.error(f"API返回格式错误: 无info字段, 完整响应: {data}")
                    raise Exception(f"API错误: {data.get('msg', '未知错误')}")
                
                logger.info(f"API请求成功，获取到 {len(info_data)} 条记录")
                logger.info(f"API剩余调用次数: {remaining}")
                
                return data
                    
            except requests.exceptions.RequestException as e:
                logger.error(f"API请求失败 (尝试 {retry + 1}/{MAX_RETRIES}): {e}")
                if 'response' in locals():
                    logger.error(f"响应状态码: {response.status_code}")
                    logger.error(f"响应内容: {response.text[:500]}")
                if retry < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY)
                else:
                    raise
    
    def generate_steam_url(self, market_hash_name: str) -> str:
        """生成Steam社区市场链接"""
        if not market_hash_name:
            return None
        
        try:
            # URL编码market_hash_name
            encoded_name = urllib.parse.quote(market_hash_name)
            steam_url = f"{STEAM_MARKET_BASE_URL}{encoded_name}"
            return steam_url
        except Exception as e:
            logger.error(f"生成Steam URL失败: {market_hash_name}, 错误: {e}")
            return None
    
    def parse_price_data(self, api_response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
        """解析价格数据，返回三种类型的数据列表
        
        Returns:
            tuple: (daily_price_list, goods_info_list, tag_price_list)
        """
        logger.info("🔍 开始解析数据...")
        
        stat_time = api_response.get("stat_time", 0)
        info_data = api_response.get("info", [])
        
        logger.info(f"  stat_time: {datetime.fromtimestamp(stat_time)}")
        logger.info(f"  info列表长度: {len(info_data)}")
        
        daily_price_list = []  # 无tag的价格数据
        tag_price_list = []    # 有tag的价格数据
        goods_info_list = []   # 商品信息数据
        
        start_time = time.time()
        
        for item in info_data:
            goods_id = item.get("goods_id")
            name = item.get("name")
            market_hash_name = item.get("market_hash_name")  # 获取market_hash_name
            update_time = item.get("update_time", stat_time)
            sales = item.get("sales", [])
            
            # 生成链接
            buff_url = f"https://buff.163.com/goods/{goods_id}"
            steam_url = self.generate_steam_url(market_hash_name)
            
            # 处理每个sales条目
            has_base_price = False
            for sale in sales:
                min_price = sale.get("min_price")
                buy_max_price = sale.get("buy_max_price")
                tag_id = sale.get("tag_id")
                tag_name = sale.get("tag_name")
                
                if tag_id and tag_name:
                    # 有tag的数据
                    tag_data = {
                        "goods_id": goods_id,
                        "name": name,
                        "min_price": min_price,
                        "buy_max_price": buy_max_price,
                        "tag_id": tag_id,
                        "tag_name": tag_name,
                        "stat_time": stat_time,
                        "update_time": update_time
                    }
                    tag_price_list.append(tag_data)
                else:
                    # 无tag的数据
                    price_data = {
                        "goods_id": goods_id,
                        "name": name,
                        "min_price": min_price,
                        "buy_max_price": buy_max_price,
                        "stat_time": stat_time,
                        "update_time": update_time
                    }
                    daily_price_list.append(price_data)
                    
                    # 如果是第一个无tag的条目，也用于商品信息
                    if not has_base_price:
                        has_base_price = True
                        goods_info_data = {
                            "goods_id": goods_id,
                            "name": name,
                            "min_price": min_price,
                            "buy_max_price": buy_max_price,
                            "avg_7d_min_price": None,
                            "avg_30d_min_price": None,
                            "stat_time": stat_time,
                            "update_time": update_time,
                            "buff_url": buff_url,
                            "steam_url": steam_url  # 使用生成的steam_url
                        }
                        goods_info_list.append(goods_info_data)
        
        elapsed = time.time() - start_time
        logger.info(f"  解析完成，耗时: {elapsed:.2f}秒")
        logger.info(f"    无tag价格数据: {len(daily_price_list)} 条")
        logger.info(f"    有tag价格数据: {len(tag_price_list)} 条")
        logger.info(f"    商品信息数据: {len(goods_info_list)} 条")
        
        # 统计steam_url生成情况
        steam_url_count = sum(1 for item in goods_info_list if item.get('steam_url'))
        logger.info(f"    成功生成steam_url: {steam_url_count}/{len(goods_info_list)} 条")
        
        return daily_price_list, goods_info_list, tag_price_list


# ============================================================
# 主逻辑（修正清空表顺序）
# ============================================================

class PriceSyncTask:
    """价格数据同步任务"""
    
    def __init__(self):
        self.db = None
        self.api = BuffPriceAPI()
        self.last_sync_time = None
    
    def _init_db(self):
        """初始化数据库连接"""
        if self.db is None:
            self.db = PriceDatabase()
    
    def run(self) -> bool:
        """执行同步任务"""
        start_time = datetime.now()
        logger.info("=" * 60)
        logger.info(f"开始执行价格数据同步任务: {start_time}")
        logger.info("=" * 60)
        
        try:
            # 初始化数据库连接
            self._init_db()
            
            # 检查是否需要执行
            last_sync = self.db.get_last_sync_time()
            if last_sync:
                time_diff = (start_time - last_sync).total_seconds()
                if time_diff < 300:  # 5分钟内已同步过
                    logger.info(f"最近 {time_diff:.0f} 秒内已同步过，跳过本次执行")
                    return True
            
            # Step 1: 清空中间表
            logger.info("[Step 1/9] 清空中间表...")
            clear_start = time.time()
            self.db.clear_table("buff_goods_daily_price_middle")
            self.db.clear_table("buff_goods_info_middle")
            self.db.clear_table("buff_goods_daily_tag_price_middle")
            logger.info(f"清空中间表耗时: {time.time() - clear_start:.2f}秒")
            
            # Step 2: 调用BUFF API获取数据
            logger.info("[Step 2/9] 调用BUFF API获取价格数据...")
            api_start = time.time()
            api_response = self.api.fetch_price_data(game="cs2", include_sticker=1)
            logger.info(f"API调用耗时: {time.time() - api_start:.2f}秒")
            
            # Step 3: 解析数据
            logger.info("[Step 3/9] 解析数据...")
            parse_start = time.time()
            daily_price_list, goods_info_list, tag_price_list = self.api.parse_price_data(api_response)
            logger.info(f"数据解析耗时: {time.time() - parse_start:.2f}秒")
            
            if not daily_price_list:
                logger.warning("未获取到任何价格数据，任务取消")
                return False
            
            # Step 4: 插入无tag价格数据到中间表
            logger.info("[Step 4/9] 插入无tag价格数据到中间表...")
            inserted_daily_count = self.db.insert_daily_price_middle(daily_price_list)
            
            # Step 5: 更新主价格表
            logger.info("[Step 5/9] 更新主价格表...")
            copy_start = time.time()
            target_inserted = self.db.copy_from_middle_to_daily_price()
            logger.info(f"主价格表插入完成: {target_inserted} 条，耗时: {time.time() - copy_start:.2f}秒")
            
            # 每小时清理一次旧数据
            current_minute = datetime.now().minute
            if current_minute == 0:
                logger.info("[Step 5a/9] 清理旧价格数据...")
                cleanup_start = time.time()
                self.db.cleanup_old_price_data()
                logger.info(f"清理旧数据耗时: {time.time() - cleanup_start:.2f}秒")
            else:
                logger.info("[Step 5a/9] 跳过清理（非整点）")
            
            # Step 6: 批量计算平均价格
            logger.info("[Step 6/9] 批量计算平均价格...")
            avg_start = time.time()
            goods_ids = [str(item['goods_id']) for item in goods_info_list]
            avg_prices = self.db.calculate_average_prices_batch(goods_ids)
            
            # 更新goods_info_list中的平均价格
            updated_count = 0
            for goods_info in goods_info_list:
                goods_id = str(goods_info['goods_id'])
                if goods_id in avg_prices:
                    avg_7d, avg_30d = avg_prices[goods_id]
                    goods_info['avg_7d_min_price'] = avg_7d
                    goods_info['avg_30d_min_price'] = avg_30d
                    updated_count += 1
            
            logger.info(f"平均价格计算完成: {updated_count}/{len(goods_info_list)} 条，耗时: {time.time() - avg_start:.2f}秒")
            
            # Step 7: 插入商品信息到中间表
            logger.info("[Step 7/9] 插入商品信息到中间表...")
            inserted_info_count = self.db.insert_goods_info_middle(goods_info_list)
            
            # Step 7a: 清空目标表并更新商品信息主表
            logger.info("[Step 7a/9] 清空buff_goods_info表并更新数据...")
            copy_info_start = time.time()
            info_inserted = self.db.copy_from_middle_to_goods_info()
            logger.info(f"商品信息主表插入完成: {info_inserted} 条，耗时: {time.time() - copy_info_start:.2f}秒")
            
            # Step 8: 插入有tag价格数据到中间表
            if tag_price_list:
                logger.info("[Step 8/9] 插入有tag价格数据到中间表...")
                inserted_tag_count = self.db.insert_daily_tag_price_middle(tag_price_list)
                
                # Step 9: 清空目标表并更新有tag价格主表
                logger.info("[Step 9/9] 清空buff_goods_daily_tag_price表并更新数据...")
                copy_tag_start = time.time()
                tag_inserted = self.db.copy_from_middle_to_daily_tag_price()
                logger.info(f"有tag价格主表插入完成: {tag_inserted} 条，耗时: {time.time() - copy_tag_start:.2f}秒")
            else:
                logger.info("没有有tag的价格数据，跳过步骤8-9")
                inserted_tag_count = 0
            
            # 清理中间表
            logger.info("[清理] 清空中间表数据...")
            cleanup_mid_start = time.time()
            self.db.clear_table("buff_goods_daily_price_middle")
            self.db.clear_table("buff_goods_info_middle")
            self.db.clear_table("buff_goods_daily_tag_price_middle")
            logger.info(f"清理中间表耗时: {time.time() - cleanup_mid_start:.2f}秒")
            
            # 任务完成
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()
            logger.info("=" * 60)
            logger.info(f"价格数据同步任务完成! 总耗时: {duration:.2f}秒")
            logger.info(f"同步数据量:")
            logger.info(f"  无tag价格: {inserted_daily_count} 条")
            logger.info(f"  商品信息: {inserted_info_count} 条")
            logger.info(f"  有tag价格: {inserted_tag_count} 条")
            logger.info(f"  总处理数据: {len(daily_price_list)} 条饰品记录")
            logger.info("=" * 60)
            
            return True
            
        except Exception as e:
            logger.exception(f"同步任务失败: {e}")
            return False
        finally:
            if self.db:
                self.db.close()
                self.db = None


def run_once() -> bool:
    """执行一次同步任务"""
    task = PriceSyncTask()
    return task.run()


def run_daemon():
    """守护进程模式，持续运行并定时执行任务"""
    logger.info(f"启动守护进程模式... 定时任务: 每{SYNC_INTERVAL_MINUTES}分钟执行一次")
    
    # 立即执行一次
    run_once()
    
    # 设置定时任务：每10分钟执行一次
    schedule.every(SYNC_INTERVAL_MINUTES).minutes.do(run_once)
    
    logger.info(f"定时任务已设置: 每{SYNC_INTERVAL_MINUTES}分钟执行一次")
    logger.info(f"下次执行时间: {schedule.next_run()}")
    
    while True:
        try:
            schedule.run_pending()
            time.sleep(1)
        except KeyboardInterrupt:
            logger.info("收到中断信号，停止守护进程")
            break
        except Exception as e:
            logger.error(f"定时任务执行异常: {e}")
            time.sleep(60)


def install_cron_job():
    """安装到系统定时任务（每10分钟执行）"""
    script_path = os.path.abspath(__file__)
    
    cron_expression = f"*/{SYNC_INTERVAL_MINUTES} * * * * /usr/bin/python3 {script_path}\n"
    
    try:
        import subprocess
        
        result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
        current_crontab = result.stdout
        
        if script_path in current_crontab:
            logger.info("定时任务已存在，更新为每10分钟执行")
            lines = current_crontab.split('\n')
            new_lines = [line for line in lines if script_path not in line]
            new_crontab = '\n'.join(new_lines) + '\n' + cron_expression
        else:
            new_crontab = current_crontab + cron_expression
        
        subprocess.run(["crontab", "-"], input=new_crontab, text=True)
        logger.info(f"定时任务安装成功: {cron_expression.strip()}")
        
    except subprocess.CalledProcessError as e:
        if "no crontab" in str(e.stderr).lower():
            new_crontab = cron_expression
            subprocess.run(["crontab", "-"], input=new_crontab, text=True)
            logger.info(f"定时任务安装成功: {cron_expression.strip()}")
        else:
            logger.error(f"安装定时任务失败: {e}")
            sys.exit(1)
    except Exception as e:
        logger.error(f"安装定时任务失败: {e}")
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="BUFF饰品价格数据同步脚本")
    parser.add_argument("--daemon", action="store_true", help="守护进程模式（每10分钟执行一次）")
    parser.add_argument("--install-cron", action="store_true", help="安装到系统定时任务（每10分钟执行）")
    
    args = parser.parse_args()
    
    if args.install_cron:
        install_cron_job()
    elif args.daemon:
        run_daemon()
    else:
        success = run_once()
        sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()