535 lines
20 KiB
Python
535 lines
20 KiB
Python
"""三国杀游戏主控制器"""
|
||
import logging
|
||
from typing import Optional, Tuple
|
||
from games.base import BaseGame
|
||
from games.sgs_game import GameState, get_game_manager
|
||
from games.sgs_core import Phase, Role
|
||
from utils.parser import CommandParser
|
||
from core.database import get_db
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SanguoshaGame(BaseGame):
|
||
"""三国杀游戏"""
|
||
|
||
def __init__(self):
|
||
"""初始化游戏"""
|
||
super().__init__()
|
||
self.manager = get_game_manager()
|
||
|
||
async def handle(self, command: str, chat_id: int, user_id: int) -> str:
|
||
"""处理游戏指令
|
||
|
||
Args:
|
||
command: 指令
|
||
chat_id: 会话ID
|
||
user_id: 用户ID
|
||
|
||
Returns:
|
||
回复消息
|
||
"""
|
||
try:
|
||
# 提取指令和参数
|
||
cmd, args = CommandParser.extract_command_args(command)
|
||
args = args.strip().lower()
|
||
|
||
# 获取用户信息
|
||
user = self.db.get_or_create_user(user_id)
|
||
username = user.get('username', f'用户{user_id}')
|
||
|
||
# 路由指令
|
||
if not args or args == 'help':
|
||
return self.get_help()
|
||
|
||
elif args == 'create' or args == '创建':
|
||
return await self._handle_create(chat_id, user_id, username)
|
||
|
||
elif args == 'join' or args == '加入':
|
||
return await self._handle_join(chat_id, user_id, username)
|
||
|
||
elif args == 'leave' or args == '离开':
|
||
return await self._handle_leave(chat_id, user_id)
|
||
|
||
elif args == 'start' or args == '开始':
|
||
return await self._handle_start(chat_id, user_id)
|
||
|
||
elif args == 'status' or args == '状态':
|
||
return await self._handle_status(chat_id, user_id)
|
||
|
||
elif args == 'players' or args == '玩家':
|
||
return await self._handle_players(chat_id)
|
||
|
||
elif args == 'hand' or args == '手牌':
|
||
return await self._handle_hand(chat_id, user_id)
|
||
|
||
elif args == 'next' or args == '下一阶段':
|
||
return await self._handle_next_phase(chat_id, user_id)
|
||
|
||
elif args.startswith('play ') or args.startswith('出牌 '):
|
||
card_name = args.split(maxsplit=1)[1]
|
||
return await self._handle_play_card(chat_id, user_id, card_name)
|
||
|
||
elif args.startswith('use ') or args.startswith('使用 '):
|
||
card_name = args.split(maxsplit=1)[1]
|
||
return await self._handle_use_card(chat_id, user_id, card_name)
|
||
|
||
elif args == 'draw' or args == '摸牌':
|
||
return await self._handle_draw(chat_id, user_id)
|
||
|
||
elif args == 'discard' or args == '弃牌':
|
||
return await self._handle_discard(chat_id, user_id)
|
||
|
||
elif args == 'cancel' or args == '取消':
|
||
return await self._handle_cancel(chat_id, user_id)
|
||
|
||
elif args == 'stats' or args == '战绩':
|
||
return await self._handle_stats(user_id)
|
||
|
||
else:
|
||
return f"❌ 未知指令: {args}\n\n使用 `.sgs help` 查看帮助"
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理三国杀指令错误: {e}", exc_info=True)
|
||
return f"❌ 处理指令出错: {str(e)}"
|
||
|
||
async def _handle_create(self, chat_id: int, user_id: int, username: str) -> str:
|
||
"""创建游戏"""
|
||
# 检查是否已有游戏
|
||
if self.manager.has_active_game(chat_id):
|
||
return "❌ 当前已有进行中的游戏"
|
||
|
||
# 创建游戏
|
||
game = self.manager.create_game(chat_id, user_id, username)
|
||
|
||
return f"""✅ 三国杀游戏已创建！
|
||
|
||
**房主**: {username}
|
||
**游戏ID**: {game.game_id}
|
||
|
||
📝 其他玩家使用 `.sgs join` 加入游戏
|
||
📝 人数达到 {game.min_players}-{game.max_players} 人后，房主使用 `.sgs start` 开始游戏
|
||
|
||
当前玩家: 1/{game.max_players}"""
|
||
|
||
async def _handle_join(self, chat_id: int, user_id: int, username: str) -> str:
|
||
"""加入游戏"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏，使用 `.sgs create` 创建游戏"
|
||
|
||
if game.is_started:
|
||
return "❌ 游戏已经开始，无法加入"
|
||
|
||
if game.get_player_by_id(user_id):
|
||
return "❌ 你已经在游戏中了"
|
||
|
||
if not game.add_player(user_id, username):
|
||
return f"❌ 加入失败，游戏已满（{game.max_players}人）"
|
||
|
||
return f"""✅ {username} 加入游戏！
|
||
|
||
当前玩家: {len(game.players)}/{game.max_players}
|
||
玩家列表: {', '.join(p.username for p in game.players)}
|
||
|
||
{f'📝 人数已满，房主可以使用 `.sgs start` 开始游戏' if len(game.players) >= game.min_players else f'📝 还需要 {game.min_players - len(game.players)} 人才能开始'}"""
|
||
|
||
async def _handle_leave(self, chat_id: int, user_id: int) -> str:
|
||
"""离开游戏"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if game.is_started:
|
||
return "❌ 游戏已经开始，无法离开"
|
||
|
||
player = game.get_player_by_id(user_id)
|
||
if not player:
|
||
return "❌ 你不在游戏中"
|
||
|
||
if not game.remove_player(user_id):
|
||
return "❌ 离开失败"
|
||
|
||
# 如果房主离开且还有其他玩家，转移房主
|
||
if user_id == game.host_id and game.players:
|
||
game.host_id = game.players[0].user_id
|
||
return f"✅ {player.username} 离开游戏，房主已转移给 {game.players[0].username}"
|
||
|
||
# 如果没有玩家了，删除游戏
|
||
if not game.players:
|
||
self.manager.remove_game(chat_id)
|
||
return f"✅ {player.username} 离开游戏，游戏已解散"
|
||
|
||
return f"✅ {player.username} 离开游戏\n\n当前玩家: {len(game.players)}/{game.max_players}"
|
||
|
||
async def _handle_start(self, chat_id: int, user_id: int) -> str:
|
||
"""开始游戏"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if user_id != game.host_id:
|
||
return "❌ 只有房主可以开始游戏"
|
||
|
||
success, message = game.start_game()
|
||
if not success:
|
||
return f"❌ {message}"
|
||
|
||
# 构建游戏开始消息
|
||
result = "## 🎮 三国杀游戏开始！\n\n"
|
||
|
||
# 显示主公信息
|
||
lord = game.lord_player
|
||
if lord:
|
||
result += f"### 👑 主公\n"
|
||
result += f"**{lord.username}** - {lord.general.name}（{lord.general.kingdom}）\n"
|
||
result += f"体力: {lord.hp}/{lord.general.max_hp}\n"
|
||
result += f"技能: {', '.join(s.name for s in lord.general.skills)}\n\n"
|
||
|
||
# 显示其他玩家信息（不显示身份）
|
||
result += f"### 👥 玩家列表\n"
|
||
for idx, player in enumerate(game.players, 1):
|
||
if player.role != Role.LORD:
|
||
result += f"{idx}. **{player.username}** - {player.general.name}（{player.general.kingdom}）\n"
|
||
result += f" 体力: {player.hp}/{player.general.max_hp}\n"
|
||
|
||
result += f"\n### 📋 游戏信息\n"
|
||
result += f"- 玩家数: {len(game.players)}\n"
|
||
result += f"- 当前回合: {game.current_player.username}\n"
|
||
result += f"- 当前阶段: {game.current_phase.value}\n\n"
|
||
|
||
result += "💡 使用 `.sgs status` 查看游戏状态\n"
|
||
result += "💡 使用 `.sgs hand` 查看手牌\n"
|
||
result += "💡 使用 `.sgs next` 进入下一阶段"
|
||
|
||
return result
|
||
|
||
async def _handle_status(self, chat_id: int, user_id: int) -> str:
|
||
"""查看游戏状态"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if not game.is_started:
|
||
return f"""## 📋 游戏状态 (准备中)
|
||
|
||
**房主**: {game.players[0].username if game.players else '无'}
|
||
**玩家数**: {len(game.players)}/{game.max_players}
|
||
**玩家列表**: {', '.join(p.username for p in game.players)}
|
||
|
||
📝 使用 `.sgs start` 开始游戏"""
|
||
|
||
# 游戏进行中
|
||
result = f"## 📋 游戏状态\n\n"
|
||
result += f"### 🕒 当前回合\n"
|
||
result += f"- 玩家: **{game.current_player.username}**\n"
|
||
result += f"- 阶段: **{game.current_phase.value}**\n"
|
||
result += f"- 回合数: {game.round_number}\n\n"
|
||
|
||
result += f"### 👥 玩家状态\n"
|
||
for idx, player in enumerate(game.players, 1):
|
||
status = "💀" if not player.is_alive else "✅"
|
||
role_display = player.role.value if player.role == Role.LORD or not player.is_alive else "???"
|
||
|
||
result += f"{idx}. {status} **{player.username}** ({role_display})\n"
|
||
result += f" 武将: {player.general.name}（{player.general.kingdom}）\n"
|
||
result += f" 体力: {'❤️' * player.hp}{'🖤' * (player.general.max_hp - player.hp)} {player.hp}/{player.general.max_hp}\n"
|
||
result += f" 手牌: {player.hand_count}张\n"
|
||
|
||
if player.equipment:
|
||
equip_list = ', '.join(f"{k}:{v.name}" for k, v in player.equipment.items())
|
||
result += f" 装备: {equip_list}\n"
|
||
|
||
result += "\n"
|
||
|
||
return result
|
||
|
||
async def _handle_players(self, chat_id: int) -> str:
|
||
"""查看玩家列表"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
result = f"## 👥 玩家列表\n\n"
|
||
for idx, player in enumerate(game.players, 1):
|
||
result += f"{idx}. **{player.username}**"
|
||
if game.is_started:
|
||
result += f" - {player.general.name}"
|
||
if player.role == Role.LORD or not player.is_alive:
|
||
result += f"（{player.role.value}）"
|
||
result += "\n"
|
||
|
||
return result
|
||
|
||
async def _handle_hand(self, chat_id: int, user_id: int) -> str:
|
||
"""查看手牌"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if not game.is_started:
|
||
return "❌ 游戏还未开始"
|
||
|
||
player = game.get_player_by_id(user_id)
|
||
if not player:
|
||
return "❌ 你不在游戏中"
|
||
|
||
if not player.is_alive:
|
||
return "❌ 你已经阵亡"
|
||
|
||
if not player.hand_cards:
|
||
return "📋 你的手牌为空"
|
||
|
||
result = f"## 🃏 你的手牌（{len(player.hand_cards)}张）\n\n"
|
||
|
||
# 按类型分组
|
||
basic_cards = [c for c in player.hand_cards if c.card_type.value == "基本牌"]
|
||
trick_cards = [c for c in player.hand_cards if c.card_type.value == "锦囊牌"]
|
||
equip_cards = [c for c in player.hand_cards if c.card_type.value == "装备牌"]
|
||
|
||
if basic_cards:
|
||
result += "### 基本牌\n"
|
||
for idx, card in enumerate(basic_cards, 1):
|
||
result += f"{idx}. {card}\n"
|
||
result += "\n"
|
||
|
||
if trick_cards:
|
||
result += "### 锦囊牌\n"
|
||
for idx, card in enumerate(trick_cards, 1):
|
||
result += f"{idx}. {card}\n"
|
||
result += "\n"
|
||
|
||
if equip_cards:
|
||
result += "### 装备牌\n"
|
||
for idx, card in enumerate(equip_cards, 1):
|
||
result += f"{idx}. {card}\n"
|
||
result += "\n💡 使用 `.sgs play 卡牌名` 出牌"
|
||
|
||
return result
|
||
|
||
async def _handle_next_phase(self, chat_id: int, user_id: int) -> str:
|
||
"""进入下一阶段"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if not game.is_started:
|
||
return "❌ 游戏还未开始"
|
||
|
||
if game.current_player.user_id != user_id:
|
||
return "❌ 不是你的回合"
|
||
|
||
# 执行阶段转换
|
||
new_phase, current_player = game.next_phase()
|
||
|
||
# 检查游戏是否结束
|
||
is_over, winner_role = game.check_game_over()
|
||
if is_over:
|
||
game.is_finished = True
|
||
game.winner_role = winner_role
|
||
return await self._handle_game_over(game)
|
||
|
||
result = f"✅ 进入下一阶段\n\n"
|
||
result += f"**当前玩家**: {current_player.username}\n"
|
||
result += f"**当前阶段**: {new_phase.value}\n\n"
|
||
|
||
# 阶段提示
|
||
if new_phase == Phase.DRAW:
|
||
result += "💡 摸牌阶段，使用 `.sgs draw` 摸牌"
|
||
elif new_phase == Phase.PLAY:
|
||
result += "💡 出牌阶段，使用 `.sgs play 卡牌名` 出牌"
|
||
elif new_phase == Phase.DISCARD:
|
||
result += "💡 弃牌阶段，使用 `.sgs discard` 弃牌"
|
||
else:
|
||
result += "请使用正确的语法"
|
||
|
||
return result
|
||
|
||
async def _handle_draw(self, chat_id: int, user_id: int) -> str:
|
||
"""摸牌"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if game.current_player.user_id != user_id:
|
||
return "❌ 不是你的回合"
|
||
|
||
if game.current_phase != Phase.DRAW:
|
||
return f"❌ 当前不是摸牌阶段（当前: {game.current_phase.value}）"
|
||
|
||
# 摸2张牌
|
||
cards = game.deck.draw(2)
|
||
game.current_player.hand_cards.extend(cards)
|
||
|
||
result = f"✅ 摸牌成功！\n\n"
|
||
result += f"摸到: {', '.join(str(c) for c in cards)}\n"
|
||
result += f"当前手牌数: {game.current_player.hand_count}\n\n"
|
||
result += "💡 使用 `.sgs next` 进入出牌阶段"
|
||
|
||
return result
|
||
|
||
async def _handle_play_card(self, chat_id: int, user_id: int, card_name: str) -> str:
|
||
"""出牌"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if game.current_player.user_id != user_id:
|
||
return "❌ 不是你的回合"
|
||
|
||
if game.current_phase != Phase.PLAY:
|
||
return f"❌ 当前不是出牌阶段（当前: {game.current_phase.value}）"
|
||
|
||
player = game.current_player
|
||
|
||
# 查找卡牌
|
||
card = None
|
||
for c in player.hand_cards:
|
||
if c.name == card_name:
|
||
card = c
|
||
break
|
||
|
||
if not card:
|
||
return f"❌ 你没有【{card_name}】这张牌"
|
||
|
||
# 简化处理：直接打出（实际游戏需要选择目标等）
|
||
player.remove_card(card)
|
||
game.deck.discard([card])
|
||
|
||
return f"✅ 使用了【{card}】\n\n💡 继续出牌或使用 `.sgs next` 进入弃牌阶段"
|
||
|
||
async def _handle_use_card(self, chat_id: int, user_id: int, card_name: str) -> str:
|
||
"""使用卡牌（同play_card）"""
|
||
return await self._handle_play_card(chat_id, user_id, card_name)
|
||
|
||
async def _handle_discard(self, chat_id: int, user_id: int) -> str:
|
||
"""弃牌"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if game.current_player.user_id != user_id:
|
||
return "❌ 不是你的回合"
|
||
|
||
if game.current_phase != Phase.DISCARD:
|
||
return f"❌ 当前不是弃牌阶段（当前: {game.current_phase.value}）"
|
||
|
||
player = game.current_player
|
||
|
||
# 检查是否需要弃牌
|
||
max_hand = player.hp
|
||
if player.hand_count <= max_hand:
|
||
return f"✅ 手牌数({player.hand_count})未超过体力值({max_hand})，无需弃牌\n\n💡 使用 `.sgs next` 结束回合"
|
||
|
||
# 简化处理：自动弃掉多余的牌
|
||
discard_count = player.hand_count - max_hand
|
||
discarded = player.hand_cards[:discard_count]
|
||
player.hand_cards = player.hand_cards[discard_count:]
|
||
game.deck.discard(discarded)
|
||
|
||
return f"✅ 弃置了 {discard_count} 张牌\n\n💡 使用 `.sgs next` 结束回合"
|
||
|
||
async def _handle_cancel(self, chat_id: int, user_id: int) -> str:
|
||
"""取消游戏"""
|
||
game = self.manager.get_game(chat_id)
|
||
if not game:
|
||
return "❌ 当前没有游戏"
|
||
|
||
if user_id != game.host_id:
|
||
return "❌ 只有房主可以取消游戏"
|
||
|
||
self.manager.remove_game(chat_id)
|
||
return "✅ 游戏已取消"
|
||
|
||
async def _handle_stats(self, user_id: int) -> str:
|
||
"""查看战绩"""
|
||
stats = self.db.get_game_stats(user_id, 'sanguosha')
|
||
|
||
if stats['total_plays'] == 0:
|
||
return "三国杀游戏记录"
|
||
|
||
win_rate = (stats['wins'] / stats['total_plays'] * 100) if stats['total_plays'] > 0 else 0
|
||
|
||
result ="战绩\n\n"
|
||
result += f"- 总局数: {stats['total_plays']}\n"
|
||
result += f"- 胜利: {stats['wins']} 次\n"
|
||
result += f"- 失败: {stats['losses']} 次\n"
|
||
result += f"- 胜率: {win_rate:.1f}%\n"
|
||
|
||
return result
|
||
|
||
async def _handle_game_over(self, game: GameState) -> str:
|
||
"""处理游戏结束"""
|
||
result = "## 🎉 游戏结束！\n\n"
|
||
|
||
if game.winner_role == Role.LORD:
|
||
result += "## 🎉公和忠臣获胜！\n\n"
|
||
winners = [p for p in game.players if p.role in [Role.LORD, Role.LOYAL]]
|
||
losers = [p for p in game.players if p.role in [Role.REBEL, Role.SPY]]
|
||
else:
|
||
result += "### ⚔️ 反贼获胜！\n\n"
|
||
winners = [p for p in game.players if p.role == Role.REBEL]
|
||
losers = [p for p in game.players if p.role in [Role.LORD, Role.LOYAL, Role.SPY]]
|
||
|
||
result += "**获胜方**:\n"
|
||
for player in winners:
|
||
result += f"- {player.username} ({player.role.value}) - {player.general.name}\n"
|
||
# 更新战绩
|
||
self.db.update_game_stats(player.user_id, 'sanguosha', win=True)
|
||
|
||
result += "\n**失败方**:\n"
|
||
for player in losers:
|
||
result += f"- {player.username} ({player.role.value}) - {player.general.name}\n"
|
||
# 更新战绩
|
||
self.db.update_game_stats(player.user_id, 'sanguosha', loss=True)
|
||
|
||
result += f"\n游戏时长: {game.round_number} 回合"
|
||
|
||
# 清理游戏
|
||
self.manager.remove_game(game.chat_id)
|
||
|
||
return result
|
||
|
||
def get_help(self) -> str:
|
||
"""获取帮助信息"""
|
||
return """## 🎮 三国杀游戏帮助
|
||
|
||
### 游戏准备
|
||
- `.sgs create` - 创建游戏房间
|
||
- `.sgs join` - 加入游戏
|
||
- `.sgs leave` - 离开游戏
|
||
- `.sgs start` - 开始游戏（房主）
|
||
- `.sgs cancel` - 取消游戏（房主）
|
||
|
||
### 游戏中
|
||
- `.sgs status` - 查看游戏状态
|
||
- `.sgs players` - 查看玩家列表
|
||
- `.sgs hand` - 查看手牌
|
||
- `.sgs next` - 进入下一阶段
|
||
- `.sgs draw` - 摸牌（摸牌阶段）
|
||
- `.sgs play 卡牌名` - 出牌（出牌阶段）
|
||
- `.sgs discard` - 弃牌（弃牌阶段）
|
||
|
||
### 其他
|
||
- `.sgs stats` - 查看个人战绩
|
||
- `.sgs help` - 显示帮助
|
||
|
||
### 游戏规则
|
||
1. **身份**: 主公、忠臣、反贼、内奸
|
||
2. **胜利条件**:
|
||
- 主公+忠臣: 消灭所有反贼和内奸
|
||
- 反贼: 击杀主公
|
||
- 内奸: 成为最后存活的人
|
||
3. **回合流程**:
|
||
- 准备阶段 → 判定阶段 → 摸牌阶段 → 出牌阶段 → 弃牌阶段 → 结束阶段
|
||
|
||
### 卡牌类型
|
||
- **基本牌**: 杀、闪、桃
|
||
- **锦囊牌**: 决斗、过河拆桥、顺手牵羊、南蛮入侵、万箭齐发等
|
||
- **装备牌**: 武器、防具、马
|
||
|
||
---
|
||
💡 提示：游戏支持 2-8 人，建议 5-8 人游戏体验最佳
|
||
"""
|
||
|