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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品流通性数据同步脚本

功能：
1. 请求 buff 接口获取饰品流通性数据
2. 将数据存入 buff_goods_liquidity_middle 表
3. 确认存储数量正确后，清空 buff_goods_liquidity 表
4. 将 buff_goods_liquidity_middle 插入到 buff_goods_liquidity 表中

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

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

import os
import sys
import time
import logging
import json
import requests
import pymysql
from datetime import datetime
from typing import List, Dict, Any

from sync_db_errors import is_database_connection_error

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

# BUFF API配置
BUFF_API_URL = "https://buff.163.com/api/market/enterprise/items/liquidity_rank"
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")
# 与 sync_goods / sync_float_price 一致，大批量写入避免客户端读写过短触发 2013
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"))

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

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

def setup_logging():
    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, "logs", f"sync_liquidity_{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 LiquidityDatabase:
    """流通性数据库操作类"""

    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 get_middle_count(self) -> int:
        """获取中间表数据数量"""
        with self.connection.cursor() as cursor:
            cursor.execute("SELECT COUNT(*) as count FROM buff_goods_liquidity_middle")
            result = cursor.fetchone()
            return result['count']
    
    def clear_target_table(self):
        """清空目标表"""
        with self.connection.cursor() as cursor:
            cursor.execute("DELETE FROM buff_goods_liquidity")
        self.connection.commit()
        logger.info("已清空 buff_goods_liquidity 表")
    
    def clear_middle_table(self):
        """清空中间表"""
        with self.connection.cursor() as cursor:
            cursor.execute("DELETE FROM buff_goods_liquidity_middle")
        self.connection.commit()
        logger.info("已清空 buff_goods_liquidity_middle 表")
    
    def insert_to_middle(self, data_list: List[Dict[str, Any]]) -> int:
        """批量插入数据到中间表；断连类错误重连并重试当前批，每批提交。"""
        if not data_list:
            return 0

        sql = """
        INSERT INTO buff_goods_liquidity_middle 
        (goods_id, name, tag_id, tag_name, liquidity_rank, stat_time, create_time)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
        """

        total_inserted = 0

        for i in range(0, len(data_list), BATCH_SIZE):
            batch = data_list[i:i + BATCH_SIZE]
            values = [
                (
                    item['goods_id'],
                    item['name'],
                    item.get('tag_id'),
                    item.get('tag_name'),
                    item.get('liquidity_rank'),
                    datetime.fromtimestamp(item['update_time']),
                    datetime.now(),
                )
                for item in batch
            ]
            attempt = 0
            while True:
                try:
                    with self.connection.cursor() as cursor:
                        cursor.executemany(sql, values)
                    self.connection.commit()
                    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()

            total_inserted += len(batch)
            logger.info(f"中间表插入进度: {total_inserted}/{len(data_list)}")

        return total_inserted
    
    def insert_from_middle_to_target(self) -> int:
        """将中间表数据插入到目标表"""
        with self.connection.cursor() as cursor:
            sql = """
            INSERT INTO buff_goods_liquidity 
            (goods_id, name, tag_id, tag_name, liquidity_rank, stat_time, create_time, update_time)
            SELECT goods_id, name, tag_id, tag_name, liquidity_rank, stat_time, create_time, NOW()
            FROM buff_goods_liquidity_middle
            """
            cursor.execute(sql)
            affected_rows = cursor.connection.affected_rows()
        
        self.connection.commit()
        return affected_rows
    
    def get_target_count(self) -> int:
        """获取目标表数据数量"""
        with self.connection.cursor() as cursor:
            cursor.execute("SELECT COUNT(*) as count FROM buff_goods_liquidity")
            result = cursor.fetchone()
            return result['count']


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

class BuffLiquidityAPI:
    """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_liquidity_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()

                # 🔧 调试：打印API返回的完整数据结构
                logger.info("=" * 60)
                logger.info("🔍 API返回数据结构:")
                logger.info(f"  顶层keys: {list(data.keys())}")
                logger.info(f"  总记录数: {len(data)}")

                # 打印Info字段详情
                if "Info" in data:
                    logger.info(f"  Info字段类型: {type(data['Info'])}")
                    if isinstance(data['Info'], list):
                        logger.info(f"  Info列表长度: {len(data['Info'])}")
                        if len(data['Info']) > 0:
                            logger.info(f"  第一条记录keys: {list(data['Info'][0].keys())}")
                            logger.info(f"  第一条记录示例: {json.dumps(data['Info'][0], ensure_ascii=False, indent=2)}")
                    else:
                        logger.info(f"  Info字段内容: {data['Info']}")
                elif "info" in data:
                    logger.info(f"  info字段类型: {type(data['info'])}")
                    if isinstance(data['info'], list):
                        logger.info(f"  info列表长度: {len(data['info'])}")
                        if len(data['info']) > 0:
                            logger.info(f"  第一条记录keys: {list(data['info'][0].keys())}")
                            logger.info(f"  第一条记录示例: {json.dumps(data['info'][0], ensure_ascii=False, indent=2)}")
                    else:
                        logger.info(f"  info字段内容: {data['info']}")

                # 打印其他关键字段
                for key in ["Stat_time", "stat_time", "Req_remaining", "req_remaining", "code"]:
                    if key in data:
                        logger.info(f"  {key}: {data[key]}")

                logger.info("=" * 60)

                # 兼容处理：检查是否有Info字段（成功标志）
                info_key = "Info" if "Info" in data else "info"
                if info_key in data:
                    # 获取信息列表
                    info_count = len(data.get(info_key, []))
                    logger.info(f"API请求成功，获取到 {info_count} 条记录")
                    
                    # 获取剩余请求次数（兼容大小写）
                    remaining = data.get("Req_remaining") or data.get("req_remaining", 0)
                    logger.info(f"API剩余调用次数: {remaining}")
                    return data
                else:
                    # 没有Info字段，检查是否有code
                    code = data.get("code")
                    if code == 200 or code == "200":
                        logger.info(f"API请求成功")
                        return data
                    else:
                        logger.error(f"API返回错误: 无Info字段, 完整响应: {data}")
                        raise Exception(f"API错误: {data.get('msg', '未知错误')}")
                    
            except requests.exceptions.RequestException as e:
                logger.error(f"API请求失败 (尝试 {retry + 1}/{MAX_RETRIES}): {e}")
                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_liquidity_data(self, api_response: Dict[str, Any]) -> List[Dict[str, Any]]:
        """解析流通性数据"""

        # 🔧 调试：打印解析前数据
        logger.info("=" * 60)
        logger.info("🔍 开始解析数据...")
        logger.info(f"  Stat_time: {api_response.get('Stat_time', '不存在')}")

        # 获取Info数据（兼容大小写）
        info_data = api_response.get("Info") or api_response.get("info", [])
        logger.info(f"  Info列表长度: {len(info_data) if info_data else 0}")

        items = []
        stat_time = api_response.get("Stat_time", 0)

        # 🔧 调试：检查第一条数据
        if info_data and len(info_data) > 0:
            first_item = info_data[0]
            logger.info(f"  第一条goods_id: {first_item.get('goods_id')}")
            logger.info(f"  第一条market_hash_name: {first_item.get('market_hash_name')}")
            logger.info(f"  第一条sales字段: {first_item.get('sales', '不存在')}")
            if 'sales' in first_item:
                logger.info(f"  sales类型: {type(first_item['sales'])}")
                if isinstance(first_item['sales'], list) and len(first_item['sales']) > 0:
                    logger.info(f"  第一条sales[0]内容: {json.dumps(first_item['sales'][0], ensure_ascii=False)}")

        for item in info_data:
            goods_id = item.get("goods_id")
            market_hash_name = item.get("market_hash_name")
            name = item.get("name")  # 中文名称
            update_time = item.get("update_time", stat_time)

            sales = item.get("sales", [])
            logger.info(f"  处理 goods_id={goods_id}, sales数量={len(sales) if sales else 0}")

            for sale in sales:
                items.append({
                    "goods_id": goods_id,
                    "market_hash_name": market_hash_name,
                    "name": name,  # 存储中文名称
                    "update_time": update_time,
                    "tag_id": sale.get("tag_id"),
                    "tag_name": sale.get("tag_name"),
                    "liquidity_rank": sale.get("liquidity_rank")
                })

        logger.info(f"  解析完成，共 {len(items)} 条流通性记录")
        logger.info("=" * 60)
        return items


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

class LiquiditySyncTask:
    """流通性数据同步任务"""
    
    def __init__(self):
        self.db = LiquidityDatabase()
        self.api = BuffLiquidityAPI()
    
    def run(self) -> bool:
        """执行同步任务"""
        start_time = datetime.now()
        logger.info("=" * 60)
        logger.info(f"开始执行流通性数据同步任务: {start_time}")
        logger.info("=" * 60)
        
        try:
            # Step 1: 清空中间表
            logger.info("[Step 1/5] 清空中间表...")
            self.db.clear_middle_table()
            
            # Step 2: 调用BUFF API获取数据
            logger.info("[Step 2/5] 调用BUFF API获取流通性数据...")
            api_response = self.api.fetch_liquidity_data(game="cs2", include_sticker=1)
            # 兼容大小写字段名
            remaining = api_response.get("Req_remaining") or api_response.get("req_remaining", 0)
            logger.info(f"API剩余调用次数: {remaining}")
            
            # Step 3: 解析数据
            logger.info("[Step 3/5] 解析数据...")
            data_list = self.api.parse_liquidity_data(api_response)
            
            if not data_list:
                logger.warning("未获取到任何流通性数据，任务取消")
                return False
            
            # Step 4: 插入中间表
            logger.info("[Step 4/5] 插入中间表...")
            inserted_count = self.db.insert_to_middle(data_list)
            middle_count = self.db.get_middle_count()
            logger.info(f"中间表插入完成: 预期 {inserted_count}, 实际 {middle_count}")
            
            # 验证数据量（允许1%误差）
            if middle_count < inserted_count * 0.99:
                logger.error(f"数据量验证失败: 预期 {inserted_count}, 实际 {middle_count}")
                return False
            
            logger.info("数据量验证通过")
            
            # Step 5: 清空目标表并从中间表插入
            logger.info("[Step 5/5] 更新目标表...")
            self.db.clear_target_table()
            target_inserted = self.db.insert_from_middle_to_target()
            target_count = self.db.get_target_count()
            
            logger.info(f"目标表更新完成: 插入 {target_inserted}, 当前 {target_count}")
            
            # 验证目标表数据量
            if target_count != middle_count:
                logger.error(f"目标表数据量不匹配: 中间表 {middle_count}, 目标表 {target_count}")
                return False
            
            # 清理中间表数据
            self.db.clear_middle_table()
            
            # 任务完成
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()
            logger.info("=" * 60)
            logger.info(f"流通性数据同步任务完成! 耗时: {duration:.2f}秒")
            logger.info(f"同步数据量: {target_count} 条")
            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:
            self.db.close()


def run_once() -> bool:
    """执行一次同步任务"""
    try:
        task = LiquiditySyncTask()
        return task.run()
    except Exception as e:
        if is_database_connection_error(e):
            logger.exception("数据库连接异常，本轮跳过，下一轮将重试: %s", e)
            return True
        raise


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