buff3.0/buff-api-bak/sync_liquidity/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                    # 手动执行一次
    python sync_liquidity.py --daemon           # 守护进程模式（定时执行）
    python sync_liquidity.py --install-cron     # 安装到系统定时任务

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

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

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

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

# 数据库配置（从环境变量读取）
DB_HOST = os.getenv("DB_HOST", "125.122.32.114")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "Buff123456")
DB_NAME = os.getenv("DB_NAME", "buff")

# 请求配置
REQUEST_TIMEOUT = 60
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
            )
            logger.info(f"数据库连接成功: {DB_HOST}:{DB_PORT}/{DB_NAME}")
        except Exception as e:
            logger.error(f"数据库连接失败: {e}")
            raise
    
    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
        
        with self.connection.cursor() as cursor:
            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
                ]
                cursor.executemany(sql, values)
                total_inserted += len(batch)
                logger.info(f"中间表插入进度: {total_inserted}/{len(data_list)}")
        
        self.connection.commit()
        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:
            logger.exception(f"同步任务失败: {e}")
            return False
        finally:
            self.db.close()


def run_once() -> bool:
    """执行一次同步任务"""
    task = LiquiditySyncTask()
    return task.run()


def run_daemon():
    """守护进程模式，持续运行并定时执行任务"""
    logger.info("启动守护进程模式... 定时任务: 每天 02:00 执行")
    schedule.every().day.at("02:00").do(run_once)
    run_once()  # 立即执行一次
    
    while True:
        schedule.run_pending()
        time.sleep(60)


def main():
    parser = argparse.ArgumentParser(description="BUFF饰品流通性数据同步脚本")
    parser.add_argument("--daemon", action="store_true", help="守护进程模式")
    parser.add_argument("--install-cron", action="store_true", help="安装到系统定时任务（每天02:00执行）")
    
    args = parser.parse_args()
    
    if args.install_cron:
        script_path = os.path.abspath(__file__)
        cron_entry = f"0 2 * * * /usr/bin/python3 {script_path}\n"
        
        try:
            import subprocess
            result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
            current_crontab = result.stdout
            
            if script_path in current_crontab:
                logger.info("定时任务已存在，无需重复安装")
            else:
                subprocess.run(["crontab", "-"], input=current_crontab + cron_entry, text=True)
                logger.info(f"定时任务安装成功: {cron_entry.strip()}")
        except Exception as e:
            logger.error(f"安装定时任务失败: {e}")
            sys.exit(1)
    elif args.daemon:
        run_daemon()
    else:
        success = run_once()
        sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()