buff3.0/buff-api/sync_goods/sync_goods.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品价格数据同步脚本（优化版：累加存储 + 实时计算平均价）

功能：
1. 请求buff接口获取饰品价格数据（最低售价、最高求购价）
2. 正确处理steam_url生成
3. 使用高性能批量插入技术处理数据
4. 维护近50天的价格数据（每饰品每天1条，累加sum/cnt）
5. 每次同步完成后直接计算7天和30天的平均价格并同步到 goods_info 表

使用方法：
    python sync_goods.py                    # 执行一次同步（周期调度请使用 run_buff_sync_loop.sh）

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

import os
import sys
import time
import logging
import json
import requests
import pymysql

from sync_db_errors import is_database_connection_error
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple
import urllib.parse

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

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

# 数据库配置（从环境变量读取）
DB_HOST = os.getenv("DB_HOST", "47.99.164.66")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "hn02le.34lkdLKD")
DB_NAME = os.getenv("DB_NAME", "buff")
# 大批量 INSERT/SELECT 可能远超 30s；过短会触发 2013 Lost connection（客户端读超时）
DB_READ_TIMEOUT = int(os.getenv("DB_READ_TIMEOUT", "1800"))
DB_WRITE_TIMEOUT = int(os.getenv("DB_WRITE_TIMEOUT", "1800"))
DB_BULK_INSERT_RETRIES = int(os.getenv("DB_BULK_INSERT_RETRIES", "3"))

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

# 请求配置
REQUEST_TIMEOUT = 60
BATCH_SIZE = int(os.getenv("BULK_INSERT_BATCH_SIZE", "5000"))
MAX_RETRIES = 3
RETRY_DELAY = 5

# 数据保留天数
PRICE_RETENTION_DAYS = 50

# 平均价格计算批次大小
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=DB_READ_TIMEOUT,
                write_timeout=DB_WRITE_TIMEOUT,
            )
            logger.info(
                f"数据库连接成功: {DB_HOST}:{DB_PORT}/{DB_NAME} "
                f"(read_timeout={DB_READ_TIMEOUT}s, write_timeout={DB_WRITE_TIMEOUT}s)"
            )
        except Exception as e:
            logger.error(f"数据库连接失败: {e}")
            raise

    def _reconnect(self):
        try:
            if self.connection:
                self.connection.close()
        except Exception:
            pass
        self.connection = None
        self._connect()

    @staticmethod
    def _is_transient_db_error(exc: BaseException) -> bool:
        if not isinstance(exc, pymysql.err.OperationalError):
            return False
        code = exc.args[0] if exc.args else 0
        return code in (2006, 2013, 2014, 1053, 2003)
    
    def close(self):
        if self.connection and self.connection.open:
            self.connection.close()
            logger.info("数据库连接已关闭")

    def _safe_rollback(self) -> None:
        """连接已断开时 rollback 会抛 InterfaceError，避免二次异常掩盖原始错误。"""
        try:
            if self.connection and getattr(self.connection, "open", False):
                self.connection.rollback()
        except Exception as ex:
            logger.warning(f"回滚跳过（连接可能已断开）: {ex}")
    
    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_one_batch(
        self,
        cursor,
        sql: str,
        batch: List[Dict[str, Any]],
        columns: List[str],
    ) -> None:
        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))
        cursor.executemany(sql, values)

    def _safe_restore_session_defaults(self, cursor) -> None:
        if not self.connection:
            return
        try:
            if getattr(self.connection, "open", False):
                cursor.execute("SET SESSION autocommit = 1")
                cursor.execute("SET SESSION unique_checks = 1")
                cursor.execute("SET SESSION foreign_key_checks = 1")
        except Exception as e:
            logger.warning(f"恢复会话设置失败（连接可能已断开），可忽略: {e}")

    def bulk_insert(self, table_name: str, data_list: List[Dict[str, Any]],
                    columns: List[str]) -> int:
        """高性能批量插入；断连类错误自动重连并重试当前批。"""
        if not data_list:
            return 0

        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
        cursor = self.connection.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]
                attempt = 0
                while True:
                    try:
                        self._bulk_insert_one_batch(cursor, sql, batch, columns)
                        break
                    except pymysql.err.OperationalError as e:
                        attempt += 1
                        if not self._is_transient_db_error(e) or attempt > DB_BULK_INSERT_RETRIES:
                            raise
                        logger.warning(
                            "批量插入单批失败 (%s/%s)，将重连后重试当前批: %s",
                            attempt,
                            DB_BULK_INSERT_RETRIES,
                            e,
                        )
                        time.sleep(RETRY_DELAY * attempt)
                        self._reconnect()
                        cursor = self.connection.cursor()
                        cursor.execute("SET SESSION autocommit = 0")
                        cursor.execute("SET SESSION unique_checks = 0")
                        cursor.execute("SET SESSION foreign_key_checks = 0")

                total_inserted += len(batch)
                self.connection.commit()

                if i > 0 and i % (BATCH_SIZE * 3) == 0:
                    logger.info(f"批量插入进度: {total_inserted}/{len(data_list)}")
        finally:
            self._safe_restore_session_defaults(cursor)
            try:
                cursor.close()
            except Exception:
                pass

        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 表（按日期聚合）
        
        逻辑：
        1. 按 goods_id 和日期分组，累加 sum 和 cnt
        2. 使用 ON DUPLICATE KEY UPDATE 实现累加
        3. 同时计算当天的平均价 min_price_avg
        """
        try:
            with self.connection.cursor() as cursor:
                sql = """
                INSERT INTO buff_goods_daily_price 
                (goods_id, stat_date, name, min_price, min_price_sum, min_price_cnt, 
                 buy_max_price, buy_max_price_sum, buy_max_price_cnt,
                 min_price_avg, buy_max_price_avg,
                 create_time, update_time)
                SELECT 
                    goods_id,
                    DATE(stat_time) as stat_date,
                    MAX(name) as name,
                    MIN(min_price) as min_price,
                    SUM(min_price) as min_price_sum,
                    COUNT(*) as min_price_cnt,
                    MAX(buy_max_price) as buy_max_price,
                    SUM(COALESCE(buy_max_price, 0)) as buy_max_price_sum,
                    SUM(CASE WHEN buy_max_price > 0 THEN 1 ELSE 0 END) as buy_max_price_cnt,
                    ROUND(SUM(min_price) / COUNT(*), 2) as min_price_avg,
                    ROUND(SUM(COALESCE(buy_max_price, 0)) / NULLIF(SUM(CASE WHEN buy_max_price > 0 THEN 1 ELSE 0 END), 0), 2) as buy_max_price_avg,
                    NOW() as create_time,
                    NOW() as update_time
                FROM buff_goods_daily_price_middle
                GROUP BY goods_id, DATE(stat_time)
                ON DUPLICATE KEY UPDATE
                    name = VALUES(name),
                    min_price = LEAST(COALESCE(min_price, 999999999), VALUES(min_price)),
                    min_price_sum = min_price_sum + VALUES(min_price_sum),
                    min_price_cnt = min_price_cnt + VALUES(min_price_cnt),
                    min_price_avg = ROUND((min_price_sum + VALUES(min_price_sum)) / (min_price_cnt + VALUES(min_price_cnt)), 2),
                    buy_max_price = GREATEST(COALESCE(buy_max_price, 0), VALUES(buy_max_price)),
                    buy_max_price_sum = buy_max_price_sum + VALUES(buy_max_price_sum),
                    buy_max_price_cnt = buy_max_price_cnt + VALUES(buy_max_price_cnt),
                    buy_max_price_avg = ROUND((buy_max_price_sum + VALUES(buy_max_price_sum)) / NULLIF((buy_max_price_cnt + VALUES(buy_max_price_cnt)), 0), 2),
                    update_time = NOW()
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            
            self.connection.commit()
            logger.info(f"累加到主价格表完成，影响行数: {affected_rows}")
            return affected_rows
        except Exception as e:
            logger.error(f"累加数据失败: {e}")
            self._safe_rollback()
            return 0
    
    def cleanup_old_price_data(self):
        """清理超过50天的价格数据（分批删除，避免锁表）"""
        cutoff_date = datetime.now() - timedelta(days=50)
        
        try:
            with self.connection.cursor() as cursor:
                total_deleted = 0
                while True:
                    sql = """
                    DELETE FROM buff_goods_daily_price 
                    WHERE stat_date < %s
                    LIMIT 10000
                    """
                    cursor.execute(sql, (cutoff_date.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} 条超过50天的旧价格数据")
        except Exception as e:
            logger.error(f"清理旧数据失败: {e}")
            self._safe_rollback()
    
    def calculate_and_sync_avg_prices(self, goods_ids: List[str]) -> int:
        """计算7天/30天平均价并同步到 buff_goods_info 表
        
        这个方法在每次同步完成后调用，直接用1条SQL计算所有饰品的平均价
        """
        if not goods_ids:
            return 0
        
        try:
            with self.connection.cursor() as cursor:
                # 1. 计算并更新 buff_goods_daily_price 表的 week/month avg
                # 7天平均价
                cursor.execute("""
                UPDATE buff_goods_daily_price d
                INNER JOIN (
                    SELECT 
                        goods_id,
                        stat_date,
                        ROUND(SUM(min_price_sum) / SUM(min_price_cnt), 2) as week_avg
                    FROM buff_goods_daily_price
                    WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
                    GROUP BY goods_id, stat_date
                ) t ON d.goods_id = t.goods_id AND d.stat_date = t.stat_date
                SET d.min_price_week_avg = t.week_avg
                """)
                
                # 30天平均价
                cursor.execute("""
                UPDATE buff_goods_daily_price d
                INNER JOIN (
                    SELECT 
                        goods_id,
                        stat_date,
                        ROUND(SUM(min_price_sum) / SUM(min_price_cnt), 2) as month_avg
                    FROM buff_goods_daily_price
                    WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)
                    GROUP BY goods_id, stat_date
                ) t ON d.goods_id = t.goods_id AND d.stat_date = t.stat_date
                SET d.min_price_month_avg = t.month_avg
                """)
                
                # 2. 同步到 buff_goods_info 表（直接用1条SQL）
                # 7天平均价
                sql_7d = """
                UPDATE buff_goods_info g
                INNER JOIN (
                    SELECT 
                        goods_id,
                        ROUND(SUM(min_price_sum) / SUM(min_price_cnt), 2) as price_7d_avg
                    FROM buff_goods_daily_price
                    WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
                    AND min_price_sum > 0 AND min_price_cnt > 0
                    GROUP BY goods_id
                ) t ON g.goods_id = t.goods_id
                SET g.d7_avg_min_price = t.price_7d_avg,
                    g.update_time = NOW()
                """
                cursor.execute(sql_7d)
                affected_7d = cursor.connection.affected_rows()
                
                # 30天平均价
                sql_30d = """
                UPDATE buff_goods_info g
                INNER JOIN (
                    SELECT 
                        goods_id,
                        ROUND(SUM(min_price_sum) / SUM(min_price_cnt), 2) as price_30d_avg
                    FROM buff_goods_daily_price
                    WHERE stat_date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)
                    AND min_price_sum > 0 AND min_price_cnt > 0
                    GROUP BY goods_id
                ) t ON g.goods_id = t.goods_id
                SET g.d30_avg_min_price = t.price_30d_avg,
                    g.update_time = NOW()
                """
                cursor.execute(sql_30d)
                affected_30d = cursor.connection.affected_rows()
                
                self.connection.commit()
                logger.info(f"平均价计算并同步完成: 7天={affected_7d}条, 30天={affected_30d}条")
                return affected_7d
                
        except Exception as e:
            logger.error(f"计算平均价失败: {e}")
            self._safe_rollback()
            return 0
    
    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._safe_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._safe_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: 计算7天/30天平均价并同步到 goods_info 表
            logger.info("[Step 6/9] 计算平均价并同步到 goods_info...")
            avg_start = time.time()
            goods_ids = [str(item['goods_id']) for item in goods_info_list]
            avg_updated = self.db.calculate_and_sync_avg_prices(goods_ids)
            logger.info(f"平均价计算并同步完成: {avg_updated} 条，耗时: {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:
            if is_database_connection_error(e):
                logger.exception("数据库连接异常，本轮跳过，下一轮将重试: %s", e)
                return True
            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()


if __name__ == "__main__":
    sys.exit(0 if run_once() else 1)