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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品磨损区间价格数据同步脚本（修正数据逻辑）

功能：
1. 将有tag_id、tag_name、min_float、max_float的详细价格数据存储到float_price表
2. 将没有这些属性的基本价格数据存储到float_price_attr表
3. 周期调度由 run_buff_sync_loop.sh 统一控制

使用方法：
    python sync_float_price.py                    # 执行一次同步

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

import os
import sys
import time
import logging
import json
import requests
import pymysql
from datetime import datetime, timedelta

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

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

# BUFF API配置
BUFF_FLOAT_API_URL = "https://buff.163.com/api/market/enterprise/float/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")

# 请求配置
REQUEST_TIMEOUT = 60
# 批量插入：单批过大或网络慢时易触发超时；可通过环境变量调小单批或调大超时
BATCH_SIZE = int(os.getenv("BULK_INSERT_BATCH_SIZE", "5000"))
MAX_RETRIES = 3
RETRY_DELAY = 5
# 大批量写入建议 ≥300s，避免 PyMySQL 在长时间读写时触发 read/write 超时（默认 30 过短）
DB_READ_TIMEOUT = int(os.getenv("DB_READ_TIMEOUT", "600"))
DB_WRITE_TIMEOUT = int(os.getenv("DB_WRITE_TIMEOUT", "600"))
DB_BULK_INSERT_RETRIES = int(os.getenv("DB_BULK_INSERT_RETRIES", "3"))

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

def setup_logging():
    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, "logs", f"sync_float_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 FloatPriceDatabase:
    """磨损区间价格数据库操作类"""

    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}")
        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
        # 2006 lost connection, 2013 lost during query, 2014 commands out of sync,
        # 1053 server shutdown, 2003 can't connect
        return code in (2006, 2013, 2014, 1053, 2003)

    def _safe_rollback(self) -> None:
        """连接已断开时避免 rollback 再抛错掩盖原始异常。"""
        if not self.connection:
            return
        try:
            if getattr(self.connection, "open", False):
                self.connection.rollback()
        except Exception as e:
            logger.warning(f"rollback 失败（连接可能已断开）: {e}")

    def execute_with_transient_retry(self, label: str, fn: Callable[[], Any]) -> Any:
        """对断连、服务端关闭等可重试错误：安全回滚、重连后重新执行整段逻辑（如先 TRUNCATE 再 INSERT）。"""
        attempt = 0
        while True:
            try:
                return fn()
            except pymysql.err.OperationalError as e:
                attempt += 1
                self._safe_rollback()
                if not self._is_transient_db_error(e) or attempt > DB_BULK_INSERT_RETRIES:
                    raise
                logger.warning(
                    "%s 失败 (%s/%s)，将重连后重试: %s",
                    label,
                    attempt,
                    DB_BULK_INSERT_RETRIES,
                    e,
                )
                time.sleep(RETRY_DELAY * attempt)
                self._reconnect()
            except Exception:
                self._safe_rollback()
                raise

    def close(self):
        if self.connection and self.connection.open:
            self.connection.close()
            logger.info("数据库连接已关闭")
    
    def clear_table(self, table_name: str):
        """清空表"""
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(f"TRUNCATE TABLE {table_name}")
            self.connection.commit()
            logger.info(f"已清空 {table_name} 表")
        except Exception as e:
            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:
        """连接已断开时避免在 finally 里再次抛错掩盖原始异常。"""
        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)
                # 每批提交，缩短长事务、降低单次失败回滚粒度（中间表下一轮会 TRUNCATE）
                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_float_price_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到buff_goods_float_price_middle表（存储有tag和磨损区间的详细数据）"""
        if not data_list:
            return 0
        
        # 定义列和对应的数据字段
        columns = [
            'goods_id', 'name', 'qj_min_price', 'qj_buy_max_price',
            'qj_avg_min_price', 'qj_sell_num', 'qj_buy_order_num',
            'min_float', 'max_float', 'tag_id', 'tag_name',
            '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.get('goods_id', '')),
                'name': item.get('name', ''),
                'qj_min_price': float(item.get('min_price', 0)) if item.get('min_price') else 0.00,
                'qj_buy_max_price': float(item.get('buy_max_price', 0)) if item.get('buy_max_price') else 0.00,
                'qj_avg_min_price': float(item.get('avg_min_price', 0)) if item.get('avg_min_price') else 0.00,
                'qj_sell_num': int(item.get('sell_num', 0)) if item.get('sell_num') else 0,
                'qj_buy_order_num': int(item.get('buy_order_num', 0)) if item.get('buy_order_num') else 0,
                'min_float': float(item.get('min_float', 0)) if item.get('min_float') is not None else 0.000000,
                'max_float': float(item.get('max_float', 0)) if item.get('max_float') is not None else 0.000000,
                'tag_id': item.get('tag_id'),
                'tag_name': item.get('tag_name'),
                'stat_time': datetime.fromtimestamp(item.get('stat_time', 0)),
                'create_time': now
            })
        
        prep_time = time.time() - start_time
        logger.info(f"数据准备完成，耗时: {prep_time:.2f}秒")
        logger.info(f"开始插入 {len(processed_data)} 条数据到buff_goods_float_price_middle...")
        
        insert_start = time.time()
        result = self.bulk_insert('buff_goods_float_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_float_price(self) -> int:
        """将中间表数据插入到buff_goods_float_price表（先清空再插入）"""

        def _do_copy() -> int:
            self.clear_table("buff_goods_float_price")
            with self.connection.cursor() as cursor:
                sql = """
                INSERT INTO buff_goods_float_price 
                (goods_id, name, qj_min_price, qj_buy_max_price, qj_avg_min_price,
                 qj_sell_num, qj_buy_order_num, min_float, max_float, tag_id, tag_name,
                 stat_time, create_time, update_time)
                SELECT goods_id, name, qj_min_price, qj_buy_max_price, qj_avg_min_price,
                       qj_sell_num, qj_buy_order_num, min_float, max_float, tag_id, tag_name,
                       stat_time, NOW(), NOW()
                FROM buff_goods_float_price_middle
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            self.connection.commit()
            return affected_rows

        try:
            return self.execute_with_transient_retry(
                "复制详细价格到 buff_goods_float_price",
                _do_copy,
            )
        except Exception as e:
            logger.error(f"复制详细价格数据失败: {e}")
            raise
    
    def insert_float_price_attr_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到buff_goods_float_price_attr_middle表（存储没有tag和磨损区间的基本数据）"""
        if not data_list:
            return 0
        
        # 定义列和对应的数据字段
        columns = [
            'goods_id', 'name', 'qj_min_price', 'qj_buy_max_price',
            'qj_avg_min_price', 'qj_sell_num', 'qj_buy_order_num',
            '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.get('goods_id', '')),
                'name': item.get('name', ''),
                'qj_min_price': float(item.get('min_price', 0)) if item.get('min_price') else 0.00,
                'qj_buy_max_price': float(item.get('buy_max_price', 0)) if item.get('buy_max_price') else 0.00,
                'qj_avg_min_price': float(item.get('avg_min_price', 0)) if item.get('avg_min_price') else 0.00,
                'qj_sell_num': int(item.get('sell_num', 0)) if item.get('sell_num') else 0,
                'qj_buy_order_num': int(item.get('buy_order_num', 0)) if item.get('buy_order_num') else 0,
                'stat_time': datetime.fromtimestamp(item.get('stat_time', 0)),
                'create_time': now
            })
        
        prep_time = time.time() - start_time
        logger.info(f"数据准备完成，耗时: {prep_time:.2f}秒")
        logger.info(f"开始插入 {len(processed_data)} 条数据到buff_goods_float_price_attr_middle...")
        
        insert_start = time.time()
        result = self.bulk_insert('buff_goods_float_price_attr_middle', processed_data, columns)
        insert_time = time.time() - insert_start
        
        logger.info(f"插入完成: {result} 条，耗时: {insert_time:.2f}秒")
        return result
    
    def copy_from_middle_to_float_price_attr(self) -> int:
        """将中间表数据插入到buff_goods_float_price_attr表（先清空再插入）"""

        def _do_copy() -> int:
            self.clear_table("buff_goods_float_price_attr")
            with self.connection.cursor() as cursor:
                sql = """
                INSERT INTO buff_goods_float_price_attr 
                (goods_id, name, qj_min_price, qj_buy_max_price, qj_avg_min_price,
                 qj_sell_num, qj_buy_order_num, stat_time, create_time, update_time)
                SELECT goods_id, name, qj_min_price, qj_buy_max_price, qj_avg_min_price,
                       qj_sell_num, qj_buy_order_num, stat_time, NOW(), NOW()
                FROM buff_goods_float_price_attr_middle
                """
                cursor.execute(sql)
                affected_rows = cursor.connection.affected_rows()
            self.connection.commit()
            return affected_rows

        try:
            return self.execute_with_transient_retry(
                "复制基本价格到 buff_goods_float_price_attr",
                _do_copy,
            )
        except Exception as e:
            logger.error(f"复制基本价格数据失败: {e}")
            raise


# ============================================================
# BUFF API 操作类
# ============================================================

class BuffFloatPriceAPI:
    """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_float_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_FLOAT_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)}")
                
                # 检查数据格式
                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')}")
                    sales = first_item.get('sales', [])
                    logger.info(f"    sales数量: {len(sales)}")
                    
                    if sales:
                        for i, sale in enumerate(sales[:2]):  # 显示前2个sales
                            logger.info(f"    第{i+1}个sales:")
                            logger.info(f"      字段: {list(sale.keys())}")
                            logger.info(f"      min_float: {sale.get('min_float')}")
                            logger.info(f"      max_float: {sale.get('max_float')}")
                            logger.info(f"      tag_id: {sale.get('tag_id')}")
                            logger.info(f"      tag_name: {sale.get('tag_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 parse_float_price_data(self, api_response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
        """解析磨损区间价格数据，返回两种类型的数据列表
        
        Returns:
            tuple: (detailed_price_list, basic_price_list)
            detailed_price_list: 包含有tag和磨损区间的详细价格数据（用于buff_goods_float_price表）
            basic_price_list: 不包含tag和磨损区间的基本价格数据（用于buff_goods_float_price_attr表）
        """
        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)}")
        
        detailed_price_list = []    # 包含有tag和磨损区间的详细数据
        basic_price_list = []       # 不包含tag和磨损区间的基本数据
        
        start_time = time.time()
        total_detailed_count = 0
        total_basic_count = 0
        
        for item in info_data:
            goods_id = item.get("goods_id")
            name = item.get("name")
            sales = item.get("sales", [])
            
            # 标记是否已找到基本价格数据
            found_basic_price = False
            
            # 处理每个sales条目
            for sale in sales:
                # 检查是否有tag和磨损区间
                has_tag = sale.get("tag_id") is not None or sale.get("tag_name") is not None
                has_float = sale.get("min_float") is not None or sale.get("max_float") is not None
                
                # 构建销售数据
                sale_data = {
                    "goods_id": goods_id,
                    "name": name,
                    "min_price": sale.get("min_price"),
                    "avg_min_price": sale.get("avg_min_price"),
                    "buy_max_price": sale.get("buy_max_price"),
                    "sell_num": sale.get("sell_num"),
                    "buy_order_num": sale.get("buy_order_num"),
                    "stat_time": stat_time,
                    "min_float": sale.get("min_float"),
                    "max_float": sale.get("max_float"),
                    "tag_id": sale.get("tag_id"),
                    "tag_name": sale.get("tag_name")
                }
                
                # 如果有tag或磨损区间，则添加到详细价格列表
                if has_tag or has_float:
                    total_detailed_count += 1
                    detailed_price_list.append(sale_data)
                # 否则添加到基本价格列表（每个饰品只取第一个没有tag和磨损区间的数据）
                elif not found_basic_price:
                    total_basic_count += 1
                    found_basic_price = True
                    basic_price_list.append(sale_data)
        
        elapsed = time.time() - start_time
        logger.info(f"  解析完成，耗时: {elapsed:.2f}秒")
        logger.info(f"    总饰品记录: {len(info_data)} 条")
        logger.info(f"    详细价格数据（有tag或有磨损）: {total_detailed_count} 条")
        logger.info(f"    基本价格数据（无tag无磨损）: {total_basic_count} 条")
        
        # 统计不同类型的数据
        if detailed_price_list:
            logger.info(f"  详细价格数据样本:")
            for i in range(min(2, len(detailed_price_list))):
                sample = detailed_price_list[i]
                logger.info(f"    [{i}] goods_id: {sample.get('goods_id')}, "
                           f"min_float: {sample.get('min_float')}, tag_id: {sample.get('tag_id')}")
        
        if basic_price_list:
            logger.info(f"  基本价格数据样本:")
            for i in range(min(2, len(basic_price_list))):
                sample = basic_price_list[i]
                logger.info(f"    [{i}] goods_id: {sample.get('goods_id')}, "
                           f"min_price: {sample.get('min_price')}")
        
        if total_basic_count != len(info_data):
            logger.warning(f"警告: 基本价格数据数量({total_basic_count})与饰品记录数量({len(info_data)})不一致")
        
        return detailed_price_list, basic_price_list


# ============================================================
# 主逻辑
# ============================================================

class FloatPriceSyncTask:
    """磨损区间价格数据同步任务"""
    
    def __init__(self):
        self.db = None
        self.api = BuffFloatPriceAPI()
    
    def _init_db(self):
        """初始化数据库连接"""
        if self.db is None:
            self.db = FloatPriceDatabase()
    
    def run(self) -> bool:
        """执行同步任务"""
        start_time = datetime.now()
        logger.info("=" * 60)
        logger.info(f"开始执行磨损区间价格数据同步任务: {start_time}")
        logger.info("=" * 60)
        
        try:
            # 初始化数据库连接
            self._init_db()
            
            # Step 1: 清空磨损区间中间表
            logger.info("[Step 1/8] 清空buff_goods_float_price_middle表...")
            clear_start = time.time()
            self.db.clear_table("buff_goods_float_price_middle")
            logger.info(f"清空中间表耗时: {time.time() - clear_start:.2f}秒")
            
            # Step 2: 调用BUFF API获取磨损区间数据
            logger.info("[Step 2/8] 调用BUFF API获取磨损区间价格数据...")
            api_start = time.time()
            api_response = self.api.fetch_float_price_data(game="cs2", include_sticker=1)
            logger.info(f"API调用耗时: {time.time() - api_start:.2f}秒")
            
            # Step 3: 解析数据
            logger.info("[Step 3/8] 解析磨损区间数据...")
            parse_start = time.time()
            detailed_price_list, basic_price_list = self.api.parse_float_price_data(api_response)
            logger.info(f"数据解析耗时: {time.time() - parse_start:.2f}秒")
            
            if not detailed_price_list and not basic_price_list:
                logger.warning("未获取到任何价格数据，任务取消")
                return False
            
            # Step 4: 插入详细价格数据到中间表（有tag和磨损区间的数据）
            logger.info("[Step 4/8] 插入详细价格数据到中间表...")
            inserted_detailed_count = self.db.insert_float_price_middle(detailed_price_list)
            
            # Step 5: 清空目标表并更新主表
            logger.info("[Step 5/8] 清空buff_goods_float_price表并更新数据...")
            copy_start = time.time()
            float_inserted = self.db.copy_from_middle_to_float_price()
            logger.info(f"详细价格主表插入完成: {float_inserted} 条，耗时: {time.time() - copy_start:.2f}秒")
            
            # Step 6: 清空属性中间表
            logger.info("[Step 6/8] 清空buff_goods_float_price_attr_middle表...")
            clear_attr_start = time.time()
            self.db.clear_table("buff_goods_float_price_attr_middle")
            logger.info(f"清空属性中间表耗时: {time.time() - clear_attr_start:.2f}秒")
            
            # Step 7: 插入基本价格数据到中间表
            logger.info("[Step 7/8] 插入基本价格数据到中间表...")
            inserted_basic_count = self.db.insert_float_price_attr_middle(basic_price_list)
            
            # Step 8: 清空目标表并更新属性主表
            logger.info("[Step 8/8] 清空buff_goods_float_price_attr表并更新数据...")
            copy_attr_start = time.time()
            attr_inserted = self.db.copy_from_middle_to_float_price_attr()
            logger.info(f"基本价格主表插入完成: {attr_inserted} 条，耗时: {time.time() - copy_attr_start:.2f}秒")
            
            # 清理中间表
            logger.info("[清理] 清空中间表数据...")
            cleanup_mid_start = time.time()
            self.db.clear_table("buff_goods_float_price_middle")
            self.db.clear_table("buff_goods_float_price_attr_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"  详细价格数据: {inserted_detailed_count} 条（有tag或有磨损）")
            logger.info(f"  基本价格数据: {inserted_basic_count} 条（无tag无磨损）")
            logger.info(f"  总处理饰品记录: {len(api_response.get('info', []))} 条")
            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 = FloatPriceSyncTask()
    return task.run()


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