# static_db.py import sqlite3 import json import threading import time import queue import logging from typing import List, Dict, Any, Optional logger = logging.getLogger(__name__) # ==================== 静态数据内存缓存 ==================== class StaticDataCache: """单例，存放从 game.db 加载的静态配置（只读）""" _instance = None _lock = threading.Lock() def __new__(cls): with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._initialized = True self._load_all() def _load_all(self): """从 game.db 加载所有静态数据到内存""" conn = sqlite3.connect('game.db') conn.row_factory = sqlite3.Row try: # 1. 装备（items 表，非药水） rows = conn.execute("SELECT * FROM items WHERE type != 'potion'").fetchall() equipment = { "weapons": [], "armors": [], "helmets": [], "necklaces": [], "bracelets": [], "rings": [], "belts": [], "shoes": [], "medals": [], "specialRings": [] } type_map = { "weapon": "weapons", "armor": "armors", "helmet": "helmets", "necklace": "necklaces", "bracelet": "bracelets", "ring": "rings", "belt": "belts", "shoes": "shoes", "medal": "medals", "special_ring": "specialRings" } for row in rows: item = dict(row) cat = type_map.get(item['type']) if not cat: continue converted = { 'id': item['id'], 'icon': item.get('icon', ''), 'name': item['name'], 'type': item['type'], 'level': item['level'], 'attackMin': item['attack_min'], 'attackMax': item['attack_max'], 'defenseMin': item['defense_min'], 'defenseMax': item['defense_max'], 'price': item['price'], 'quality': item['quality'], 'specialEffect': item['special_effect'], 'special': item.get('special', ''), 'description': item['description'] } if item['type'] == 'special_ring': equipment["specialRings"].append(converted) elif cat: equipment[cat].append(converted) self.equipment = equipment # 2. 药水（items 表，type='potion'） rows = conn.execute("SELECT * FROM items WHERE type = 'potion'").fetchall() potions = [] for row in rows: p = dict(row) potions.append({ 'id': p['id'], 'icon': p.get('icon', ''), 'name': p['name'], 'type': 'potion', 'price': p['price'], 'effectType': p['effect_type'], 'effectValue': p['effect_value'], 'maxStack': p['max_stack'], 'level': p['level'], 'hp': p.get('hp_recover', p['effect_value'] if p['effect_type'] == 'hp' else 0), 'mp': p.get('mp_recover', 0), 'description': p['description'] }) self.potions = potions # 3. 技能（skills 表） rows = conn.execute("SELECT * FROM skills").fetchall() player_skills = [] hero_skills = [] for row in rows: s = dict(row) skill = { 'id': s['id'], 'name': s['name'], 'description': s['description'], 'level': s['level_required'], 'damageMultiplier': s['damage_multiplier'], 'freezeChance': s['freeze_chance'], 'freezeTurns': s['freeze_turns'], 'targetType': s['target_type'], 'mp_cost': s.get('mp_cost', 0) } if s['is_hero']: hero_skills.append(skill) else: player_skills.append(skill) self.skills = {'playerSkills': player_skills, 'heroSkills': hero_skills} # 4. 怪物（monsters 表） rows = conn.execute("SELECT * FROM monsters").fetchall() monsters = [] for row in rows: m = dict(row) monsters.append({ 'id': m['id'], 'name': m['name'], 'type': m['type'], 'level': m['level'], 'hp': m['hp'], 'maxHp': m['hp'], 'attackMin': m['attack_min'], 'attackMax': m['attack_max'], 'defenseMin': m['defense_min'], 'defenseMax': m['defense_max'], 'exp': m['exp'], 'gold': m['gold'], 'isBoss': bool(m['is_boss']), 'isWorldBoss': bool(m['is_world_boss']), 'spawnRate': m['spawn_rate'], 'skills': json.loads(m['skills']) if m['skills'] else [], 'icon': m.get('icon', '') }) self.monsters = monsters finally: conn.close() logger.info("静态数据从 game.db 加载完成") def get_equipment(self) -> Dict: return self.equipment def get_potions(self) -> List: return self.potions def get_skills(self) -> Dict: return self.skills def get_monsters(self) -> List: return self.monsters def reload(self): """重新加载静态数据（管理员调用）""" self._load_all() # ==================== 数据库连接池管理 ==================== class DatabasePool: """管理多个数据库的连接池（静态数据用 game.db，动态数据用 chat.db）""" def __init__(self): self.pools = {} # {db_name: queue.Queue} self.configs = { 'static': {'path': 'game.db', 'maxsize': 20}, # 静态数据（只读） 'game': {'path': 'chat.db', 'maxsize': 100}, # 玩家动态数据 'chat': {'path': 'chatroom.db', 'maxsize': 50} } self._init_pools() def _init_pools(self): for name, cfg in self.configs.items(): q = queue.Queue(maxsize=cfg['maxsize']) for _ in range(cfg['maxsize']): conn = sqlite3.connect(cfg['path'], timeout=30, check_same_thread=False) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.execute("PRAGMA cache_size=10000") conn.execute("PRAGMA busy_timeout=10000") q.put(conn) self.pools[name] = q logger.info(f"数据库连接池 [{name}] 初始化完成，大小 {cfg['maxsize']}，路径 {cfg['path']}") def get_connection(self, db_name: str): if db_name not in self.pools: raise ValueError(f"未知数据库: {db_name}") pool = self.pools[db_name] max_retries = 3 for attempt in range(max_retries): try: conn = pool.get(timeout=10) conn.rollback() conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") return conn except queue.Empty: if attempt == max_retries - 1: raise Exception(f"数据库 {db_name} 连接池繁忙") time.sleep(1) def return_connection(self, db_name: str, conn): if conn and db_name in self.pools: try: conn.rollback() conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") self.pools[db_name].put(conn) except Exception as e: logger.error(f"归还连接失败 {db_name}: {e}") cfg = self.configs[db_name] new_conn = sqlite3.connect(cfg['path'], timeout=30, check_same_thread=False) new_conn.execute("PRAGMA journal_mode=WAL") new_conn.execute("PRAGMA synchronous=NORMAL") new_conn.execute("PRAGMA cache_size=10000") new_conn.execute("PRAGMA busy_timeout=10000") self.pools[db_name].put(new_conn) def execute(self, db_name: str, sql: str, params=(), fetchone=False, fetchall=False): conn = self.get_connection(db_name) try: cursor = conn.execute(sql, params) if fetchone: row = cursor.fetchone() return dict(row) if row else None if fetchall: return [dict(r) for r in cursor.fetchall()] conn.commit() return cursor.lastrowid except Exception: conn.rollback() raise finally: self.return_connection(db_name, conn) def execute_many(self, db_name: str, sql: str, seq_of_params): conn = self.get_connection(db_name) try: cursor = conn.cursor() cursor.executemany(sql, seq_of_params) conn.commit() return cursor.rowcount except Exception: conn.rollback() raise finally: self.return_connection(db_name, conn) def transaction(self, db_name: str, func): conn = self.get_connection(db_name) try: result = func(conn) conn.commit() return result except Exception: conn.rollback() raise finally: self.return_connection(db_name, conn) # ==================== 统一对外接口 ==================== class UnifiedDB: """单例，整合静态缓存 + 动态连接池""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.cache = StaticDataCache() cls._instance.pool = DatabasePool() return cls._instance # ---- 静态数据接口（从 game.db 内存缓存）---- def get_equipment(self): return self.cache.get_equipment() def get_potions(self): return self.cache.get_potions() def get_skills(self): return self.cache.get_skills() def get_monsters(self): return self.cache.get_monsters() def reload_static(self): self.cache.reload() # ---- 动态数据库操作（chat.db）---- def execute_game(self, sql, params=(), fetchone=False, fetchall=False): return self.pool.execute('game', sql, params, fetchone, fetchall) def execute_chat(self, sql, params=(), fetchone=False, fetchall=False): return self.pool.execute('chat', sql, params, fetchone, fetchall) def transaction_game(self, func): return self.pool.transaction('game', func) def transaction_chat(self, func): return self.pool.transaction('chat', func) # 获取原始连接（特殊场景） def get_game_conn(self): return self.pool.get_connection('game') def return_game_conn(self, conn): self.pool.return_connection('game', conn) def get_static_conn(self): return self.pool.get_connection('static') def return_static_conn(self, conn): self.pool.return_connection('static', conn) # 全局单例实例 db = UnifiedDB()