#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ BUFF饰品属性信息获取脚本（使用连接池版本） 特点： 1. 使用DBUtils连接池管理数据库连接 2. 自动处理连接失效和重连 3. 更稳定的长连接管理 4. 避免频繁创建和关闭连接的开销 5. 通过代理请求接口，403 自动切换代理重试 每轮从「尚无属性」的饰品中随机取一批（ORDER BY RAND()），默认每批 100 条，处理完后休息 45 分钟再取下一批。 可直接长期运行：python sync_goods_attr.py 可选环境变量：BATCH_SIZE（默认 100）、DB_FLUSH_EVERY（默认每 10 条成功获取即写库一次）、REST_SECONDS（批间休息秒数，默认 2700）、GOODS_ATTR_RUN_ONCE=1 仅跑一批后退出。 安装依赖： pip install DBUtils==3.0.2 """ import os import random import sys import concurrent.futures 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") # 与 sync_goods.py 一致：大批量 SQL 时显式放宽客户端读写超时（未设置时 PyMySQL 为 None，无此 30s 问题） DB_READ_TIMEOUT = int(os.getenv("DB_READ_TIMEOUT", "1800")) DB_WRITE_TIMEOUT = int(os.getenv("DB_WRITE_TIMEOUT", "1800")) # 连接池配置 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, 'connect_timeout': 10, 'read_timeout': DB_READ_TIMEOUT, 'write_timeout': DB_WRITE_TIMEOUT, '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", "100")) # 每轮随机拉取的饰品数量，处理完一批后休息 REST_SECONDS REST_SECONDS = int(os.getenv("REST_SECONDS", "2700")) # 每批结束后休息秒数（默认 45 分钟），降低 429 风险 DB_FLUSH_EVERY = max(1, int(os.getenv("DB_FLUSH_EVERY", "10"))) # 每成功获取多少条就提交一次数据库 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 后直接停止脚本的阈值 # 设为 1/true/yes 时只执行一批（处理条数由 BATCH_SIZE 决定）后退出，供外层 shell 控制休息间隔 GOODS_ATTR_RUN_ONCE = os.getenv("GOODS_ATTR_RUN_ONCE", "").strip().lower() in ("1", "true", "yes") # ============================================================ # 配置日志 # ============================================================ 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} " f"(read_timeout={DB_READ_TIMEOUT}s, write_timeout={DB_WRITE_TIMEOUT}s)" ) 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() # PooledDB：必须 close 才会把连接还回池子；否则池会耗尽并出现“连不上库” if conn: conn.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) affected_rows = cursor.rowcount return affected_rows except pymysql.Error as e: logger.error(f"数据库更新失败: {e}, SQL: {sql}") raise finally: if cursor: cursor.close() if conn: conn.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() if conn: conn.close() def get_goods_without_attr(self, limit: int = BATCH_SIZE) -> List[Dict[str, Any]]: """ 获取没有属性信息的饰品列表。 从 buff_goods_info 中找出在 buff_goods_attr 中不存在的饰品，按 RAND() 随机取 limit 条， 避免每次重启脚本都按 goods_id 固定顺序扫同一批。 已成功写入的属性下一轮不再出现在结果集中，无需 OFFSET 分页。 """ 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 RAND() LIMIT %s """ results = self.execute_query(sql, (limit,)) 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() if conn: conn.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() def fetch_and_save_goods_attrs(self, goods_ids: List[str]) -> Dict[str, int]: """获取并保存饰品属性信息；每成功积累 DB_FLUSH_EVERY 条即提交一次数据库。""" results = {'success': 0, 'failed': 0, 'total': len(goods_ids)} pending: List[Dict[str, Any]] = [] def flush_pending(chunk: List[Dict[str, Any]]) -> None: if not chunk: return logger.info( "提交 %s 条属性到 %s/%s.buff_goods_attr（每 %s 条一批）...", len(chunk), DB_HOST, DB_NAME, DB_FLUSH_EVERY, ) saved_count = self.db.batch_insert_goods_attrs(chunk) results['success'] += saved_count if saved_count < len(chunk): results['failed'] += len(chunk) - saved_count 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: pending.append(attr_data) logger.info(f"获取饰品 {gid} 属性成功") while len(pending) >= DB_FLUSH_EVERY: flush_pending(pending[:DB_FLUSH_EVERY]) del pending[:DB_FLUSH_EVERY] else: results['failed'] += 1 logger.warning(f"获取饰品 {gid} 属性失败") if pending: flush_pending(pending) 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: 获取需要同步的饰品列表（随机一批，避免每次执行顺序固定） logger.info("[Step 1/3] 查询需要同步的饰品...") goods_list = self.db.get_goods_without_attr(BATCH_SIZE) if not goods_list: logger.info("没有需要同步的饰品") return True logger.info("找到 %s 个需要同步的饰品（本批随机抽取）", len(goods_list)) # 提取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) return results['success'] > 0 except Exception as e: logger.exception(f"同步任务失败: {e}") return False def run_continuous(self): """连续运行：每批处理 BATCH_SIZE 条（随机抽取），批间休息 REST_SECONDS 秒。""" execution_count = 0 while True: execution_count += 1 logger.info(f"\n{'='*60}") logger.info(f"开始第 {execution_count} 轮（每轮最多 {BATCH_SIZE} 条，批间休息 {REST_SECONDS}s）") logger.info(f"{'='*60}") success = self.run_single_batch() if success: logger.info(f"第 {execution_count} 轮执行结束（有成功写入或无待同步项视为正常）") else: logger.warning(f"第 {execution_count} 轮执行未写入成功记录，仍按间隔休息后继续") logger.info( "本轮结束，休息 %s 秒（约 %.1f 分钟）后继续下一轮...", REST_SECONDS, REST_SECONDS / 60.0, ) time.sleep(REST_SECONDS) 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（请确认您查询的是同一库）") mode = "单批退出(GOODS_ATTR_RUN_ONCE)" if GOODS_ATTR_RUN_ONCE else f"连续运行（批间 {REST_SECONDS}s）" logger.info( "配置信息: 模式=%s, 每轮随机最多 %s 个饰品, 每 %s 条提交 DB 一次, 批间休息 %s 秒, 串行单请求, 并发数 %s, 代理请求(403自动换IP)", mode, BATCH_SIZE, DB_FLUSH_EVERY, REST_SECONDS, CONCURRENT_WORKERS, ) # 创建任务实例 task = GoodsAttrSyncTask() if GOODS_ATTR_RUN_ONCE: try: ok = task.run_single_batch() finally: task.cleanup() sys.exit(0 if ok else 1) 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()