新增成语接龙游戏

2025-10-28 16:12:32 +08:00
parent ff3d9cc343
commit c761a12377
7 changed files with 871 additions and 1 deletions
--- a/games/idiom.py
+++ b/games/idiom.py
@@ -0,0 +1,626 @@
+"""成语接龙游戏"""
+import random
+import logging
+import time
+import re
+from typing import Optional, Tuple, Dict, Any
+from pypinyin import pinyin, Style
+from games.base import BaseGame
+from utils.parser import CommandParser
+from config import GAME_CONFIG
+
+logger = logging.getLogger(__name__)
+
+
+class IdiomGame(BaseGame):
+    """成语接龙游戏"""
+    
+    def __init__(self):
+        """初始化游戏"""
+        super().__init__()
+        self.config = GAME_CONFIG.get('idiom', {})
+        self.max_history_display = self.config.get('max_history_display', 10)
+        self.starter_idioms = self.config.get('starter_idioms', [
+            "一马当先", "龙马精神", "马到成功", "开门见山"
+        ])
+    
+    async def handle(self, command: str, chat_id: int, user_id: int) -> str:
+        """处理成语接龙指令
+        
+        Args:
+            command: 指令，如 ".idiom start" 或 ".idiom 马到成功"
+            chat_id: 会话ID
+            user_id: 用户ID
+            
+        Returns:
+            回复消息
+        """
+        try:
+            # 提取参数
+            _, args = CommandParser.extract_command_args(command)
+            args = args.strip()
+            
+            # 没有参数，显示帮助
+            if not args:
+                return self.get_help()
+            
+            # 解析参数
+            parts = args.split(maxsplit=1)
+            action = parts[0].lower()
+            
+            # 开始游戏
+            if action in ['start', '开始']:
+                starter = parts[1].strip() if len(parts) > 1 else None
+                return self._start_game(chat_id, user_id, starter)
+            
+            # 结束游戏
+            if action in ['stop', '结束', 'end']:
+                return self._stop_game(chat_id, user_id)
+            
+            # 查看状态
+            if action in ['status', '状态']:
+                return self._show_status(chat_id)
+            
+            # 查看黑名单
+            if action in ['blacklist', '黑名单']:
+                return self._show_blacklist(chat_id)
+            
+            # 查看帮助
+            if action in ['help', '帮助']:
+                return self.get_help()
+            
+            # 裁判拒绝
+            if action in ['reject', '拒绝']:
+                if len(parts) < 2:
+                    return "❌ 请指定要拒绝的词语，如：`.idiom reject 词语`"
+                idiom_to_reject = parts[1].strip()
+                return self._reject_idiom(chat_id, user_id, idiom_to_reject)
+            
+            # 指定下一位
+            if action in ['next', '下一位']:
+                if len(parts) < 2:
+                    return "❌ 请@要指定的用户"
+                mentioned = self._parse_mentioned_user(args)
+                if not mentioned:
+                    return "❌ 未识别到@的用户"
+                return self._set_next_user(chat_id, user_id, mentioned)
+            
+            # 默认：接龙
+            # 整个args都是成语（可能包含@用户）
+            return self._make_chain(chat_id, user_id, args)
+            
+        except Exception as e:
+            logger.error(f"处理成语接龙指令错误: {e}", exc_info=True)
+            return f"❌ 处理指令出错: {str(e)}"
+    
+    def _get_pinyin(self, char: str, all_readings: bool = True) -> list:
+        """获取单字拼音
+        
+        Args:
+            char: 单个汉字
+            all_readings: 是否返回多音字的所有读音
+            
+        Returns:
+            拼音列表
+        """
+        try:
+            result = pinyin(char, style=Style.NORMAL, heteronym=all_readings)
+            return result[0] if result else []
+        except Exception as e:
+            logger.error(f"获取拼音错误: {e}")
+            return []
+    
+    def _check_pinyin_match(self, last_char: str, first_char: str) -> Tuple[bool, str, str]:
+        """检查拼音匹配
+        
+        Args:
+            last_char: 上一个成语的最后一个字
+            first_char: 当前成语的第一个字
+            
+        Returns:
+            (是否匹配, 上一个拼音, 当前拼音)
+        """
+        last_pinyins = self._get_pinyin(last_char)
+        first_pinyins = self._get_pinyin(first_char)
+        
+        # 只要有任何一个读音匹配就算成功
+        for lp in last_pinyins:
+            for fp in first_pinyins:
+                if lp.lower() == fp.lower():
+                    return True, lp, fp
+        
+        # 没有匹配，返回第一个读音
+        return False, last_pinyins[0] if last_pinyins else '', first_pinyins[0] if first_pinyins else ''
+    
+    def _parse_mentioned_user(self, content: str) -> Optional[int]:
+        """解析消息中@的用户ID
+        
+        Args:
+            content: 消息内容
+            
+        Returns:
+            用户ID或None
+        """
+        # 简化实现：查找@后的数字
+        # 实际WPS API可能有特定格式，需要根据文档调整
+        match = re.search(r'@.*?(\d+)', content)
+        if match:
+            try:
+                return int(match.group(1))
+            except ValueError:
+                pass
+        return None
+    
+    def _can_answer(self, state_data: Dict, user_id: int) -> Tuple[bool, str]:
+        """检查用户是否可以接龙
+        
+        Args:
+            state_data: 游戏状态数据
+            user_id: 用户ID
+            
+        Returns:
+            (是否可以, 错误消息)
+        """
+        # 不能是上一个接龙的人
+        if state_data.get('last_user_id') == user_id:
+            return False, "❌ 不能连续接龙哦！让其他人来吧"
+        
+        # 如果指定了下一位，必须是指定的人
+        if state_data.get('next_user_id') is not None:
+            if state_data['next_user_id'] != user_id:
+                return False, f"❌ 现在轮到指定的人接龙了"
+        
+        return True, ""
+    
+    def _validate_idiom(self, idiom: str, state_data: Dict) -> Tuple[bool, str]:
+        """验证词语有效性
+        
+        Args:
+            idiom: 待验证的词语
+            state_data: 游戏状态数据
+            
+        Returns:
+            (是否有效, 错误消息)
+        """
+        # 检查长度
+        if len(idiom) != 4:
+            return False, "❌ 词语必须是4个字"
+        
+        # 检查是否已使用
+        if idiom in state_data.get('used_idioms', []):
+            return False, f"❌ 「{idiom}」已经用过了"
+        
+        # 检查是否在黑名单
+        if idiom in state_data.get('blacklist', []):
+            return False, f"❌ 「{idiom}」在黑名单中"
+        
+        # 检查拼音匹配
+        current_idiom = state_data.get('current_idiom', '')
+        if current_idiom:
+            last_char = current_idiom[-1]
+            first_char = idiom[0]
+            is_match, last_py, first_py = self._check_pinyin_match(last_char, first_char)
+            
+            if not is_match:
+                return False, f"❌ 首字「{first_char}」拼音[{first_py}]不匹配上个成语尾字「{last_char}」拼音[{last_py}]"
+        
+        return True, ""
+    
+    def _start_game(self, chat_id: int, user_id: int, starter_idiom: Optional[str]) -> str:
+        """开始新游戏
+        
+        Args:
+            chat_id: 会话ID
+            user_id: 用户ID
+            starter_idiom: 起始成语（可选）
+            
+        Returns:
+            提示消息
+        """
+        # 检查是否已有进行中的游戏（user_id=0表示群级别状态）
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if state:
+            return "⚠️ 已经有一个进行中的游戏了！\n\n输入 `.idiom stop` 结束当前游戏"
+        
+        # 确定起始成语
+        if starter_idiom:
+            # 验证起始成语
+            if len(starter_idiom) != 4:
+                return "❌ 起始成语必须是4个字"
+            idiom = starter_idiom
+        else:
+            # 随机选择
+            idiom = random.choice(self.starter_idioms)
+        
+        # 获取最后一个字的拼音
+        last_char = idiom[-1]
+        last_pinyin_list = self._get_pinyin(last_char)
+        last_pinyin = last_pinyin_list[0] if last_pinyin_list else ''
+        
+        # 创建游戏状态
+        state_data = {
+            'creator_id': user_id,
+            'current_idiom': idiom,
+            'current_pinyin_last': last_pinyin,
+            'last_user_id': user_id,  # 发起人可以接第一个
+            'next_user_id': None,
+            'used_idioms': [idiom],
+            'blacklist': [],
+            'chain_length': 1,
+            'participants': {},
+            'history': [
+                {
+                    'user_id': user_id,
+                    'idiom': idiom,
+                    'timestamp': int(time.time())
+                }
+            ],
+            'status': 'playing'
+        }
+        
+        # 保存群状态（user_id=0）
+        self.db.save_game_state(chat_id, 0, 'idiom', state_data)
+        
+        text = f"## 🀄 成语接龙开始！\n\n"
+        text += f"**起始成语**：{idiom} [{last_pinyin}]\n\n"
+        text += f"任何人都可以接龙，输入 `.idiom [成语]` 开始吧！\n\n"
+        text += f"💡 提示：可以用 `.idiom [成语] @某人` 指定下一位"
+        
+        return text
+    
+    def _make_chain(self, chat_id: int, user_id: int, args: str) -> str:
+        """玩家接龙
+        
+        Args:
+            chat_id: 会话ID
+            user_id: 用户ID
+            args: 参数（成语 + 可能的@用户）
+            
+        Returns:
+            结果消息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！\n\n输入 `.idiom start` 开始游戏"
+        
+        state_data = state['state_data']
+        
+        # 检查用户权限
+        can_answer, error_msg = self._can_answer(state_data, user_id)
+        if not can_answer:
+            return error_msg
+        
+        # 解析成语和@用户
+        # 提取成语（去除@部分）
+        idiom_match = re.match(r'^([^\s@]+)', args)
+        if not idiom_match:
+            return "❌ 请输入4个字的词语"
+        
+        idiom = idiom_match.group(1).strip()
+        
+        # 验证词语
+        is_valid, error_msg = self._validate_idiom(idiom, state_data)
+        if not is_valid:
+            return error_msg
+        
+        # 解析@用户
+        mentioned_user_id = self._parse_mentioned_user(args)
+        
+        # 获取拼音
+        last_char = idiom[-1]
+        last_pinyin_list = self._get_pinyin(last_char)
+        last_pinyin = last_pinyin_list[0] if last_pinyin_list else ''
+        
+        # 更新状态
+        state_data['current_idiom'] = idiom
+        state_data['current_pinyin_last'] = last_pinyin
+        state_data['last_user_id'] = user_id
+        state_data['next_user_id'] = mentioned_user_id
+        state_data['used_idioms'].append(idiom)
+        state_data['chain_length'] += 1
+        
+        # 更新参与者统计
+        if str(user_id) not in state_data['participants']:
+            state_data['participants'][str(user_id)] = 0
+        state_data['participants'][str(user_id)] += 1
+        
+        # 记录历史
+        state_data['history'].append({
+            'user_id': user_id,
+            'idiom': idiom,
+            'timestamp': int(time.time())
+        })
+        
+        # 保存状态
+        self.db.save_game_state(chat_id, 0, 'idiom', state_data)
+        
+        # 构建回复
+        text = f"## ✅ 接龙成功！\n\n"
+        text += f"**{idiom}** [{last_pinyin}]\n\n"
+        text += f"**当前链长**：{state_data['chain_length']}\n\n"
+        
+        user_count = state_data['participants'][str(user_id)]
+        text += f"@用户{user_id} 成功次数：{user_count}\n\n"
+        
+        if mentioned_user_id:
+            text += f"已指定 @用户{mentioned_user_id} 接龙\n\n"
+        else:
+            text += "任何人都可以接龙\n\n"
+        
+        text += "继续加油！💪"
+        
+        return text
+    
+    def _set_next_user(self, chat_id: int, user_id: int, next_user_id: int) -> str:
+        """指定下一位接龙者
+        
+        Args:
+            chat_id: 会话ID
+            user_id: 当前用户ID
+            next_user_id: 指定的下一位用户ID
+            
+        Returns:
+            提示消息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！"
+        
+        state_data = state['state_data']
+        
+        # 检查是否是最后接龙的人
+        if state_data.get('last_user_id') != user_id:
+            return "❌ 只有最后接龙成功的人可以指定下一位"
+        
+        # 更新状态
+        state_data['next_user_id'] = next_user_id
+        self.db.save_game_state(chat_id, 0, 'idiom', state_data)
+        
+        return f"✅ 已指定 @用户{next_user_id} 接龙"
+    
+    def _reject_idiom(self, chat_id: int, user_id: int, idiom: str) -> str:
+        """裁判拒绝词语
+        
+        Args:
+            chat_id: 会话ID
+            user_id: 用户ID
+            idiom: 要拒绝的词语
+            
+        Returns:
+            提示消息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！"
+        
+        state_data = state['state_data']
+        
+        # 检查权限（仅发起人）
+        if state_data.get('creator_id') != user_id:
+            return "❌ 只有游戏发起人可以执行裁判操作"
+        
+        # 添加到黑名单
+        if idiom not in state_data['blacklist']:
+            state_data['blacklist'].append(idiom)
+        
+        # 如果是最后一个成语，回退状态
+        if state_data.get('current_idiom') == idiom and len(state_data['history']) > 1:
+            # 移除最后一条历史
+            removed = state_data['history'].pop()
+            removed_user = str(removed['user_id'])
+            
+            # 减少该用户的计数
+            if removed_user in state_data['participants']:
+                state_data['participants'][removed_user] -= 1
+                if state_data['participants'][removed_user] <= 0:
+                    del state_data['participants'][removed_user]
+            
+            # 恢复到上一个成语
+            if state_data['history']:
+                last_entry = state_data['history'][-1]
+                last_idiom = last_entry['idiom']
+                last_char = last_idiom[-1]
+                last_pinyin_list = self._get_pinyin(last_char)
+                
+                state_data['current_idiom'] = last_idiom
+                state_data['current_pinyin_last'] = last_pinyin_list[0] if last_pinyin_list else ''
+                state_data['last_user_id'] = last_entry['user_id']
+                state_data['next_user_id'] = None
+                state_data['chain_length'] -= 1
+            
+            # 从已使用列表中移除
+            if idiom in state_data['used_idioms']:
+                state_data['used_idioms'].remove(idiom)
+        
+        # 保存状态
+        self.db.save_game_state(chat_id, 0, 'idiom', state_data)
+        
+        text = f"✅ 已将「{idiom}」加入黑名单"
+        if state_data.get('current_idiom') != idiom:
+            text += f"\n\n当前成语：{state_data['current_idiom']}"
+        else:
+            text += "\n\n游戏状态已回退"
+        
+        return text
+    
+    def _show_status(self, chat_id: int) -> str:
+        """显示游戏状态
+        
+        Args:
+            chat_id: 会话ID
+            
+        Returns:
+            状态信息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！\n\n输入 `.idiom start` 开始游戏"
+        
+        state_data = state['state_data']
+        
+        text = f"## 🀄 成语接龙状态\n\n"
+        text += f"**当前成语**：{state_data['current_idiom']} [{state_data['current_pinyin_last']}]\n\n"
+        text += f"**链长**：{state_data['chain_length']}\n\n"
+        
+        # 下一位
+        if state_data.get('next_user_id'):
+            text += f"**下一位**：@用户{state_data['next_user_id']}\n\n"
+        else:
+            text += f"**下一位**：任何人都可以接龙\n\n"
+        
+        # 参与者排行
+        if state_data['participants']:
+            text += f"### 🏆 参与者排行\n\n"
+            sorted_participants = sorted(
+                state_data['participants'].items(),
+                key=lambda x: x[1],
+                reverse=True
+            )
+            for idx, (uid, count) in enumerate(sorted_participants[:5], 1):
+                text += f"{idx}. @用户{uid} - {count}次\n"
+            text += "\n"
+        
+        # 最近成语
+        history = state_data.get('history', [])
+        if history:
+            display_count = min(self.max_history_display, len(history))
+            recent = history[-display_count:]
+            text += f"### 📜 最近{display_count}个成语\n\n"
+            text += " → ".join([h['idiom'] for h in recent])
+        
+        return text
+    
+    def _show_blacklist(self, chat_id: int) -> str:
+        """显示黑名单
+        
+        Args:
+            chat_id: 会话ID
+            
+        Returns:
+            黑名单信息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！"
+        
+        state_data = state['state_data']
+        blacklist = state_data.get('blacklist', [])
+        
+        if not blacklist:
+            return "📋 黑名单为空"
+        
+        text = f"## 📋 黑名单词语\n\n"
+        text += "、".join(blacklist)
+        
+        return text
+    
+    def _stop_game(self, chat_id: int, user_id: int) -> str:
+        """结束游戏
+        
+        Args:
+            chat_id: 会话ID
+            user_id: 用户ID
+            
+        Returns:
+            总结消息
+        """
+        # 获取群状态
+        state = self.db.get_game_state(chat_id, 0, 'idiom')
+        if not state:
+            return "⚠️ 还没有开始游戏呢！"
+        
+        state_data = state['state_data']
+        
+        # 构建总结
+        text = f"## 🎮 游戏结束！\n\n"
+        text += f"**总链长**：{state_data['chain_length']}\n\n"
+        text += f"**参与人数**：{len(state_data['participants'])}\n\n"
+        
+        # 排行榜
+        if state_data['participants']:
+            text += f"### 🏆 排行榜\n\n"
+            sorted_participants = sorted(
+                state_data['participants'].items(),
+                key=lambda x: x[1],
+                reverse=True
+            )
+            for idx, (uid, count) in enumerate(sorted_participants, 1):
+                text += f"{idx}. @用户{uid} - {count}次\n"
+                # 更新统计
+                try:
+                    for _ in range(count):
+                        self.db.update_game_stats(int(uid), 'idiom', win=True)
+                except Exception as e:
+                    logger.error(f"更新统计失败: {e}")
+            text += "\n"
+        
+        # 完整接龙
+        history = state_data.get('history', [])
+        if history:
+            text += f"### 📜 完整接龙\n\n"
+            idioms = [h['idiom'] for h in history]
+            text += " → ".join(idioms)
+        
+        # 删除游戏状态
+        self.db.delete_game_state(chat_id, 0, 'idiom')
+        
+        return text
+    
+    def _format_history(self, history: list, count: int) -> str:
+        """格式化历史记录
+        
+        Args:
+            history: 历史记录列表
+            count: 显示数量
+            
+        Returns:
+            格式化的字符串
+        """
+        if not history:
+            return ""
+        
+        display_count = min(count, len(history))
+        recent = history[-display_count:]
+        return " → ".join([h['idiom'] for h in recent])
+    
+    def get_help(self) -> str:
+        """获取帮助信息"""
+        return """## 🀄 成语接龙
+
+### 基础用法
+- `.idiom start [成语]` - 开始游戏（可指定起始成语）
+- `.idiom [成语]` - 接龙
+- `.idiom [成语] @某人` - 接龙并指定下一位
+- `.idiom stop` - 结束游戏（任何人可执行）
+
+### 其他指令
+- `.idiom status` - 查看游戏状态
+- `.idiom blacklist` - 查看黑名单
+- `.idiom reject [词语]` - 裁判拒绝词语（仅发起人）
+- `.idiom next @某人` - 指定下一位（仅最后接龙者）
+
+### 游戏规则
+- 词语必须是4个字
+- 首字拼音必须匹配上个成语尾字拼音（忽略声调）
+- 不能重复使用成语
+- 不能连续接龙
+- 黑名单词语不可使用
+- 任何人都可以结束游戏
+
+### 示例
+```
+.idiom start 一马当先    # 开始游戏
+.idiom 先声夺人          # 接龙
+.idiom 人山人海 @张三    # 接龙并指定下一位
+.idiom reject 某词       # 发起人拒绝某词
+.idiom stop              # 结束游戏
+```
+
+💡 提示：支持多音字和谐音接龙
+"""
+