"""狼人杀游戏""" import json import random import logging import time from typing import Optional, Dict, Any, List from games.base import BaseGame from utils.parser import CommandParser from utils.message import MessageSender logger = logging.getLogger(__name__) class WerewolfGame(BaseGame): """狼人杀游戏""" def __init__(self): """初始化游戏""" super().__init__() # 角色配置:{总人数: {'wolves': 狼人数, 'seer': 预言家数, 'witch': 女巫数, 'civilians': 平民数}} self.role_configs = { 6: {'wolves': 2, 'seer': 1, 'witch': 1, 'civilians': 2}, 8: {'wolves': 2, 'seer': 1, 'witch': 1, 'civilians': 4}, 10: {'wolves': 3, 'seer': 1, 'witch': 1, 'civilians': 5}, 12: {'wolves': 4, 'seer': 1, 'witch': 1, 'civilians': 6} } self.min_players = 6 self.max_players = 12 # 阶段名称映射:{阶段代码: {'name': 中文名称, 'role': 行动角色, 'instruction': 指令说明}} self.phase_configs = { 'night_kill': { 'name': '狼人行动阶段', 'role': '狼人', 'instruction': '使用 `.werewolf 杀 ` 投票杀人' }, 'night_seer': { 'name': '预言家验人阶段', 'role': '预言家', 'instruction': '使用 `.werewolf 验 ` 查验身份' }, 'night_witch': { 'name': '女巫行动阶段', 'role': '女巫', 'instruction': '使用 `.werewolf 救 ` 救人、`.werewolf 毒 ` 毒人,或 `.werewolf 跳过` 不行动' }, 'day_discuss': { 'name': '白天讨论阶段', 'role': '所有玩家', 'instruction': '讨论昨晚的事件' } } async def handle(self, command: str, chat_id: int, user_id: int) -> str: """处理狼人杀指令 Args: command: 指令,如 ".werewolf open" 或 ".werewolf 杀 1" chat_id: 会话ID user_id: 用户ID Returns: 回复消息 """ try: # 提取参数 _, args = CommandParser.extract_command_args(command) args = args.strip() logger.debug(f"狼人杀指令解析 - command: {command}, args: {args}") # 没有参数,显示帮助 if not args: return self.get_help() # 解析参数 parts = args.split(maxsplit=1) action = parts[0].lower() logger.debug(f"狼人杀指令解析 - parts: {parts}, action: {action}") # 创建/加入/开始游戏 if action == 'open': return self._open_game(chat_id, user_id) if action == 'join': return await self._join_game(chat_id, user_id) if action == 'start': return await self._start_game(chat_id, user_id) if action == 'status': return self._show_status(chat_id) if action == 'end': return self._end_game(chat_id, user_id) # 私聊功能 if action == '狼人': # 狼人群聊 if len(parts) < 2: return "❌ 请输入要发送的内容" content = parts[1].strip() return await self._wolf_group_chat(chat_id, user_id, content) # 普通私聊: # 检查是否是数字(玩家代号) try: target_id = int(action) if len(parts) < 2: return "❌ 请输入要发送的内容" content = parts[1].strip() return await self._private_chat(chat_id, user_id, target_id, content) except ValueError: pass # 女巫跳过指令 if action in ['跳过', 'pass']: return await self._witch_pass(chat_id, user_id) # 技能指令 if action in ['杀', '验', '救', '毒', '守']: if len(parts) < 2: return f"❌ 请指定目标ID,如:`.werewolf {action} 3`" try: target_id = int(parts[1].strip()) return await self._handle_skill(chat_id, user_id, action, target_id) except ValueError: return "❌ 目标ID必须是数字" # 查看帮助 if action in ['help', '帮助']: return self.get_help() # 未识别的指令 logger.warning(f"狼人杀未识别的指令 - args: {args}") return f"❌ 未识别的指令:{args}\n\n💡 提示:输入 `.werewolf help` 查看帮助" except Exception as e: logger.error(f"处理狼人杀指令错误: {e}", exc_info=True) return f"❌ 处理指令出错: {str(e)}" def _get_game_state(self, chat_id: int = None) -> Optional[Dict]: """获取游戏状态(全局唯一,忽略chat_id) Args: chat_id: 会话ID(保留参数用于兼容性,实际忽略) Returns: 游戏状态或None """ # 查询全局唯一的狼人杀游戏(不依赖chat_id) cursor = self.db.conn.cursor() cursor.execute( "SELECT chat_id, state_data FROM game_states WHERE game_type = ? AND user_id = 0 ORDER BY updated_at DESC LIMIT 1", ('werewolf',) ) row = cursor.fetchone() if row: state_data = json.loads(row['state_data']) if row['state_data'] else None game_chat_id = row['chat_id'] if state_data: logger.debug(f"获取全局游戏状态成功: game_chat_id={game_chat_id}, status={state_data.get('status')}, phase={state_data.get('phase')}") return state_data logger.warning(f"获取游戏状态失败: 全局无进行中的游戏") return None def _save_game_state(self, chat_id: int, state_data: Dict): """保存游戏状态 Args: chat_id: 会话ID state_data: 状态数据 """ logger.debug(f"保存游戏状态: chat_id={chat_id}, status={state_data.get('status')}, phase={state_data.get('phase')}") self.db.save_game_state(chat_id, 0, 'werewolf', state_data) def _get_game_chat_id(self) -> Optional[int]: """获取全局游戏所在的chat_id Returns: 游戏所在的chat_id,如果没有游戏返回None """ cursor = self.db.conn.cursor() cursor.execute( "SELECT chat_id FROM game_states WHERE game_type = ? AND user_id = 0 ORDER BY updated_at DESC LIMIT 1", ('werewolf',) ) row = cursor.fetchone() if row: return row['chat_id'] return None def _find_player_game(self, user_id: int) -> tuple[Optional[int], Optional[Dict]]: """查找玩家参与的游戏(全局唯一游戏) Args: user_id: 玩家用户ID Returns: (游戏所在chat_id, 游戏状态数据) 或 (None, None) """ # 获取全局游戏状态 state_data = self._get_game_state() if not state_data: logger.debug(f"未找到全局游戏") return None, None # 只查找进行中的游戏 if state_data.get('status') != 'playing': logger.debug(f"全局游戏未在进行中: status={state_data.get('status')}") return None, None # 检查玩家是否在游戏中 players = state_data.get('players', []) for player in players: if player.get('user_id') == user_id: game_chat_id = self._get_game_chat_id() logger.debug(f"找到玩家 {user_id} 的游戏: chat_id={game_chat_id}") return game_chat_id, state_data logger.debug(f"玩家 {user_id} 未参与全局游戏") return None, None def _get_phase_description(self, phase: str) -> Dict[str, str]: """获取阶段描述信息 Args: phase: 阶段代码 Returns: 阶段描述字典,包含name、role、instruction """ return self.phase_configs.get(phase, { 'name': phase, 'role': '未知', 'instruction': '' }) def _get_next_phase(self, current_phase: str) -> Optional[str]: """获取下一阶段 Args: current_phase: 当前阶段代码 Returns: 下一阶段代码,如果无下一阶段则返回None """ phase_sequence = ['night_kill', 'night_seer', 'night_witch', 'day_discuss'] try: current_index = phase_sequence.index(current_phase) if current_index < len(phase_sequence) - 1: return phase_sequence[current_index + 1] except ValueError: pass return None async def _advance_phase(self, chat_id: int, state_data: Dict) -> str: """推进到下一阶段 Args: chat_id: 会话ID state_data: 游戏状态 Returns: 阶段推进提示消息 """ current_phase = state_data.get('phase') if not current_phase: return "" next_phase = self._get_next_phase(current_phase) if not next_phase: return "" # 更新阶段 state_data['phase'] = next_phase day_intro_msg = "" if next_phase == 'day_discuss': day_intro_msg = self._resolve_night_results(chat_id, state_data) else: self._save_game_state(chat_id, state_data) # 如果进入女巫阶段,且女巫有解药,通知女巫刀人情况 if next_phase == 'night_witch': # 检查女巫是否还有解药 if not state_data.get('witch_save', False): kill_target = state_data.get('kill_target') # 找到女巫并私聊通知 for p in state_data['players']: if p['role'] == 'witch' and p['alive']: if kill_target == 0: witch_info = f"## 🧪 女巫提示\n\n✨ 今晚是平安夜,没有人被刀" else: witch_info = f"## 🧪 女巫提示\n\n⚠️ 今晚 **{kill_target}号玩家** 被刀\n\n你可以选择救人、毒人或跳过" await self._send_to_player(p['user_id'], witch_info, sender="系统") break # 获取下一阶段描述 next_phase_desc = self._get_phase_description(next_phase) # 构建提示消息 round_num = state_data.get('round', 1) if next_phase.startswith('night'): time_desc = "夜晚" else: time_desc = "白天" msg = f"\n\n---\n\n## 🌙 进入下一阶段\n\n" msg += f"**{time_desc} - {next_phase_desc['name']}**\n\n" msg += f"👤 {next_phase_desc['role']}请行动:\n" msg += f"{next_phase_desc['instruction']}" if day_intro_msg: msg = f"{day_intro_msg}{msg}" return msg def _resolve_night_results(self, chat_id: int, state_data: Dict) -> str: """结算夜晚结果并生成白天播报 Args: chat_id: 会话ID state_data: 游戏状态 Returns: 白天播报文本 """ players = state_data.get('players', []) deaths_info: List[Dict[str, Any]] = [] kill_target = state_data.get('kill_target') witch_saved = state_data.get('witch_save', False) witch_poison_target = state_data.get('witch_poison') def _find_player_by_id(pid: int) -> Optional[Dict[str, Any]]: for p in players: if p.get('id') == pid: return p return None # 狼人击杀(0表示空刀) if kill_target and kill_target != 0 and not witch_saved: killed_player = _find_player_by_id(kill_target) if killed_player and killed_player.get('alive', True): killed_player['alive'] = False deaths_info.append({ 'id': killed_player.get('id'), 'name': killed_player.get('name'), 'reason': 'wolf' }) # 女巫毒人 if witch_poison_target: poisoned_player = _find_player_by_id(witch_poison_target) if poisoned_player and poisoned_player.get('alive', True): poisoned_player['alive'] = False if not any(d.get('id') == poisoned_player.get('id') for d in deaths_info): deaths_info.append({ 'id': poisoned_player.get('id'), 'name': poisoned_player.get('name'), 'reason': 'poison' }) else: for death in deaths_info: if death.get('id') == poisoned_player.get('id'): death['reason'] = 'wolf_poison' break # 保存夜晚死亡名单 state_data['night_deaths'] = deaths_info # 重置夜晚临时状态 state_data['kill_target'] = None state_data['witch_save'] = False state_data['witch_poison'] = None state_data['wolf_votes'] = {} # TODO: 如有胜负判定逻辑,可在此调用 self._save_game_state(chat_id, state_data) return self._build_day_intro_message(state_data) def _build_day_intro_message(self, state_data: Dict) -> str: """构建白天播报消息 Args: state_data: 游戏状态 Returns: 播报文本 """ deaths = state_data.get('night_deaths') or [] message = "\n\n---\n\n## ☀️ 昨夜情况\n\n" if not deaths: message += "✨ 昨夜平安,没有玩家死亡。\n" return message message += "⚠️ 昨夜死亡名单:\n\n" reason_map = { 'wolf': "被狼人击杀", 'poison': "被女巫毒杀", 'wolf_poison': "遭遇狼人击杀且被女巫再次毒杀" } for death in deaths: player_name = death.get('name') or "未知玩家" player_id = death.get('id') reason = reason_map.get(death.get('reason'), "死亡") message += f"- {player_id}号 {player_name}({reason})\n" return message def _open_game(self, chat_id: int, user_id: int) -> str: """主持人开房 Args: chat_id: 会话ID user_id: 主持人用户ID Returns: 提示消息 """ # 检查全局是否已有游戏 state_data = self._get_game_state() if state_data: if state_data['status'] == 'playing': return "⚠️ 全局已经有游戏正在进行中!" elif state_data['status'] == 'open': return "⚠️ 全局房间已存在!\n\n房间状态:等待玩家加入\n\n玩家可以输入 `.werewolf join` 加入游戏" # 创建新房间 state_data = { 'creator_id': user_id, 'status': 'open', 'players': [], 'phase': None, 'round': 0, 'wolves': [], 'kill_target': None, 'wolf_votes': {}, # 狼人投票记录 {user_id: target_id} 'seer_result': {}, 'witch_save': False, 'witch_poison': None, 'guard_target': None, 'day_dead': [], 'vote_targets': {}, 'discussed': False, 'wolf_know_each_other': False, 'night_deaths': [] } self._save_game_state(chat_id, state_data) return f"## 🐺 狼人杀房间已创建!\n\n主持人:{user_id}\n\n其他玩家可以输入 `.werewolf join` 加入游戏\n\n**要求**:6-12人,必须注册用户名和个人URL" async def _join_game(self, chat_id: int, user_id: int) -> str: """玩家加入游戏 Args: chat_id: 会话ID user_id: 用户ID Returns: 提示消息 """ # 获取游戏状态 state_data = self._get_game_state(chat_id) if not state_data: return "❌ 没有可加入的游戏!主持人需要先输入 `.werewolf open` 创建房间" if state_data['status'] != 'open': return "❌ 游戏已开始或已结束,无法加入!" # 检查是否已加入 for player in state_data['players']: if player['user_id'] == user_id: return "⚠️ 你已经在这个房间里了!" # 检查人数限制 if len(state_data['players']) >= self.max_players: return f"❌ 房间已满!最多支持{self.max_players}人" # 验证用户是否注册了用户名 user = self.db.get_or_create_user(user_id) username = user.get('username') if not username: return "❌ 请先注册用户名!使用 `.register <名称>` 注册" # 验证用户是否注册了个人URL if not self.db.has_webhook_url(user_id): return "❌ 请先注册个人URL!使用 `.register url ` 注册\n\n个人URL用于接收游戏内的私聊消息" # 加入玩家 player_id = len(state_data['players']) + 1 player_info = { 'user_id': user_id, 'name': username, 'id': player_id, 'role': None, # 稍后分配 'alive': True, 'id_label': f"{player_id}号玩家" } state_data['players'].append(player_info) self._save_game_state(chat_id, state_data) return f"✅ 加入成功!你是 **{player_id}号玩家**\n\n当前房间人数:{len(state_data['players'])}/{self.max_players}\n\n其他玩家继续输入 `.werewolf join` 加入" async def _start_game(self, chat_id: int, user_id: int) -> str: """开始游戏 Args: chat_id: 会话ID user_id: 用户ID Returns: 提示消息 """ # 获取游戏状态 state_data = self._get_game_state(chat_id) if not state_data: return "❌ 没有可开始的游戏!主持人需要先输入 `.werewolf open` 创建房间" if state_data['status'] != 'open': return "❌ 游戏已经开始了!" if user_id != state_data['creator_id']: return "❌ 只有主持人可以开始游戏!" player_count = len(state_data['players']) if player_count < self.min_players or player_count > self.max_players: return f"❌ 游戏人数必须在{self.min_players}-{self.max_players}人之间!当前人数:{player_count}" # 检查人数是否在配置中 if player_count not in self.role_configs: return f"❌ 不支持{player_count}人游戏!仅支持{list(self.role_configs.keys())}人" # 分配角色 config = self.role_configs[player_count] roles = [] # 添加狼人 for _ in range(config['wolves']): roles.append('wolf') # 添加预言家 for _ in range(config['seer']): roles.append('seer') # 添加女巫 for _ in range(config['witch']): roles.append('witch') # 添加平民 for _ in range(config['civilians']): roles.append('civilian') # 随机分配 random.shuffle(roles) # 记录狼人列表 wolves = [] for i, player in enumerate(state_data['players']): player['role'] = roles[i] player['alive'] = True if roles[i] == 'wolf': wolves.append(player['id']) state_data['wolves'] = wolves state_data['status'] = 'playing' state_data['round'] = 1 state_data['phase'] = 'night_kill' state_data['wolf_know_each_other'] = True self._save_game_state(chat_id, state_data) # 私聊发送身份 await self._send_identities(chat_id, state_data) # 获取当前阶段描述 phase_desc = self._get_phase_description('night_kill') # 公共消息 msg = f"## 🎮 游戏开始!\n\n玩家已分配身份,请查看私聊消息\n\n" msg += f"---\n\n## 🌙 第一夜 - {phase_desc['name']}\n\n" msg += f"🐺 **{phase_desc['role']}请行动**:\n" msg += f"{phase_desc['instruction']}" return msg async def _send_identities(self, chat_id: int, state_data: Dict): """发送身份信息到私聊 Args: chat_id: 会话ID state_data: 游戏状态 """ role_descriptions = { 'wolf': '🐺 **狼人**', 'seer': '🔮 **预言家**', 'witch': '🧪 **女巫**', 'civilian': '👤 **平民**' } role_details = { 'wolf': '你的技能:\n- 每晚可以投票决定杀死一个目标\n- 你认识所有狼人同伴\n- 胜利条件:杀死所有神职和平民\n\n使用 `.werewolf 杀 ` 投票杀人', 'seer': '你的技能:\n- 每晚可以查验一个玩家的身份\n- 胜利条件:消灭所有狼人\n\n使用 `.werewolf 验 ` 查验身份', 'witch': '你的技能:\n- 拥有一瓶解药和一瓶毒药\n- 每晚可以救人(只能使用一次)或毒人(只能使用一次)\n- 也可以选择跳过不行动\n- 胜利条件:消灭所有狼人\n\n使用 `.werewolf 救 ` 救人\n使用 `.werewolf 毒 ` 毒人\n使用 `.werewolf 跳过` 不行动', 'civilian': '你没有特殊技能\n\n你的胜利条件:消灭所有狼人\n\n仔细观察,通过投票帮助好人阵营获胜' } for player in state_data['players']: role = player['role'] identity_msg = f"## 🎭 你的身份信息\n\n" identity_msg += f"你是:**{player['id_label']}**\n" identity_msg += f"身份:{role_descriptions[role]}\n\n" # 狼人显示同伴 if role == 'wolf': teammates = [f"{tid}号" for tid in state_data['wolves'] if tid != player['id']] if teammates: identity_msg += f"你的狼人同伴:{', '.join(teammates)}\n\n" identity_msg += role_details[role] # 发送私聊 await self._send_to_player(player['user_id'], identity_msg, sender="系统") async def _send_to_player(self, target_user_id: int, content: str, sender: str = "系统") -> bool: """发送私聊消息给指定玩家 Args: target_user_id: 目标用户ID content: 消息内容 sender: 发送者标识 Returns: 是否发送成功 """ from utils.message import send_private_message # 添加发送者信息 if sender != "系统": content = f"**来自:{sender}**\n\n{content}" return await send_private_message(target_user_id, content, 'text') async def _private_chat(self, chat_id: int, user_id: int, target_id: int, content: str) -> str: """玩家私聊 Args: chat_id: 会话ID user_id: 发送者用户ID target_id: 目标玩家代号 content: 消息内容 Returns: 提示消息 """ # 查找玩家参与的游戏(全局唯一) _, state_data = self._find_player_game(user_id) if not state_data: return "❌ 你没有参与任何进行中的游戏!" if state_data['status'] != 'playing': return "❌ 游戏未开始或已结束!" # 查找发送者 sender = None for player in state_data['players']: if player['user_id'] == user_id: sender = player break if not sender: return "❌ 你不是游戏参与者!" if not sender['alive']: return "❌ 死亡玩家无法发送消息!" # 查找目标 target = None for player in state_data['players']: if player['id'] == target_id: target = player break if not target: return f"❌ 找不到{target_id}号玩家!" if not target['alive']: return f"❌ {target_id}号玩家已死亡!" # 发送私聊 msg = f"{sender['id_label']}对你发送消息:\n\n{content}" success = await self._send_to_player(target['user_id'], msg, sender=sender['id_label']) if success: return f"✅ 消息已发送给{target_id}号玩家" else: return "❌ 消息发送失败,请稍后重试" async def _wolf_group_chat(self, chat_id: int, user_id: int, content: str) -> str: """狼人群聊 Args: chat_id: 会话ID user_id: 发送者用户ID content: 消息内容 Returns: 提示消息 """ # 查找玩家参与的游戏(全局唯一) _, state_data = self._find_player_game(user_id) if not state_data: return "❌ 你没有参与任何进行中的游戏!" if state_data['status'] != 'playing': return "❌ 游戏未开始或已结束!" # 查找发送者 sender = None for player in state_data['players']: if player['user_id'] == user_id: sender = player break if not sender: return "❌ 你不是游戏参与者!" if sender['role'] != 'wolf': return "❌ 只有狼人可以使用这个功能!" if not sender['alive']: return "❌ 死亡玩家无法发送消息!" # 发送给所有狼人 sent_count = 0 for player in state_data['players']: if player['role'] == 'wolf' and player['user_id'] != user_id: msg = f"🐺 狼人{sender['id']}号:{content}" success = await self._send_to_player(player['user_id'], msg, sender="狼人群聊") if success: sent_count += 1 if sent_count > 0: return f"✅ 消息已发送给{len(state_data['wolves'])-1}个狼人同伴" else: return "❌ 没有其他狼人在线或发送失败" async def _handle_skill(self, chat_id: int, user_id: int, skill: str, target_id: int) -> str: """处理技能指令 Args: chat_id: 会话ID user_id: 用户ID skill: 技能类型(杀/验/救/毒) target_id: 目标玩家代号 Returns: 提示消息 """ logger.debug(f"处理技能 - chat_id: {chat_id}, user_id: {user_id}, skill: {skill}, target_id: {target_id}") # 查找玩家参与的游戏(全局唯一) game_chat_id, state_data = self._find_player_game(user_id) if not state_data: logger.warning(f"技能处理失败 - 玩家未参与任何游戏: user_id={user_id}") return "❌ 你没有参与任何进行中的游戏!" logger.debug(f"找到玩家游戏: game_chat_id={game_chat_id}, user_id={user_id}") if state_data['status'] != 'playing': return "❌ 游戏未开始或已结束!" # 查找玩家 player = None for p in state_data['players']: if p['user_id'] == user_id: player = p break if not player: return "❌ 你不是游戏参与者!" if not player['alive']: return "❌ 死亡玩家无法使用技能!" # 根据技能类型处理(使用 game_chat_id) if skill == '杀': return await self._wolf_kill(game_chat_id, state_data, player, target_id) elif skill == '验': return await self._seer_check(game_chat_id, state_data, player, target_id) elif skill == '救': return await self._witch_save(game_chat_id, state_data, player, target_id) elif skill == '毒': return await self._witch_poison(game_chat_id, state_data, player, target_id) return "❌ 未知技能" async def _wolf_kill(self, chat_id: int, state_data: Dict, player: Dict, target_id: int) -> str: """狼人刀人 Args: chat_id: 会话ID state_data: 游戏状态 player: 玩家信息 target_id: 目标代号 Returns: 提示消息 """ if player['role'] != 'wolf': return "❌ 只有狼人可以刀人!" current_phase = state_data['phase'] if not current_phase or not current_phase.startswith('night'): return "❌ 只能在夜晚使用技能!" # 0表示空刀,不验证目标 if target_id != 0: # 查找目标 target = None for p in state_data['players']: if p['id'] == target_id: target = p break if not target: return f"❌ 找不到{target_id}号玩家!" if not target['alive']: return f"❌ {target_id}号玩家已死亡!" # 检查当前阶段 if current_phase != 'night_kill': return f"❌ 当前不是狼人行动阶段!当前阶段:{self._get_phase_description(current_phase)['name']}" # 记录投票(允许改票) wolf_votes = state_data.get('wolf_votes', {}) wolf_votes = {int(k): v for k, v in wolf_votes.items()} wolf_votes[player['user_id']] = target_id state_data['wolf_votes'] = wolf_votes self._save_game_state(chat_id, state_data) # 获取所有存活的狼人 alive_wolves = [] for p in state_data['players']: if p['role'] == 'wolf' and p['alive']: alive_wolves.append(p['user_id']) # 检查是否所有狼人都已投票 voted_wolves = set(wolf_votes.keys()) all_wolves = set(alive_wolves) if not all_wolves.issubset(voted_wolves): # 还有狼人未投票,私聊通知投票者 not_voted_count = len(all_wolves - voted_wolves) if target_id == 0: vote_confirm_msg = f"## 🐺 投票确认\n\n✅ 你选择 **空刀**(不杀人)\n\n⏳ 等待其他狼人投票(还有{not_voted_count}个狼人未投票)" else: vote_confirm_msg = f"## 🐺 投票确认\n\n✅ 你投票给 **{target_id}号玩家**\n\n⏳ 等待其他狼人投票(还有{not_voted_count}个狼人未投票)" await self._send_to_player(player['user_id'], vote_confirm_msg, sender="系统") # 群消息不透露目标 return f"✅ 投票已记录,等待其他狼人(还有{not_voted_count}个未投票)" # 所有狼人都已投票,统计票数(忽略空刀) vote_count = {} 空刀_count = 0 for wolf_id, target in wolf_votes.items(): if wolf_id in alive_wolves: # 只统计存活狼人的投票 if target == 0: 空刀_count += 1 else: vote_count[target] = vote_count.get(target, 0) + 1 # 如果所有狼人都空刀,本晚无人被杀 if not vote_count: # 没有有效投票 state_data['kill_target'] = 0 # 0表示空刀 state_data['wolf_votes'] = {} self._save_game_state(chat_id, state_data) # 私聊通知所有狼人 for p in state_data['players']: if p['role'] == 'wolf' and p['alive']: 空刀_msg = f"## 🐺 狼人投票结果\n\n✅ 所有狼人都选择空刀,今晚平安夜" await self._send_to_player(p['user_id'], 空刀_msg, sender="系统") # 自动推进到下一阶段 next_phase_msg = await self._advance_phase(chat_id, state_data) return f"✅ 狼人投票完成{next_phase_msg}" # 找出票数最多的目标 max_votes = max(vote_count.values()) top_targets = [tid for tid, count in vote_count.items() if count == max_votes] # 判断是否平票 if len(top_targets) > 1: # 平票,清除投票记录,要求重新投票 state_data['wolf_votes'] = {} self._save_game_state(chat_id, state_data) # 私聊通知所有狼人平票结果 targets_str = '、'.join([f"{tid}号" for tid in top_targets]) for p in state_data['players']: if p['role'] == 'wolf' and p['alive']: tie_msg = f"## 🐺 狼人投票结果\n\n⚠️ 投票结果平票!\n\n票数最多的玩家:{targets_str}(各{max_votes}票)\n\n请重新投票" await self._send_to_player(p['user_id'], tie_msg, sender="系统") # 群消息不透露具体目标 return f"⚠️ 狼人投票结果平票,请重新投票" # 票数唯一,确定击杀目标 kill_target = top_targets[0] state_data['kill_target'] = kill_target state_data['wolf_votes'] = {} # 清空投票记录 self._save_game_state(chat_id, state_data) # 私聊通知所有狼人投票结果 for p in state_data['players']: if p['role'] == 'wolf' and p['alive']: vote_msg = f"## 🐺 狼人投票结果\n\n✅ 投票完成:决定刀 **{kill_target}号玩家**({max_votes}票)" await self._send_to_player(p['user_id'], vote_msg, sender="系统") # 自动推进到下一阶段 next_phase_msg = await self._advance_phase(chat_id, state_data) # 群消息不透露击杀目标 return f"✅ 狼人投票完成{next_phase_msg}" async def _seer_check(self, chat_id: int, state_data: Dict, player: Dict, target_id: int) -> str: """预言家验人 Args: chat_id: 会话ID state_data: 游戏状态 player: 玩家信息 target_id: 目标代号 Returns: 提示消息 """ if player['role'] != 'seer': return "❌ 只有预言家可以验人!" current_phase = state_data['phase'] if not current_phase or not current_phase.startswith('night'): return "❌ 只能在夜晚使用技能!" # 检查当前阶段 if current_phase != 'night_seer': return f"❌ 当前不是预言家验人阶段!当前阶段:{self._get_phase_description(current_phase)['name']}" # 查找目标 target = None for p in state_data['players']: if p['id'] == target_id: target = p break if not target: return f"❌ 找不到{target_id}号玩家!" # 记录验人结果 result = 'wolf' if target['role'] == 'wolf' else 'good' state_data['seer_result'][player['id']] = {'target': target_id, 'result': result} self._save_game_state(chat_id, state_data) # 私聊告诉预言家结果 result_text = "🐺 狼人" if result == 'wolf' else "✅ 好人" msg = f"## 🔮 验人结果\n\n{target_id}号玩家的身份是:**{result_text}**" await self._send_to_player(player['user_id'], msg, sender="系统") # 自动推进到下一阶段 next_phase_msg = await self._advance_phase(chat_id, state_data) return f"✅ 验人成功!已私聊发送结果{next_phase_msg}" async def _witch_save(self, chat_id: int, state_data: Dict, player: Dict, target_id: int) -> str: """女巫救人 Args: chat_id: 会话ID state_data: 游戏状态 player: 玩家信息 target_id: 目标代号 Returns: 提示消息 """ if player['role'] != 'witch': return "❌ 只有女巫可以救人!" if state_data.get('witch_save', False): return "❌ 解药已使用!" current_phase = state_data['phase'] if not current_phase or not current_phase.startswith('night'): return "❌ 只能在夜晚使用技能!" # 检查当前阶段 if current_phase != 'night_witch': return f"❌ 当前不是女巫行动阶段!当前阶段:{self._get_phase_description(current_phase)['name']}" # 查找目标 target = None for p in state_data['players']: if p['id'] == target_id: target = p break if not target: return f"❌ 找不到{target_id}号玩家!" # 记录救人 state_data['witch_save'] = True self._save_game_state(chat_id, state_data) # 自动推进到下一阶段 next_phase_msg = await self._advance_phase(chat_id, state_data) return f"✅ 救人成功:救了{target_id}号玩家{next_phase_msg}" async def _witch_poison(self, chat_id: int, state_data: Dict, player: Dict, target_id: int) -> str: """女巫毒人 Args: chat_id: 会话ID state_data: 游戏状态 player: 玩家信息 target_id: 目标代号 Returns: 提示消息 """ if player['role'] != 'witch': return "❌ 只有女巫可以毒人!" if state_data.get('witch_poison') is not None: return "❌ 毒药已使用!" current_phase = state_data['phase'] if not current_phase or not current_phase.startswith('night'): return "❌ 只能在夜晚使用技能!" # 检查当前阶段 if current_phase != 'night_witch': return f"❌ 当前不是女巫行动阶段!当前阶段:{self._get_phase_description(current_phase)['name']}" # 查找目标 target = None for p in state_data['players']: if p['id'] == target_id: target = p break if not target: return f"❌ 找不到{target_id}号玩家!" if not target['alive']: return f"❌ {target_id}号玩家已死亡!" # 记录毒人 state_data['witch_poison'] = target_id self._save_game_state(chat_id, state_data) # 自动推进到下一阶段 next_phase_msg = self._advance_phase(chat_id, state_data) return f"✅ 毒人成功:毒了{target_id}号玩家{next_phase_msg}" async def _witch_pass(self, chat_id: int, user_id: int) -> str: """女巫跳过(不行动) Args: chat_id: 会话ID user_id: 用户ID Returns: 提示消息 """ # 查找玩家参与的游戏(全局唯一) game_chat_id, state_data = self._find_player_game(user_id) if not state_data: return "❌ 你没有参与任何进行中的游戏!" if state_data['status'] != 'playing': return "❌ 游戏未开始或已结束!" # 查找玩家 player = None for p in state_data['players']: if p['user_id'] == user_id: player = p break if not player: return "❌ 你不是游戏参与者!" if player['role'] != 'witch': return "❌ 只有女巫可以跳过!" if not player['alive']: return "❌ 死亡玩家无法使用技能!" current_phase = state_data['phase'] if current_phase != 'night_witch': return f"❌ 当前不是女巫行动阶段!当前阶段:{self._get_phase_description(current_phase)['name']}" # 自动推进到下一阶段(使用 game_chat_id) next_phase_msg = await self._advance_phase(game_chat_id, state_data) return f"✅ 女巫选择跳过,不行动{next_phase_msg}" def _show_status(self, chat_id: int) -> str: """显示游戏状态 Args: chat_id: 会话ID Returns: 状态消息 """ state_data = self._get_game_state(chat_id) if not state_data: return "❌ 没有正在进行的游戏!" status = state_data['status'] if status == 'open': players = state_data.get('players', []) msg = f"## 🐺 狼人杀房间\n\n状态:等待玩家加入\n人数:{len(players)}/{self.max_players}\n" if players: msg += "\n当前玩家:\n" for player in players: player_name = player.get('name') or "未知玩家" msg += f"⚪ {player['id']}号 {player_name}\n" else: msg += "\n当前房间暂无玩家\n" msg += "\n输入 `.werewolf join` 加入游戏" return msg if status != 'playing': return "❌ 游戏未开始或已结束!" # 获取当前阶段描述 current_phase = state_data.get('phase', 'unknown') phase_desc = self._get_phase_description(current_phase) # 显示玩家状态(只显示存活状态,不显示身份) msg = f"## 🎮 游戏进行中\n\n" msg += f"**回合**:第{state_data['round']}回合\n" msg += f"**当前阶段**:{phase_desc['name']}\n" msg += f"**行动角色**:{phase_desc['role']}\n" if phase_desc['instruction']: msg += f"**操作指令**:{phase_desc['instruction']}\n" msg += f"\n**玩家状态**:\n" alive_count = 0 dead_count = 0 for player in state_data['players']: player_name = player.get('name') or "未知玩家" status_icon = "❤️" if player['alive'] else "💀" msg += f"{status_icon} {player['id']}号 {player_name}\n" if player['alive']: alive_count += 1 else: dead_count += 1 msg += f"\n存活:{alive_count}人 死亡:{dead_count}人" return msg def _end_game(self, chat_id: int, user_id: int) -> str: """结束游戏(主持人专用) Args: chat_id: 会话ID user_id: 用户ID Returns: 提示消息 """ state_data = self._get_game_state(chat_id) if not state_data: return "❌ 没有正在进行的游戏!" if state_data['status'] == 'open': return "❌ 游戏尚未开始,无法结束!" if user_id != state_data['creator_id']: return "❌ 只有主持人可以结束游戏!" state_data['status'] = 'finished' self._save_game_state(chat_id, state_data) return "✅ 游戏已结束" def get_help(self) -> str: """获取帮助信息 Returns: 帮助文本 """ help_text = """## 🐺 狼人杀游戏帮助 ### 基础操作 - `.werewolf open` - 主持人创建房间 - `.werewolf join` - 加入游戏(需要注册用户名和个人URL) - `.werewolf start` - 主持人开始游戏 - `.werewolf status` - 查看游戏状态 - `.werewolf end` - 主持人结束游戏 ### 私聊功能 - `.werewolf <玩家代号> <消息>` - 私聊指定玩家 - `.werewolf 狼人 <消息>` - 狼人群聊(仅狼人) ### 技能指令 - `.werewolf 杀 ` - 狼人投票刀人 - `.werewolf 验 ` - 预言家验人 - `.werewolf 救 ` - 女巫救人 - `.werewolf 毒 ` - 女巫毒人 - `.werewolf 跳过` - 女巫跳过(不行动) ### 游戏规则 **人数配置**: - 6人:2狼 1预言家 1女巫 2平民 - 8人:2狼 1预言家 1女巫 4平民 - 10人:3狼 1预言家 1女巫 5平民 - 12人:4狼 1预言家 1女巫 6平民 **胜利条件**: - 狼人阵营:杀死所有神职和平民 - 好人阵营:消灭所有狼人 **游戏流程**: 1. 主持人创建房间 2. 玩家加入游戏 3. 主持人开始游戏,玩家收到身份 4. 夜晚阶段:狼人刀人、预言家验人、女巫用药 5. 白天阶段:死亡公示、讨论发言、投票放逐 --- 💡 提示:使用私聊功能前必须先注册用户名和个人URL """ return help_text