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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF饰品属性信息获取脚本（使用连接池版本）

特点：
1. 使用DBUtils连接池管理数据库连接
2. 自动处理连接失效和重连
3. 更稳定的长连接管理
4. 避免频繁创建和关闭连接的开销
5. 通过代理请求接口，403 自动切换代理重试

快速补齐缺失属性：可设置环境变量调大每轮处理量，例如：
  BATCH_SIZE=200 python sync_goods_attr.py
（当前为串行模式：每次 1 个请求，不主动增加固定请求间隔）

安装依赖：
pip install DBUtils==3.0.2
"""

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


def _scalar_for_db(val: Any) -> Any:
    """将 API 返回的 list/dict 等转为 DB 单值，避免 'not all arguments converted during string formatting'"""
    if val is None or isinstance(val, (str, int, float)):
        return val
    if isinstance(val, datetime):
        return val
    if isinstance(val, (list, dict)):
        return json.dumps(val, ensure_ascii=False) if val else None
    # 元组、集合等可迭代但非标量一律转成字符串，避免被当成多个参数
    return str(val) if val is not None else None


# 与 buff_goods_attr 表 INSERT 列顺序一致（不含 create_time/update_time，用 NOW()）
INSERT_ATTR_COLUMNS = [
    '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',
]
INSERT_ATTR_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())
"""


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

# BUFF API配置
BUFF_API_BASE_URL = "https://buff.163.com/api/market/goods/sell_order"
GAME = "csgo"
PROXY_POOL_API_URL = os.getenv(
    "PROXY_POOL_API_URL",
    "https://api.haiwaiip.net/api/pull?token=3d706c31d8d0267d757706b3e4d81454&num=1&format=text&protocol=http&country=&state=&city=&ip_type=datacenter&sep=1&area=",
)
PROXY_POOL_TIMEOUT = int(os.getenv("PROXY_POOL_TIMEOUT", "10"))

# 数据库配置
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,  # 使用ping检查连接：0=从不, 1=创建时, 2=获取时, 4=归还时, 7=总是
    'blocking': True,  # 连接池满时是否等待
    'setsession': [],  # 可选的会话命令列表
}

# 请求配置（可通过环境变量 BATCH_SIZE 调大每轮处理量）
REQUEST_TIMEOUT = 30
CONCURRENT_WORKERS = int(os.getenv("CONCURRENT_WORKERS", "1"))  # 并发请求数（1=串行）
GOODS_PER_REQUEST = int(os.getenv("GOODS_PER_REQUEST", "10"))  # 仅用于批量模式（当前主流程默认串行单请求）
MAX_RETRIES = 3
RETRY_DELAY = 2
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "30"))  # 每轮拉取的饰品数量，每批提交一次，调小便于数据及时入库
FAILURE_LONG_REST_THRESHOLD = int(os.getenv("FAILURE_LONG_REST_THRESHOLD", "5"))  # 连续请求失败达到阈值后触发长休息
FAILURE_LONG_REST_MIN_SEC = int(os.getenv("FAILURE_LONG_REST_MIN_SEC", "300"))    # 失败触发长休息最短秒数（默认 5 分钟）
FAILURE_LONG_REST_MAX_SEC = int(os.getenv("FAILURE_LONG_REST_MAX_SEC", "600"))    # 失败触发长休息最长秒数（默认 10 分钟）
FORBIDDEN_STOP_THRESHOLD = int(os.getenv("FORBIDDEN_STOP_THRESHOLD", "3"))  # 连续检测到 403 后直接停止脚本的阈值

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

def setup_logging():
    """配置日志系统"""
    log_dir = os.path.dirname(os.path.abspath(__file__))
    log_file = os.path.join(log_dir, "logs", f"sync_goods_attr_{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()

_request_lock = threading.Lock()
_consecutive_request_failures = [0]  # 连续请求失败计数（网络错误 / API 返回错误）
_consecutive_403_errors = [0]        # 连续 403 错误计数


def _record_request_failure():
    """
    记录一次请求失败（网络错误、状态码/业务code异常等），
    当连续失败次数达到阈值时随机长休息 5-10 分钟以降低被封 IP 风险。
    """
    with _request_lock:
        _consecutive_request_failures[0] += 1
        current_failures = _consecutive_request_failures[0]
    if current_failures >= FAILURE_LONG_REST_THRESHOLD:
        rest_sec = random.randint(FAILURE_LONG_REST_MIN_SEC, FAILURE_LONG_REST_MAX_SEC)
        logger.warning(
            "请求连续失败 %s 次，触发防封长休息，脚本将休息 %s 秒后再继续请求...",
            current_failures,
            rest_sec,
        )
        time.sleep(rest_sec)
        with _request_lock:
            _consecutive_request_failures[0] = 0


def _reset_request_failures():
    """请求成功（包括正常返回但未找到该饰品）时重置连续失败计数。"""
    with _request_lock:
        if _consecutive_request_failures[0] != 0:
            _consecutive_request_failures[0] = 0


def _record_403_and_maybe_stop():
    """
    记录一次 403 错误。
    连续 403 说明当前出口 IP 可能被风控，需要切换代理。
    当连续次数达到阈值时，执行一次短暂休息，避免过于频繁地请求。
    """
    with _request_lock:
        _consecutive_403_errors[0] += 1
        current_403 = _consecutive_403_errors[0]

    logger.error("检测到 403 Forbidden 错误（连续第 %s 次）", current_403)

    if current_403 >= FORBIDDEN_STOP_THRESHOLD:
        rest_sec = random.randint(30, 90)
        logger.warning(
            "连续检测到 %s 次 403 Forbidden，脚本将休息 %s 秒并继续尝试（期间会切换代理）。",
            current_403,
            rest_sec,
        )
        time.sleep(rest_sec)


def _reset_403_errors():
    """请求成功（含正常业务失败）时重置 403 计数。"""
    with _request_lock:
        if _consecutive_403_errors[0] != 0:
            _consecutive_403_errors[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}")
            logger.info(f"连接池配置: maxconnections={DB_POOL_CONFIG['maxconnections']}, mincached={DB_POOL_CONFIG['mincached']}")
        except Exception as e:
            logger.error(f"数据库连接池初始化失败: {e}")
            raise
    
    def get_connection(self):
        """从连接池获取连接"""
        try:
            conn = self._pool.connection()
            return conn
        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)
            
            if fetch_all:
                return cursor.fetchall()
            else:
                return cursor.fetchone()
                
        except pymysql.Error as e:
            logger.error(f"数据库查询失败: {e}, SQL: {sql}")
            raise
        finally:
            if cursor:
                cursor.close()
            # 注意：这里不需要关闭conn，连接池会自动管理
    
    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)
            
            affected_rows = cursor.rowcount
            return affected_rows
            
        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:
                    pass
            return False
        finally:
            if cursor:
                cursor.close()
    
    def get_goods_without_attr(self, limit: int = BATCH_SIZE, offset: int = 0) -> List[Dict[str, Any]]:
        """
        获取没有属性信息的饰品列表（支持分页，便于每轮拉取不同批次）
        从 buff_goods_info 表中找出在 buff_goods_attr 表中不存在的饰品。
        offset 用于轮次推进：每轮取下一批，全部取完后从 0 再开始。
        """
        try:
            sql = """
            SELECT 
                g.goods_id,
                g.name,
                g.buff_url,
                g.stat_time
            FROM buff_goods_info g
            LEFT JOIN buff_goods_attr a ON g.goods_id = a.goods_id
            WHERE a.goods_id IS NULL
            ORDER BY g.goods_id ASC
            LIMIT %s OFFSET %s
            """
            results = self.execute_query(sql, (limit, offset))
            
            goods_list = []
            for row in results:
                goods_list.append({
                    'goods_id': row['goods_id'],
                    'name': row['name'],
                    'buff_url': row['buff_url'],
                    'stat_time': row['stat_time']
                })
            
            logger.info(f"找到 {len(goods_list)} 个没有属性信息的饰品")
            return goods_list
            
        except Exception as e:
            logger.error(f"查询没有属性信息的饰品失败: {e}")
            return []
    
    def _attr_row_for_insert(self, attr_data: Dict[str, Any]) -> tuple:
        """按表列顺序生成一条 INSERT 参数元组，保证每项均为标量（30 个占位符对应 30 列）"""
        row = []
        for k in INSERT_ATTR_COLUMNS:
            v = attr_data.get(k) if k != 'stat_time' else attr_data.get('stat_time', datetime.now())
            row.append(_scalar_for_db(v))
        return tuple(row)

    def batch_insert_goods_attrs(self, attr_data_list: List[Dict[str, Any]]) -> int:
        """批量插入饰品属性信息：先全部 DELETE，再 executemany INSERT，避免占位符与参数数量错位"""
        if not attr_data_list:
            return 0
        
        delete_sql = "DELETE FROM buff_goods_attr WHERE goods_id = %s"
        rows = [self._attr_row_for_insert(attr_data) for attr_data in attr_data_list]
        # 每行必须是 30 个标量，与 INSERT_ATTR_SQL 中 30 个 %s 一致
        assert len(INSERT_ATTR_COLUMNS) == 30, "INSERT_ATTR_COLUMNS 须为 30 项"
        
        try:
            conn = None
            cursor = None
            try:
                conn = self.pool.get_connection()
                cursor = conn.cursor()
                conn.begin()
                # 先删除
                for attr_data in attr_data_list:
                    gid = _scalar_for_db(attr_data['goods_id'])
                    cursor.execute(delete_sql, (gid,))
                # 再批量插入
                cursor.executemany(INSERT_ATTR_SQL, rows)
                conn.commit()
                logger.info(f"成功批量插入 {len(attr_data_list)} 个饰品的属性信息")
                return len(attr_data_list)
            except Exception as e:
                logger.exception("事务执行失败")
                if conn:
                    try:
                        conn.rollback()
                    except Exception:
                        pass
                return 0
            finally:
                if cursor:
                    cursor.close()
        except Exception as e:
            logger.error(f"批量插入饰品属性信息异常: {e}")
            return 0
    
    def insert_goods_attr(self, attr_data: Dict[str, Any]) -> bool:
        """插入单个饰品属性信息"""
        try:
            delete_sql = "DELETE FROM buff_goods_attr WHERE goods_id = %s"
            row = self._attr_row_for_insert(attr_data)
            operations = [
                (delete_sql, (_scalar_for_db(attr_data['goods_id']),)),
                (INSERT_ATTR_SQL, row),
            ]
            success = self.execute_transaction(operations)
            if success:
                logger.debug(f"成功插入饰品 {attr_data['goods_id']} 的属性信息")
            else:
                logger.error(f"插入饰品 {attr_data['goods_id']} 属性信息失败")
            
            return success
            
        except Exception as e:
            logger.error(f"插入饰品 {attr_data.get('goods_id', '未知')} 属性信息异常: {e}")
            return False
    
    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:
    """BUFF饰品API操作类（代理请求，403/429 自动换 IP）"""
    
    def __init__(self):
        self.session = requests.Session()
        self._proxy_lock = threading.Lock()
        self._current_proxy: Optional[str] = None
        self._supports_batch_request: Optional[bool] = None
        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"
        })
        self._rotate_proxy("初始化")

    def _fetch_new_proxy_from_pool(self) -> Optional[str]:
        """从代理池接口获取新代理，格式: ip:port"""
        try:
            resp = requests.get(PROXY_POOL_API_URL, timeout=PROXY_POOL_TIMEOUT)
            resp.raise_for_status()
            proxy_text = (resp.text or "").strip()
            if not proxy_text:
                logger.error("代理池返回为空")
                return None
            # 兼容多行返回，只取第一行
            proxy_line = proxy_text.splitlines()[0].strip()
            if ":" not in proxy_line:
                logger.error("代理池返回格式不正确: %s", proxy_line)
                return None
            return proxy_line
        except Exception as e:
            logger.error("获取代理失败: %s", e)
            return None

    def _rotate_proxy(self, reason: str) -> bool:
        """切换当前代理"""
        new_proxy = self._fetch_new_proxy_from_pool()
        if not new_proxy:
            logger.error("切换代理失败（原因：%s）", reason)
            return False
        with self._proxy_lock:
            old_proxy = self._current_proxy
            self._current_proxy = new_proxy
        logger.info("代理已切换（%s）: %s -> %s", reason, old_proxy or "None", new_proxy)
        return True

    def _get_current_proxies(self) -> Optional[Dict[str, str]]:
        with self._proxy_lock:
            proxy = self._current_proxy
        if not proxy:
            return None
        proxy_url = f"http://{proxy}"
        return {"http": proxy_url, "https": proxy_url}
    
    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:
                logger.debug(f"请求饰品 {goods_id} (第{retry + 1}次尝试)...")
                
                proxies = self._get_current_proxies()
                if not proxies and not self._rotate_proxy("请求前代理为空"):
                    _record_request_failure()
                    if retry < MAX_RETRIES - 1:
                        time.sleep(RETRY_DELAY * (retry + 1))
                        continue
                    return None
                proxies = self._get_current_proxies()

                # 使用 requests.get 避免多线程共享 Session，保证并发安全
                response = requests.get(
                    url,
                    headers=self.session.headers,
                    proxies=proxies,
                    timeout=REQUEST_TIMEOUT,
                )
                if response.status_code in (403, 429):
                    status = response.status_code
                    logger.warning("饰品 %s 请求返回 %s，准备切换代理后重试", goods_id, status)
                    if status == 403:
                        _record_403_and_maybe_stop()
                    else:
                        _record_request_failure()
                    rotated = self._rotate_proxy(f"goods_id={goods_id} 触发{status}")
                    if rotated:
                        if status == 403:
                            _reset_403_errors()
                    if retry < MAX_RETRIES - 1:
                        time.sleep(RETRY_DELAY * (retry + 1))
                        continue
                    logger.error("饰品 %s 因连续 %s 最终失败", goods_id, status)
                    return None
                response.raise_for_status()
                
                data = response.json()
                
                # 检查API响应状态
                if data.get("code") != "OK":
                    logger.warning(f"饰品 {goods_id} API返回错误: {data.get('msg', '未知错误')}")
                    _record_request_failure()
                    return None
                
                # 提取饰品信息
                goods_info = self._extract_goods_info(data, goods_id)
                if goods_info:
                    _reset_request_failures()
                    _reset_403_errors()
                    logger.info(f"成功获取饰品 {goods_id} 的属性信息")
                    return goods_info
                else:
                    # 请求成功但未找到该饰品，视为正常请求，重置失败计数
                    _reset_request_failures()
                    _reset_403_errors()
                    logger.warning(f"饰品 {goods_id} 未找到属性信息")
                    return None
                    
            except requests.exceptions.RequestException as e:
                logger.error(f"请求饰品 {goods_id} 失败 (尝试 {retry + 1}/{MAX_RETRIES}): {e}")
                # 如果是 HTTP 403/429，切换代理并重试
                try:
                    response = getattr(e, "response", None)
                    if response is not None and response.status_code in (403, 429):
                        status = response.status_code
                        if status == 403:
                            _record_403_and_maybe_stop()
                        else:
                            _record_request_failure()
                        rotated = self._rotate_proxy(f"goods_id={goods_id} RequestException-{status}")
                        if rotated and status == 403:
                            _reset_403_errors()
                    else:
                        _record_request_failure()
                except Exception:
                    _record_request_failure()
                if retry < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY * (retry + 1))
                else:
                    logger.error(f"饰品 {goods_id} 请求失败，已重试{MAX_RETRIES}次")
                    return None
            except json.JSONDecodeError as e:
                logger.error(f"饰品 {goods_id} 响应JSON解析失败: {e}")
                _record_request_failure()
                return None
            except Exception as e:
                logger.error(f"处理饰品 {goods_id} 时发生未知错误: {e}")
                _record_request_failure()
                return None
        
        return None

    def _build_attr_data_from_goods_data(self, goods_id: str, goods_data: Dict[str, Any]) -> Dict[str, Any]:
        """将单个 goods_data 转为可入库结构。"""
        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

    def _extract_goods_info_batch(self, api_data: Dict[str, Any], goods_ids: List[str]) -> Dict[str, Dict[str, Any]]:
        """从响应中提取一批饰品属性信息。"""
        extracted: Dict[str, Dict[str, Any]] = {}
        try:
            data = api_data.get("data", {})
            goods_infos = data.get("goods_infos", {})
            for goods_id in goods_ids:
                goods_data = goods_infos.get(str(goods_id))
                if not goods_data:
                    continue
                extracted[str(goods_id)] = self._build_attr_data_from_goods_data(str(goods_id), goods_data)
        except Exception as e:
            logger.error(f"批量提取饰品属性失败: {e}")
        return extracted

    def _fetch_goods_info_batch_fallback(self, goods_ids: List[str]) -> Dict[str, Optional[Dict[str, Any]]]:
        """批量接口不可用时，降级为单个请求并发模式，保证任务不中断且尽量提速。"""
        logger.warning("当前批量请求不可用，降级为单个请求并发模式（本批次）")
        result: Dict[str, Optional[Dict[str, Any]]] = {}
        worker_count = max(1, min(CONCURRENT_WORKERS, len(goods_ids)))
        with concurrent.futures.ThreadPoolExecutor(max_workers=worker_count) as executor:
            future_to_id = {executor.submit(self.fetch_goods_info, str(goods_id)): str(goods_id) for goods_id in goods_ids}
            for future in concurrent.futures.as_completed(future_to_id):
                gid = future_to_id[future]
                try:
                    result[gid] = future.result()
                except Exception as e:
                    logger.error(f"降级并发模式获取饰品 {gid} 失败: {e}")
                    result[gid] = None
        return result

    def fetch_goods_info_batch(self, goods_ids: List[str]) -> Dict[str, Optional[Dict[str, Any]]]:
        """批量获取饰品属性信息（优先 1 次请求拉取多个 goods_id）。"""
        normalized_ids = [str(gid) for gid in goods_ids if str(gid).strip()]
        if not normalized_ids:
            return {}
        if len(normalized_ids) == 1:
            gid = normalized_ids[0]
            return {gid: self.fetch_goods_info(gid)}
        if self._supports_batch_request is False:
            return self._fetch_goods_info_batch_fallback(normalized_ids)

        ids_param = ",".join(normalized_ids)
        url = f"{BUFF_API_BASE_URL}?game={GAME}&goods_ids={ids_param}"

        for retry in range(MAX_RETRIES):
            try:
                logger.info("批量请求饰品属性（第%s次，数量=%s）", retry + 1, len(normalized_ids))

                proxies = self._get_current_proxies()
                if not proxies and not self._rotate_proxy("批量请求前代理为空"):
                    _record_request_failure()
                    if retry < MAX_RETRIES - 1:
                        time.sleep(RETRY_DELAY * (retry + 1))
                        continue
                    return self._fetch_goods_info_batch_fallback(normalized_ids)
                proxies = self._get_current_proxies()

                response = requests.get(
                    url,
                    headers=self.session.headers,
                    proxies=proxies,
                    timeout=REQUEST_TIMEOUT,
                )
                if response.status_code in (403, 429):
                    status = response.status_code
                    logger.warning("批量请求返回 %s，准备切换代理后重试", status)
                    if status == 403:
                        _record_403_and_maybe_stop()
                    else:
                        _record_request_failure()
                    rotated = self._rotate_proxy(f"goods_ids 批量请求触发{status}")
                    if rotated and status == 403:
                        _reset_403_errors()
                    if retry < MAX_RETRIES - 1:
                        time.sleep(RETRY_DELAY * (retry + 1))
                        continue
                    return self._fetch_goods_info_batch_fallback(normalized_ids)
                response.raise_for_status()

                data = response.json()
                if data.get("code") != "OK":
                    api_code = data.get("code")
                    api_msg = data.get("msg") or "未知错误"
                    logger.warning("批量请求 API返回错误: code=%s, msg=%s", api_code, api_msg)
                    _record_request_failure()
                    if self._supports_batch_request is None:
                        self._supports_batch_request = False
                        logger.warning("检测到批量参数 goods_ids 不可用，自动切换为单个请求并发模式")
                    return self._fetch_goods_info_batch_fallback(normalized_ids)

                extracted = self._extract_goods_info_batch(data, normalized_ids)
                if extracted:
                    self._supports_batch_request = True
                    _reset_request_failures()
                    _reset_403_errors()
                    return {gid: extracted.get(gid) for gid in normalized_ids}

                # 批量接口有可能不支持 goods_ids 参数：首轮探测不到任何目标数据时自动降级
                if self._supports_batch_request is None:
                    self._supports_batch_request = False
                    logger.warning("批量参数 goods_ids 可能不受支持，自动降级为单个请求")
                _reset_request_failures()
                _reset_403_errors()
                return self._fetch_goods_info_batch_fallback(normalized_ids)

            except requests.exceptions.RequestException as e:
                logger.error(f"批量请求失败 (尝试 {retry + 1}/{MAX_RETRIES}): {e}")
                try:
                    response = getattr(e, "response", None)
                    if response is not None and response.status_code in (403, 429):
                        status = response.status_code
                        if status == 403:
                            _record_403_and_maybe_stop()
                        else:
                            _record_request_failure()
                        rotated = self._rotate_proxy(f"goods_ids 批量RequestException-{status}")
                        if rotated and status == 403:
                            _reset_403_errors()
                    else:
                        _record_request_failure()
                except Exception:
                    _record_request_failure()
                if retry < MAX_RETRIES - 1:
                    time.sleep(RETRY_DELAY * (retry + 1))
                    continue
                return self._fetch_goods_info_batch_fallback(normalized_ids)
            except json.JSONDecodeError as e:
                logger.error(f"批量请求响应JSON解析失败: {e}")
                _record_request_failure()
                return self._fetch_goods_info_batch_fallback(normalized_ids)
            except Exception as e:
                logger.error(f"处理批量请求时发生未知错误: {e}")
                _record_request_failure()
                return self._fetch_goods_info_batch_fallback(normalized_ids)

        return self._fetch_goods_info_batch_fallback(normalized_ids)
    
    def _extract_goods_info(self, api_data: Dict[str, Any], goods_id: str) -> Optional[Dict[str, Any]]:
        """从API响应中提取饰品属性信息"""
        try:
            data = api_data.get("data", {})
            goods_infos = data.get("goods_infos", {})
            
            # 获取指定goods_id的饰品信息
            goods_data = goods_infos.get(str(goods_id))
            if not goods_data:
                logger.warning(f"饰品 {goods_id} 未在响应中找到")
                return None

            attr_data = self._build_attr_data_from_goods_data(str(goods_id), goods_data)
            logger.debug(f"饰品 {goods_id} 属性提取成功")
            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()
        # 每轮取「下一批」无属性饰品，全部取完后再从 0 开始，避免每轮都从同一批（如 40295）开始
        self._sync_offset = 0
    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]] = []

        for idx, goods_id in enumerate(goods_ids, 1):
            gid = str(goods_id)
            logger.info("处理第 %s/%s 个饰品：%s", idx, len(goods_ids), gid)
            try:
                attr_data = self.api.fetch_goods_info(gid)
            except Exception as e:
                logger.error(f"获取饰品 {gid} 属性异常: {e}")
                attr_data = None

            if attr_data:
                batch_attr_data.append(attr_data)
                logger.info(f"获取饰品 {gid} 属性成功")
            else:
                results['failed'] += 1
                logger.warning(f"获取饰品 {gid} 属性失败")

        if batch_attr_data:
            logger.info(f"本批次共获取 {len(batch_attr_data)} 条属性，即将写入数据库 {DB_HOST}/{DB_NAME} 表 buff_goods_attr ...")
            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 run_single_batch(self) -> bool:
        """执行单次批次同步任务"""
        start_time = datetime.now()
        logger.info("=" * 60)
        logger.info(f"开始执行饰品属性同步任务: {start_time}")
        logger.info("=" * 60)
        
        try:
            # 检查数据库连接
            if not self.db.check_connection():
                logger.error("数据库连接不可用，任务中止")
                return False
            
            # Step 1: 获取需要同步的饰品列表（本轮使用当前 offset，下一轮会用下一批）
            logger.info("[Step 1/3] 查询需要同步的饰品...")
            goods_list = self.db.get_goods_without_attr(BATCH_SIZE, self._sync_offset)
            
            if not goods_list:
                logger.info("没有需要同步的饰品（本轮回零 offset，下一轮从头开始）")
                self._sync_offset = 0
                return True
            
            logger.info(f"找到 {len(goods_list)} 个需要同步的饰品（offset=%s）", self._sync_offset)
            
            # 提取goods_id列表
            goods_ids = [goods['goods_id'] for goods in goods_list]
            
            # Step 2: 获取并保存饰品属性信息
            logger.info("[Step 2/3] 获取并保存饰品属性信息...")
            
            results = self.fetch_and_save_goods_attrs(goods_ids)
            
            # Step 3: 任务完成统计
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()
            
            logger.info("=" * 60)
            logger.info(f"饰品属性同步任务完成!")
            logger.info(f"处理饰品总数: {results['total']}")
            logger.info(f"成功获取并保存: {results['success']}")
            logger.info(f"失败: {results['failed']}")
            logger.info(f"耗时: {duration:.2f}秒")
            logger.info("=" * 60)

            # 本批已处理，下一轮取下一批；若本批不足 BATCH_SIZE 说明已到末尾，下轮会得到 0 条并回零
            self._sync_offset += len(goods_list)
            
            return results['success'] > 0
            
        except Exception as e:
            logger.exception(f"同步任务失败: {e}")
            return False
    
    def run_continuous(self):
        """连续运行模式，每次执行后随机暂停"""
        execution_count = 0
        
        while True:
            execution_count += 1
            logger.info(f"\n{'='*60}")
            logger.info(f"开始第 {execution_count} 次执行")
            logger.info(f"{'='*60}")
            
            # 执行同步任务
            success = self.run_single_batch()
            
            if success:
                logger.info(f"第 {execution_count} 次执行成功")
            else:
                logger.warning(f"第 {execution_count} 次执行失败")
            
            # 直接进入下一轮，不再等待
    
    def cleanup(self):
        """清理资源"""
        try:
            # 关闭连接池
            DatabaseConnectionPool().close_pool()
            logger.info("资源清理完成")
        except:
            pass

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

def main():
    """主函数"""
    task = None
    try:
        logger.info("启动饰品属性同步脚本（连接池版本）...")
        logger.info(f"数据库写入目标: {DB_HOST}:{DB_PORT}/{DB_NAME} 表 buff_goods_attr（请确认您查询的是同一库）")
        logger.info(
            f"配置信息: 每次处理 {BATCH_SIZE} 个饰品, 串行单请求(无固定间隔), 并发数 {CONCURRENT_WORKERS}, 代理请求(403自动换IP), 连续运行不主动休息"
        )
        
        # 创建任务实例
        task = GoodsAttrSyncTask()
        
        # 运行连续模式
        task.run_continuous()
        
    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()