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