""" 房间管理器模块 负责管理多个游戏房间或场景实例 """ import time import threading import uuid from typing import Dict, Any, List, Optional class RoomManager: """ 房间管理器 负责管理多个游戏房间或场景实例 """ def __init__(self, plugin): """ 初始化房间管理器 Args: plugin: 服务器架构插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 房间配置 self.room_config = { "max_rooms": 100, "max_clients_per_room": 50, "enable_room_persistence": False, "room_timeout": 3600, # 1小时无活动房间自动关闭 "enable_room_passwords": True, "enable_room_visibility": True, "default_room_settings": { "max_clients": 10, "is_public": True, "allow_spectators": True, "enable_voice_chat": True } } # 房间存储 self.rooms = {} # 房间字典 self.client_rooms = {} # 客户端所在房间映射 # 房间统计 self.room_stats = { "total_rooms": 0, "active_rooms": 0, "clients_in_rooms": 0, "rooms_created": 0, "rooms_destroyed": 0, "clients_joined": 0, "clients_left": 0, "errors": 0 } # 线程锁 self.rooms_lock = threading.RLock() # 回调函数 self.room_callbacks = { "room_created": [], "room_destroyed": [], "client_joined": [], "client_left": [], "room_updated": [] } # 时间戳记录 self.last_cleanup = 0.0 self.last_room_check = 0.0 print("✓ 房间管理器已创建") def initialize(self) -> bool: """ 初始化房间管理器 Returns: 是否初始化成功 """ try: print("正在初始化房间管理器...") self.initialized = True print("✓ 房间管理器初始化完成") return True except Exception as e: print(f"✗ 房间管理器初始化失败: {e}") self.room_stats["errors"] += 1 import traceback traceback.print_exc() return False def enable(self) -> bool: """ 启用房间管理器 Returns: 是否启用成功 """ try: if not self.initialized: print("✗ 房间管理器未初始化") return False self.enabled = True print("✓ 房间管理器已启用") return True except Exception as e: print(f"✗ 房间管理器启用失败: {e}") self.room_stats["errors"] += 1 import traceback traceback.print_exc() return False def disable(self): """禁用房间管理器""" try: self.enabled = False # 关闭所有房间 self.destroy_all_rooms() print("✓ 房间管理器已禁用") except Exception as e: print(f"✗ 房间管理器禁用失败: {e}") self.room_stats["errors"] += 1 import traceback traceback.print_exc() def finalize(self): """清理房间管理器资源""" try: # 禁用管理器 if self.enabled: self.disable() # 清理回调 self.room_callbacks.clear() self.initialized = False print("✓ 房间管理器资源已清理") except Exception as e: print(f"✗ 房间管理器资源清理失败: {e}") import traceback traceback.print_exc() def update(self, dt: float): """ 更新房间管理器状态 Args: dt: 时间增量（秒） """ try: if not self.enabled: return current_time = time.time() # 定期清理过期房间 if current_time - self.last_cleanup > 60.0: # 每分钟清理一次 self._cleanup_expired_rooms(current_time) self.last_cleanup = current_time except Exception as e: print(f"✗ 房间管理器更新失败: {e}") self.room_stats["errors"] += 1 import traceback traceback.print_exc() def _cleanup_expired_rooms(self, current_time: float): """清理过期房间""" try: expired_rooms = [] with self.rooms_lock: for room_id, room_data in self.rooms.items(): # 检查房间是否超时 if current_time - room_data["last_activity"] > self.room_config["room_timeout"]: # 检查房间是否为空 if len(room_data["clients"]) == 0: expired_rooms.append(room_id) # 销毁过期房间 for room_id in expired_rooms: self.destroy_room(room_id) except Exception as e: print(f"✗ 过期房间清理失败: {e}") self.room_stats["errors"] += 1 def create_room(self, room_name: str = None, room_settings: Dict[str, Any] = None, room_password: str = None) -> Optional[str]: """ 创建房间 Args: room_name: 房间名称 room_settings: 房间设置 room_password: 房间密码 Returns: 房间ID或None """ try: if not self.enabled: return None # 检查房间数量限制 with self.rooms_lock: if len(self.rooms) >= self.room_config["max_rooms"]: print("✗ 房间数量已达上限") return None # 生成房间ID room_id = str(uuid.uuid4()) # 使用默认设置或传入设置 settings = self.room_config["default_room_settings"].copy() if room_settings: settings.update(room_settings) # 创建房间数据 room_data = { "room_id": room_id, "room_name": room_name or f"房间-{room_id[:8]}", "settings": settings, "password": room_password, "clients": {}, # 客户端ID到客户端数据的映射 "host_client_id": None, "created_time": time.time(), "last_activity": time.time(), "room_state": {}, # 房间状态数据 "custom_data": {} # 自定义房间数据 } # 添加到房间列表 with self.rooms_lock: self.rooms[room_id] = room_data self.room_stats["total_rooms"] += 1 self.room_stats["active_rooms"] += 1 self.room_stats["rooms_created"] += 1 # 触发房间创建回调 self._trigger_room_callback("room_created", { "room_id": room_id, "room_name": room_data["room_name"], "settings": settings, "timestamp": room_data["created_time"] }) print(f"✓ 房间已创建: {room_id} ({room_data['room_name']})") return room_id except Exception as e: print(f"✗ 房间创建失败: {e}") self.room_stats["errors"] += 1 return None def destroy_room(self, room_id: str) -> bool: """ 销毁房间 Args: room_id: 房间ID Returns: 是否销毁成功 """ try: if not self.enabled: return False # 检查房间是否存在 with self.rooms_lock: if room_id not in self.rooms: print(f"✗ 房间不存在: {room_id}") return False room_data = self.rooms[room_id] # 通知房间内所有客户端 for client_id in list(room_data["clients"].keys()): self._notify_client_room_destroyed(client_id, room_id) # 移除房间 del self.rooms[room_id] self.room_stats["active_rooms"] -= 1 self.room_stats["rooms_destroyed"] += 1 # 触发房间销毁回调 self._trigger_room_callback("room_destroyed", { "room_id": room_id, "room_name": room_data["room_name"], "timestamp": time.time() }) print(f"✓ 房间已销毁: {room_id}") return True except Exception as e: print(f"✗ 房间销毁失败: {e}") self.room_stats["errors"] += 1 return False def destroy_all_rooms(self): """销毁所有房间""" try: with self.rooms_lock: room_ids = list(self.rooms.keys()) for room_id in room_ids: self.destroy_room(room_id) except Exception as e: print(f"✗ 销毁所有房间失败: {e}") self.room_stats["errors"] += 1 def add_client_to_room(self, room_id: str, client_id: str, client_data: Dict[str, Any] = None) -> bool: """ 添加客户端到房间 Args: room_id: 房间ID client_id: 客户端ID client_data: 客户端数据 Returns: 是否添加成功 """ try: if not self.enabled: return False # 检查客户端是否已认证 if not self.plugin.client_manager.is_client_authenticated(client_id): print(f"✗ 客户端未认证: {client_id}") return False # 检查房间是否存在 with self.rooms_lock: if room_id not in self.rooms: print(f"✗ 房间不存在: {room_id}") return False room_data = self.rooms[room_id] # 检查房间是否已满 if len(room_data["clients"]) >= room_data["settings"]["max_clients"]: print(f"✗ 房间已满: {room_id}") return False # 检查房间密码 if room_data["password"] and not self._verify_room_password(room_id, client_data): print(f"✗ 房间密码错误: {room_id}") return False # 添加客户端到房间 client_info = client_data or {} client_info["join_time"] = time.time() client_info["last_activity"] = time.time() room_data["clients"][client_id] = client_info room_data["last_activity"] = time.time() # 设置房间主机（如果是第一个客户端） if room_data["host_client_id"] is None: room_data["host_client_id"] = client_id # 更新客户端房间映射 self.client_rooms[client_id] = room_id self.room_stats["clients_in_rooms"] += 1 self.room_stats["clients_joined"] += 1 # 触发客户端加入回调 self._trigger_room_callback("client_joined", { "room_id": room_id, "room_name": room_data["room_name"], "client_id": client_id, "timestamp": time.time() }) print(f"✓ 客户端已加入房间: {client_id} -> {room_id}") return True except Exception as e: print(f"✗ 添加客户端到房间失败: {e}") self.room_stats["errors"] += 1 return False def remove_client_from_room(self, room_id: str, client_id: str) -> bool: """ 从房间移除客户端 Args: room_id: 房间ID client_id: 客户端ID Returns: 是否移除成功 """ try: if not self.enabled: return False # 检查房间是否存在 with self.rooms_lock: if room_id not in self.rooms: print(f"✗ 房间不存在: {room_id}") return False room_data = self.rooms[room_id] # 检查客户端是否在房间中 if client_id not in room_data["clients"]: print(f"✗ 客户端不在房间中: {client_id}") return False # 移除客户端 del room_data["clients"][client_id] room_data["last_activity"] = time.time() # 更新主机（如果移除的是主机） if room_data["host_client_id"] == client_id: if room_data["clients"]: # 选择第一个客户端作为新主机 room_data["host_client_id"] = next(iter(room_data["clients"])) else: room_data["host_client_id"] = None # 移除客户端房间映射 if client_id in self.client_rooms: del self.client_rooms[client_id] self.room_stats["clients_in_rooms"] -= 1 self.room_stats["clients_left"] += 1 # 触发客户端离开回调 self._trigger_room_callback("client_left", { "room_id": room_id, "room_name": room_data["room_name"], "client_id": client_id, "timestamp": time.time() }) # 如果房间为空，销毁房间 if len(room_data["clients"]) == 0: self.destroy_room(room_id) print(f"✓ 客户端已离开房间: {client_id} <- {room_id}") return True except Exception as e: print(f"✗ 从房间移除客户端失败: {e}") self.room_stats["errors"] += 1 return False def _verify_room_password(self, room_id: str, client_data: Dict[str, Any]) -> bool: """ 验证房间密码 Args: room_id: 房间ID client_data: 客户端数据 Returns: 密码是否正确 """ try: if not self.room_config["enable_room_passwords"]: return True with self.rooms_lock: if room_id not in self.rooms: return False room_data = self.rooms[room_id] if not room_data["password"]: return True client_password = client_data.get("password") if client_data else None return client_password == room_data["password"] except Exception as e: print(f"✗ 房间密码验证失败: {e}") self.room_stats["errors"] += 1 return False def _notify_client_room_destroyed(self, client_id: str, room_id: str): """ 通知客户端房间被销毁 Args: client_id: 客户端ID room_id: 房间ID """ try: # 从客户端房间映射中移除 if client_id in self.client_rooms: del self.client_rooms[client_id] # 发送房间销毁通知 if self.plugin.message_router: destroy_message = { "type": "room_destroyed", "room_id": room_id, "timestamp": time.time() } client_data = self.plugin.client_manager.get_client(client_id) if client_data and "socket" in client_data: self.plugin.message_router.send_message( destroy_message, client_data["socket"], client_id ) except Exception as e: print(f"✗ 通知客户端房间销毁失败: {e}") self.room_stats["errors"] += 1 def get_room_info(self, room_id: str) -> Optional[Dict[str, Any]]: """ 获取房间信息 Args: room_id: 房间ID Returns: 房间信息或None """ try: with self.rooms_lock: if room_id in self.rooms: room_data = self.rooms[room_id] # 返回不包含敏感信息的房间信息 safe_room_data = { "room_id": room_data["room_id"], "room_name": room_data["room_name"], "settings": room_data["settings"], "client_count": len(room_data["clients"]), "has_password": bool(room_data["password"]), "host_client_id": room_data["host_client_id"], "created_time": room_data["created_time"], "last_activity": room_data["last_activity"] } return safe_room_data else: return None except Exception as e: print(f"✗ 获取房间信息失败: {e}") self.room_stats["errors"] += 1 return None def get_all_rooms(self) -> List[Dict[str, Any]]: """ 获取所有房间信息 Returns: 房间信息列表 """ try: rooms_info = [] with self.rooms_lock: for room_id, room_data in self.rooms.items(): # 只返回公开房间或用户所在的房间 if room_data["settings"].get("is_public", True): safe_room_data = { "room_id": room_data["room_id"], "room_name": room_data["room_name"], "settings": room_data["settings"], "client_count": len(room_data["clients"]), "has_password": bool(room_data["password"]), "host_client_id": room_data["host_client_id"], "created_time": room_data["created_time"], "last_activity": room_data["last_activity"] } rooms_info.append(safe_room_data) return rooms_info except Exception as e: print(f"✗ 获取所有房间信息失败: {e}") self.room_stats["errors"] += 1 return [] def get_client_room(self, client_id: str) -> Optional[str]: """ 获取客户端所在房间 Args: client_id: 客户端ID Returns: 房间ID或None """ try: return self.client_rooms.get(client_id) except Exception as e: print(f"✗ 获取客户端房间失败: {e}") self.room_stats["errors"] += 1 return None def get_room_clients(self, room_id: str) -> List[str]: """ 获取房间内所有客户端 Args: room_id: 房间ID Returns: 客户端ID列表 """ try: with self.rooms_lock: if room_id in self.rooms: return list(self.rooms[room_id]["clients"].keys()) else: return [] except Exception as e: print(f"✗ 获取房间客户端失败: {e}") self.room_stats["errors"] += 1 return [] def update_room_state(self, room_id: str, state_data: Dict[str, Any]) -> bool: """ 更新房间状态 Args: room_id: 房间ID state_data: 状态数据 Returns: 是否更新成功 """ try: with self.rooms_lock: if room_id not in self.rooms: return False self.rooms[room_id]["room_state"].update(state_data) self.rooms[room_id]["last_activity"] = time.time() # 触发房间更新回调 self._trigger_room_callback("room_updated", { "room_id": room_id, "state_data": state_data, "timestamp": time.time() }) return True except Exception as e: print(f"✗ 更新房间状态失败: {e}") self.room_stats["errors"] += 1 return False def broadcast_to_room(self, room_id: str, message_data: Dict[str, Any], exclude_client_id: str = None) -> bool: """ 向房间内所有客户端广播消息 Args: room_id: 房间ID message_data: 消息数据 exclude_client_id: 要排除的客户端ID Returns: 是否广播成功 """ try: if not self.enabled: return False with self.rooms_lock: if room_id not in self.rooms: return False room_data = self.rooms[room_id] success_count = 0 # 向房间内每个客户端发送消息 for client_id in room_data["clients"]: # 跳过排除的客户端 if client_id == exclude_client_id: continue # 获取客户端数据 client_data = self.plugin.client_manager.get_client(client_id) if client_data and "socket" in client_data: # 发送消息 if self.plugin.message_router: send_success = self.plugin.message_router.send_message( message_data, client_data["socket"], client_id ) if send_success: success_count += 1 return success_count > 0 except Exception as e: print(f"✗ 房间广播失败: {e}") self.room_stats["errors"] += 1 return False def set_room_custom_data(self, room_id: str, key: str, value: Any) -> bool: """ 设置房间自定义数据 Args: room_id: 房间ID key: 数据键 value: 数据值 Returns: 是否设置成功 """ try: with self.rooms_lock: if room_id not in self.rooms: return False self.rooms[room_id]["custom_data"][key] = value return True except Exception as e: print(f"✗ 设置房间自定义数据失败: {e}") self.room_stats["errors"] += 1 return False def get_room_custom_data(self, room_id: str, key: str, default: Any = None) -> Any: """ 获取房间自定义数据 Args: room_id: 房间ID key: 数据键 default: 默认值 Returns: 数据值或默认值 """ try: with self.rooms_lock: if room_id not in self.rooms: return default return self.rooms[room_id]["custom_data"].get(key, default) except Exception as e: print(f"✗ 获取房间自定义数据失败: {e}") self.room_stats["errors"] += 1 return default def get_room_stats(self) -> Dict[str, Any]: """ 获取房间统计信息 Returns: 房间统计字典 """ return { "stats": self.room_stats.copy(), "config": self.room_config.copy(), "current_rooms": len(self.rooms), "current_clients_in_rooms": len(self.client_rooms) } def reset_stats(self): """重置房间统计信息""" try: self.room_stats = { "total_rooms": 0, "active_rooms": 0, "clients_in_rooms": 0, "rooms_created": 0, "rooms_destroyed": 0, "clients_joined": 0, "clients_left": 0, "errors": 0 } print("✓ 房间统计信息已重置") except Exception as e: print(f"✗ 房间统计信息重置失败: {e}") def set_room_config(self, config: Dict[str, Any]) -> bool: """ 设置房间配置 Args: config: 房间配置字典 Returns: 是否设置成功 """ try: self.room_config.update(config) print(f"✓ 房间配置已更新: {self.room_config}") return True except Exception as e: print(f"✗ 房间配置设置失败: {e}") return False def get_room_config(self) -> Dict[str, Any]: """ 获取房间配置 Returns: 房间配置字典 """ return self.room_config.copy() def _trigger_room_callback(self, callback_type: str, data: Dict[str, Any]): """ 触发房间回调 Args: callback_type: 回调类型 data: 回调数据 """ try: if callback_type in self.room_callbacks: for callback in self.room_callbacks[callback_type]: try: callback(data) except Exception as e: print(f"✗ 房间回调执行失败: {callback_type} - {e}") except Exception as e: print(f"✗ 房间回调触发失败: {e}") def register_room_callback(self, callback_type: str, callback: callable): """ 注册房间回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.room_callbacks: self.room_callbacks[callback_type].append(callback) print(f"✓ 房间回调已注册: {callback_type}") else: print(f"✗ 无效的回调类型: {callback_type}") except Exception as e: print(f"✗ 房间回调注册失败: {e}") def unregister_room_callback(self, callback_type: str, callback: callable): """ 注销房间回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.room_callbacks: if callback in self.room_callbacks[callback_type]: self.room_callbacks[callback_type].remove(callback) print(f"✓ 房间回调已注销: {callback_type}") else: print(f"✗ 无效的回调类型: {callback_type}") except Exception as e: print(f"✗ 房间回调注销失败: {e}")