""" GameBase 游戏基类 定义游戏的通用接口和状态管理 """ from abc import ABC, abstractmethod from typing import Dict, List, Any, Optional from enum import Enum import time import json class GamePhase(Enum): """游戏阶段枚举""" WAITING = "waiting" # 等待玩家 READY = "ready" # 准备开始 STARTING = "starting" # 开始中 PLAYING = "playing" # 游戏中 PAUSED = "paused" # 暂停 ENDED = "ended" # 已结束 class GameState: """ 游戏状态类 管理游戏的所有状态数据 """ def __init__(self, game_type: str, room_id: str): self.game_type = game_type self.room_id = room_id self.phase = GamePhase.WAITING self.round = 0 self.turn_count = 0 self.current_turn_player_id: Optional[str] = None self.winner_id: Optional[str] = None self.started_at: Optional[float] = None self.ended_at: Optional[float] = None # 玩家状态 self.player_states: Dict[str, Dict] = {} # 游戏特定数据 self.data: Dict[str, Any] = {} # 历史记录 self.history: List[Dict] = [] def start(self): """开始游戏""" self.phase = GamePhase.PLAYING self.round = 1 self.started_at = time.time() def end(self, winner_id: Optional[str] = None): """结束游戏""" self.phase = GamePhase.ENDED self.winner_id = winner_id self.ended_at = time.time() def next_turn(self, player_id: str): """切换回合""" self.current_turn_player_id = player_id self.turn_count += 1 def add_history(self, action: str, player_id: str, data: Dict): """添加历史记录""" self.history.append({ 'turn': self.turn_count, 'action': action, 'player_id': player_id, 'data': data, 'timestamp': time.time() }) def set_player_state(self, player_id: str, state: Dict): """设置玩家状态""" self.player_states[player_id] = state def get_player_state(self, player_id: str) -> Optional[Dict]: """获取玩家状态""" return self.player_states.get(player_id) def to_dict(self, player_id: Optional[str] = None) -> Dict[str, Any]: """ 序列化为字典 player_id: 如果提供,包含该玩家的私有信息 """ return { 'game_type': self.game_type, 'room_id': self.room_id, 'phase': self.phase.value, 'round': self.round, 'turn_count': self.turn_count, 'current_turn': self.current_turn_player_id, 'winner': self.winner_id, 'started_at': self.started_at, 'ended_at': self.ended_at, 'player_states': self.player_states, 'data': self.data, 'history': self.history[-20:] if len(self.history) > 20 else self.history, } def to_json(self) -> str: """序列化为 JSON""" return json.dumps(self.to_dict(), default=str) class GameBase(ABC): """ 游戏基类 所有具体游戏必须继承此类 """ def __init__(self, room_id: str, game_type: str): self.room_id = room_id self.game_type = game_type self.state = GameState(game_type, room_id) self.players: Dict[str, Any] = {} # player_id -> Player self.player_order: List[str] = [] # 玩家顺序 self.is_started = False self.is_ended = False @property @abstractmethod def min_players(self) -> int: """最少玩家数""" pass @property @abstractmethod def max_players(self) -> int: """最多玩家数""" pass @property def player_count(self) -> int: """当前玩家数""" return len(self.players) def add_player(self, player): """添加玩家""" if self.player_count >= self.max_players: raise ValueError("房间已满") self.players[player.id] = player if player.id not in self.player_order: self.player_order.append(player.id) self.state.set_player_state(player.id, { 'name': player.name, 'is_ai': player.is_ai, 'seat': player.seat_number, }) def remove_player(self, player_id: str): """移除玩家""" if player_id in self.players: del self.players[player_id] if player_id in self.player_order: self.player_order.remove(player_id) @abstractmethod def can_start(self) -> bool: """检查是否可以开始游戏""" pass @abstractmethod def start(self): """开始游戏""" pass @abstractmethod def handle_action(self, player_id: str, action: str, data: Dict) -> Dict: """ 处理玩家操作 返回: {'success': bool, 'message': str, 'state_update': Dict} """ pass @abstractmethod def get_valid_actions(self, player_id: str) -> List[str]: """获取玩家当前可用的操作列表""" pass @abstractmethod def check_winner(self) -> Optional[str]: """检查是否有获胜者""" pass def get_current_player(self) -> Optional[Any]: """获取当前回合玩家""" if self.state.current_turn_player_id: return self.players.get(self.state.current_turn_player_id) return None def next_player(self) -> Optional[Any]: """切换到下一个玩家""" if not self.player_order: return None current_id = self.state.current_turn_player_id if not current_id or current_id not in self.player_order: next_id = self.player_order[0] else: current_idx = self.player_order.index(current_id) next_idx = (current_idx + 1) % len(self.player_order) next_id = self.player_order[next_idx] self.state.next_turn(next_id) return self.players.get(next_id) def end_game(self, winner_id: Optional[str] = None): """结束游戏""" self.is_ended = True self.state.end(winner_id) def to_dict(self) -> Dict[str, Any]: """序列化游戏状态""" return { 'room_id': self.room_id, 'game_type': self.game_type, 'min_players': self.min_players, 'max_players': self.max_players, 'player_count': self.player_count, 'is_started': self.is_started, 'is_ended': self.is_ended, 'state': self.state.to_dict(), 'players': { pid: p.to_dict() for pid, p in self.players.items() } }