buff3.0/buff-api/sync_goods_attr/sync_goods_attr_by_ids.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品属性获取脚本（按指定饰品ID）

对指定的饰品ID进行数据爬取并写入 buff_goods_attr。
支持多个饰品ID，命令行用逗号分隔。

用法：
  python sync_goods_attr_by_ids.py 1115519,1115528,1234567
  python sync_goods_attr_by_ids.py 1115519

依赖、数据库与请求配置与 sync_goods_attr.py 一致（DBUtils、每 3 秒一次请求等）。
"""

import os
import sys
import time
import logging
import threading
import json
import requests
from decimal import Decimal
import pymysql
from dbutils.pooled_db import PooledDB
import concurrent.futures
from datetime import datetime
from typing import List, Dict, Any, Optional, Tuple

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

BUFF_API_BASE_URL = "https://buff.163.com/api/market/goods/sell_order"
GAME = "csgo"

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")

DB_POOL_CONFIG = {
    'creator': pymysql,
    'host': DB_HOST,
    'port': DB_PORT,
    'user': DB_USER,
    'password': DB_PASSWORD,
    'database': DB_NAME,
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor,
    'autocommit': True,
    'maxconnections': 10,
    'mincached': 3,
    'maxcached': 5,
    'maxusage': 1000,
    'ping': 7,
    'blocking': True,
    'setsession': [],
}

REQUEST_TIMEOUT = 30
REQUEST_INTERVAL = 3
CONCURRENT_WORKERS = int(os.getenv("CONCURRENT_WORKERS", "1"))
MAX_RETRIES = 3
RETRY_DELAY = 2

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

def setup_logging():
    """配置日志系统"""
    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, "logs", f"sync_goods_attr_by_ids_{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 - [%(threadName)s] - %(message)s',
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler(log_file, encoding='utf-8')
        ]
    )
    return logging.getLogger(__name__)

logger = setup_logging()


def _to_db_scalar(v):
    """将可能为 list/dict 的值转为标量，避免 cursor.execute 时 'not all arguments converted'"""
    if v is None:
        return None
    if isinstance(v, (list, dict)):
        return json.dumps(v, ensure_ascii=False) if v else None
    if isinstance(v, (datetime, Decimal)):
        return v
    if isinstance(v, (str, int, float, bool)):
        return v
    return str(v)


_request_lock = threading.Lock()
_last_request_time = [0.0]

# ============================================================
# 数据库连接池操作类
# ============================================================

class DatabaseConnectionPool:
    _instance = None
    _pool = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._init_pool()
        return cls._instance

    def _init_pool(self):
        try:
            self._pool = PooledDB(**DB_POOL_CONFIG)
            logger.info(f"数据库连接池初始化成功: {DB_HOST}:{DB_PORT}/{DB_NAME}")
        except Exception as e:
            logger.error(f"数据库连接池初始化失败: {e}")
            raise

    def get_connection(self):
        try:
            return self._pool.connection()
        except Exception as e:
            logger.error(f"从连接池获取连接失败: {e}")
            time.sleep(2)
            self._init_pool()
            return self._pool.connection()

    def close_pool(self):
        if self._pool:
            self._pool.close()
            logger.info("数据库连接池已关闭")


class GoodsAttrDatabase:
    def __init__(self):
        self.pool = DatabaseConnectionPool()

    def execute_query(self, sql: str, params=None, fetch_all=True):
        conn = None
        cursor = None
        try:
            conn = self.pool.get_connection()
            cursor = conn.cursor()
            if params:
                cursor.execute(sql, params)
            else:
                cursor.execute(sql)
            return cursor.fetchall() if fetch_all else cursor.fetchone()
        except pymysql.Error as e:
            logger.error(f"数据库查询失败: {e}, SQL: {sql}")
            raise
        finally:
            if cursor:
                cursor.close()

    def execute_update(self, sql: str, params=None, many=False):
        conn = None
        cursor = None
        try:
            conn = self.pool.get_connection()
            cursor = conn.cursor()
            if many and params:
                cursor.executemany(sql, params)
            elif params:
                cursor.execute(sql, params)
            else:
                cursor.execute(sql)
            return cursor.rowcount
        except pymysql.Error as e:
            logger.error(f"数据库更新失败: {e}, SQL: {sql}")
            raise
        finally:
            if cursor:
                cursor.close()

    def execute_transaction(self, operations: List[Tuple[str, tuple]]):
        conn = None
        cursor = None
        try:
            conn = self.pool.get_connection()
            cursor = conn.cursor()
            conn.begin()
            for sql, params in operations:
                if params:
                    cursor.execute(sql, params)
                else:
                    cursor.execute(sql)
            conn.commit()
            return True
        except Exception as e:
            logger.error(f"事务执行失败: {e}")
            if conn:
                try:
                    conn.rollback()
                except Exception:
                    pass
            return False
        finally:
            if cursor:
                cursor.close()

    def batch_insert_goods_attrs(self, attr_data_list: List[Dict[str, Any]]) -> int:
        if not attr_data_list:
            return 0
        operations = []
        for attr_data in attr_data_list:
            delete_sql = "DELETE FROM buff_goods_attr WHERE goods_id = %s"
            insert_sql = """
            INSERT INTO buff_goods_attr
            (goods_id, name, icon_url, type, type_name, category, category_name,
             category_group, category_group_name, rarity, rarity_name,
             quality, quality_name, exterior, exterior_name, weapon, weapon_name,
             series, series_name, custom, custom_name, itemset, itemset_name,
             extra_tag_ids, extra_tag_ids_name, custom_patch, custom_patch_name,
             custom_charm, custom_charm_name, stat_time, create_time, update_time)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
            """
            values = (
                _to_db_scalar(attr_data.get('goods_id')),
                _to_db_scalar(attr_data.get('name')) or '',
                _to_db_scalar(attr_data.get('icon_url')),
                _to_db_scalar(attr_data.get('type')),
                _to_db_scalar(attr_data.get('type_name')),
                _to_db_scalar(attr_data.get('category')),
                _to_db_scalar(attr_data.get('category_name')),
                _to_db_scalar(attr_data.get('category_group')),
                _to_db_scalar(attr_data.get('category_group_name')),
                _to_db_scalar(attr_data.get('rarity')),
                _to_db_scalar(attr_data.get('rarity_name')),
                _to_db_scalar(attr_data.get('quality')),
                _to_db_scalar(attr_data.get('quality_name')),
                _to_db_scalar(attr_data.get('exterior')),
                _to_db_scalar(attr_data.get('exterior_name')),
                _to_db_scalar(attr_data.get('weapon')),
                _to_db_scalar(attr_data.get('weapon_name')),
                _to_db_scalar(attr_data.get('series')),
                _to_db_scalar(attr_data.get('series_name')),
                _to_db_scalar(attr_data.get('custom')),
                _to_db_scalar(attr_data.get('custom_name')),
                _to_db_scalar(attr_data.get('itemset')),
                _to_db_scalar(attr_data.get('itemset_name')),
                _to_db_scalar(attr_data.get('extra_tag_ids')),
                _to_db_scalar(attr_data.get('extra_tag_ids_name')),
                _to_db_scalar(attr_data.get('custom_patch')),
                _to_db_scalar(attr_data.get('custom_patch_name')),
                _to_db_scalar(attr_data.get('custom_charm')),
                _to_db_scalar(attr_data.get('custom_charm_name')),
                _to_db_scalar(attr_data.get('stat_time')) or datetime.now(),
            )
            goods_id_val = _to_db_scalar(attr_data.get('goods_id')) or attr_data['goods_id']
            operations.append((delete_sql, (goods_id_val,)))
            operations.append((insert_sql, values))
        try:
            if self.execute_transaction(operations):
                logger.info(f"成功批量插入 {len(attr_data_list)} 个饰品的属性信息")
                return len(attr_data_list)
            return 0
        except Exception as e:
            logger.error(f"批量插入饰品属性信息异常: {e}")
            return 0

    def check_connection(self):
        try:
            self.execute_query("SELECT 1")
            return True
        except Exception as e:
            logger.error(f"数据库连接检查失败: {e}")
            return False


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

class BuffGoodsAPI:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
            "Referer": "https://buff.163.com/",
            "Sec-Fetch-Dest": "empty",
            "Sec-Fetch-Mode": "cors",
            "Sec-Fetch-Site": "same-origin"
        })

    def fetch_goods_info(self, goods_id: str) -> Optional[Dict[str, Any]]:
        url = f"{BUFF_API_BASE_URL}?game={GAME}&goods_id={goods_id}"
        for retry in range(MAX_RETRIES):
            try:
                with _request_lock:
                    elapsed = time.time() - _last_request_time[0]
                    if elapsed < REQUEST_INTERVAL:
                        time.sleep(REQUEST_INTERVAL - elapsed)
                    _last_request_time[0] = time.time()
                response = requests.get(
                    url,
                    headers=self.session.headers,
                    timeout=REQUEST_TIMEOUT,
                )
                response.raise_for_status()
                data = response.json()
                if data.get("code") != "OK":
                    logger.warning(f"饰品 {goods_id} API返回错误: {data.get('msg', '未知错误')}")
                    return None
                goods_info = self._extract_goods_info(data, goods_id)
                if goods_info:
                    return goods_info
                return None
            except requests.exceptions.RequestException as e:
                logger.error(f"请求饰品 {goods_id} 失败 (尝试 {retry + 1}/{MAX_RETRIES}): {e}")
                if retry < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY * (retry + 1))
                else:
                    return None
            except json.JSONDecodeError:
                return None
            except Exception as e:
                logger.error(f"处理饰品 {goods_id} 时发生错误: {e}")
                return None
        return None

    def _extract_goods_info(self, api_data: Dict[str, Any], goods_id: str) -> Optional[Dict[str, Any]]:
        try:
            data = api_data.get("data", {})
            goods_infos = data.get("goods_infos", {})
            goods_data = goods_infos.get(str(goods_id))
            if not goods_data:
                return None
            tags = goods_data.get("tags", {})
            attr_data = {
                'goods_id': goods_id,
                'name': goods_data.get("name", ""),
                'icon_url': goods_data.get("icon_url"),
                'stat_time': datetime.now(),
                'create_time': datetime.now()
            }
            tag_mapping = {
                'type': ('type', 'type_name'),
                'rarity': ('rarity', 'rarity_name'),
                'category': ('category', 'category_name'),
                'category_group': ('category_group', 'category_group_name'),
                'quality': ('quality', 'quality_name'),
                'exterior': ('exterior', 'exterior_name'),
                'weapon': ('weapon', 'weapon_name'),
                'series': ('series', 'series_name'),
                'custom': ('custom', 'custom_name'),
                'itemset': ('itemset', 'itemset_name')
            }
            for tag_key, (field_key, field_name_key) in tag_mapping.items():
                if tag_key in tags:
                    tag_data = tags[tag_key]
                    attr_data[field_key] = tag_data.get("internal_name")
                    attr_data[field_name_key] = tag_data.get("localized_name")
            return attr_data
        except Exception as e:
            logger.error(f"提取饰品 {goods_id} 属性时出错: {e}")
            return None


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

class GoodsAttrSyncTask:
    def __init__(self):
        self.db = GoodsAttrDatabase()
        self.api = BuffGoodsAPI()

    def fetch_and_save_goods_attrs(self, goods_ids: List[str]) -> Dict[str, int]:
        results = {'success': 0, 'failed': 0, 'total': len(goods_ids)}
        batch_attr_data: List[Dict[str, Any]] = []

        def fetch_one(goods_id: str) -> Optional[Dict[str, Any]]:
            try:
                return self.api.fetch_goods_info(goods_id)
            except Exception as e:
                logger.error(f"获取饰品 {goods_id} 属性异常: {e}")
                return None

        with concurrent.futures.ThreadPoolExecutor(max_workers=CONCURRENT_WORKERS) as executor:
            future_to_id = {executor.submit(fetch_one, gid): gid for gid in goods_ids}
            for future in concurrent.futures.as_completed(future_to_id):
                goods_id = future_to_id[future]
                try:
                    attr_data = future.result()
                    if attr_data:
                        batch_attr_data.append(attr_data)
                        logger.info(f"获取饰品 {goods_id} 属性成功")
                    else:
                        results['failed'] += 1
                        logger.warning(f"获取饰品 {goods_id} 属性失败")
                except Exception as e:
                    results['failed'] += 1
                    logger.error(f"获取饰品 {goods_id} 异常: {e}")

        if batch_attr_data:
            saved_count = self.db.batch_insert_goods_attrs(batch_attr_data)
            results['success'] += saved_count
            if saved_count < len(batch_attr_data):
                results['failed'] += len(batch_attr_data) - saved_count
        return results

    def cleanup(self):
        try:
            DatabaseConnectionPool().close_pool()
            logger.info("资源清理完成")
        except Exception:
            pass


def parse_goods_ids(ids_str: str) -> List[str]:
    """从逗号分隔的字符串解析出饰品ID列表，去重并过滤空串"""
    if not ids_str or not ids_str.strip():
        return []
    parts = [p.strip() for p in ids_str.split(",") if p.strip()]
    seen = set()
    result = []
    for p in parts:
        if p not in seen:
            seen.add(p)
            result.append(p)
    return result


# ============================================================
# 主函数
# ============================================================

def main():
    task = None
    try:
        # 从命令行获取逗号分隔的饰品ID：第一个参数
        if len(sys.argv) < 2:
            print("用法: python sync_goods_attr_by_ids.py <饰品ID1>,<饰品ID2>,...")
            print("示例: python sync_goods_attr_by_ids.py 1115519,1115528,1234567")
            sys.exit(1)

        ids_str = sys.argv[1]
        goods_ids = parse_goods_ids(ids_str)
        if not goods_ids:
            logger.error("未解析到有效饰品ID，请使用逗号分隔的ID列表")
            sys.exit(1)

        logger.info("启动按指定饰品ID属性同步脚本...")
        logger.info(f"待处理饰品ID: {goods_ids}")
        logger.info(f"共 {len(goods_ids)} 个，请求间隔 {REQUEST_INTERVAL} 秒")

        task = GoodsAttrSyncTask()
        if not task.db.check_connection():
            logger.error("数据库连接不可用，任务中止")
            sys.exit(1)

        start_time = datetime.now()
        results = task.fetch_and_save_goods_attrs(goods_ids)
        duration = (datetime.now() - start_time).total_seconds()

        logger.info("=" * 60)
        logger.info("任务完成")
        logger.info(f"总数: {results['total']}, 成功: {results['success']}, 失败: {results['failed']}, 耗时: {duration:.2f}秒")
        logger.info("=" * 60)
        sys.exit(0 if results['failed'] == 0 else 1)

    except KeyboardInterrupt:
        logger.info("\n收到中断信号，正在退出...")
        if task:
            task.cleanup()
        sys.exit(0)
    except Exception as e:
        logger.exception(f"程序运行异常: {e}")
        if task:
            task.cleanup()
        sys.exit(1)


if __name__ == "__main__":
    try:
        import dbutils
    except ImportError:
        print("请先安装依赖: pip install DBUtils==3.0.2")
        sys.exit(1)
    main()