#!/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)