from typing import Dict, Any, List, Optional, Tuple from loguru import logger import re import xml.etree.ElementTree as ET from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.decorator.points_decorator import plugin_points_cost from utils.revoke.message_auto_revoke import MessageAutoRevoke from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager from utils.wechat.contact_manager import ContactManager from wechat_ipad import WechatAPIClient class RobotMenuPlugin(MessagePluginInterface): """功能菜单插件""" # 功能权限常量 FEATURE_KEY = "ROBOT_MENU" FEATURE_DESCRIPTION = "📋 功能菜单 [菜单 - 显示功能菜单 | 菜单 状态 - 显示功能状态]" AT_QUERY_KEYWORDS = ( "功能清单", "功能菜单", "功能列表", "菜单", "怎么用", "如何用", "帮助", "help", ) @property def name(self) -> str: return "功能菜单" @property def version(self) -> str: return "1.0.0" @property def description(self) -> str: return "提供功能菜单功能，支持查看和使用机器人功能" @property def author(self) -> str: return "liu.wei" @property def command_prefix(self) -> Optional[str]: return "" # 不需要前缀，直接匹配命令 @property def commands(self) -> List[str]: return self._commands @property def feature_key(self) -> Optional[str]: return self.FEATURE_KEY @property def feature_description(self) -> Optional[str]: return self.FEATURE_DESCRIPTION def __init__(self): super().__init__() # 注册功能权限 self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" self.LOG = logger self.LOG.debug(f"正在初始化 {self.name} 插件...") # 保存上下文对象 self.event_system = context.get("event_system") # 从配置文件加载命令列表 self._commands = self._config.get("RobotMenu", {}).get("command", ["菜单", "功能"]) self.command_format = self._config.get("RobotMenu", {}).get("command-format", "菜单 功能名") self.enable = self._config.get("RobotMenu", {}).get("enable", True) self.LOG.debug(f"[{self.name}] 插件初始化完成，指令：{self._commands}") return True def start(self) -> bool: """启动插件""" self.LOG.debug(f"[{self.name}] 插件已启动") self.status = PluginStatus.RUNNING return True def stop(self) -> bool: """停止插件""" self.LOG.info(f"[{self.name}] 插件已停止") self.status = PluginStatus.STOPPED return True def can_process(self, message: Dict[str, Any]) -> bool: """检查是否可以处理该消息""" if not self.enable: return False content = str(message.get("content", "")).strip() command = content.split(" ")[0] if content else "" roomid = str(message.get("roomid", "") or "") is_at = bool(message.get("is_at", False)) # 指令入口 if command in self._commands: return True # @语义入口：只在群聊 + @机器人 + 明确菜单意图时触发 if roomid and is_at and self._is_at_menu_query(content): return True return False @staticmethod def _normalize_text(content: str) -> str: text = str(content or "").lower().strip() text = re.sub(r"\s+", "", text) return text def _is_at_menu_query(self, content: str) -> bool: text = self._normalize_text(content) if not text: return False return any(self._normalize_text(keyword) in text for keyword in self.AT_QUERY_KEYWORDS) @staticmethod def _extract_usage_from_description(description: str) -> str: """ 从 Feature.description 中提取 [ ... ] 内的触发方式 例如: '📋 功能菜单 [菜单 - 显示功能菜单]' -> '菜单' """ desc = str(description or "") match = re.search(r"\[(.*?)\]", desc) if not match: return "请发送“菜单”查看" inner = match.group(1).strip() # 取第一段触发方式，避免太长 first = inner.split("|")[0].strip() if "-" in first: first = first.split("-", 1)[0].strip() return first or "请发送“菜单”查看" @staticmethod def _extract_brief_from_description(description: str) -> str: desc = str(description or "") # 去掉开头 emoji 和 [..] 指令块，只保留一句用途 desc = re.sub(r"^[^\u4e00-\u9fa5A-Za-z0-9]+", "", desc) desc = re.sub(r"\[.*?\]", "", desc).strip() return desc or "暂无说明" def display_menu_status(self, group_id: str) -> str: """显示所有功能列表及其在指定群组中的当前状态，带emoji""" menu = [] for feature in Feature: status = GroupBotManager.get_group_permission(group_id, feature) status_emoji = "✅" if status == PermissionStatus.ENABLED else "❌" status_str = "启用" if status == PermissionStatus.ENABLED else "关闭" menu.append(f"{status_emoji} {status_str}-{feature.value}-{feature.description}") return "\n".join(menu) def _get_enabled_feature_items(self, group_id: str) -> List[Dict[str, str]]: """获取本群已启用且可对用户展示的功能条目""" items: List[Dict[str, str]] = [] # 检查群是否在列表中 if group_id not in GroupBotManager.local_cache["group_list"]: return items # 遍历所有功能，检查哪些已启用且包含指令 for feature in Feature: status = GroupBotManager.get_group_permission(group_id, feature) # 只添加已启用且描述中包含方括号的功能 if status == PermissionStatus.ENABLED and "[" in feature.description and "]" in feature.description: items.append({ "id": feature.value, "name": feature.name, "description": feature.description }) return items def get_enabled_features(self, group_id: str) -> str: """兼容旧接口：返回简化文本菜单""" enabled_features = self._get_enabled_feature_items(group_id) if group_id not in GroupBotManager.local_cache["group_list"]: return "该群未启用机器人功能" if not enabled_features: return "该群未启用任何带指令的功能" result = "群功能菜单：\n" for feature in enabled_features: usage = self._extract_usage_from_description(feature["description"]) brief = self._extract_brief_from_description(feature["description"]) result += f"{feature['id']}. {feature['name']} | 触发：{usage} | {brief}\n" return result.strip() def build_user_friendly_menu(self, group_id: str) -> str: """构建给普通群成员看的直观功能菜单""" if group_id not in GroupBotManager.local_cache["group_list"]: return "当前群未开通机器人功能，请联系管理员开启。" enabled_features = self._get_enabled_feature_items(group_id) if not enabled_features: return "当前群暂无可用功能。" lines = [ "本群已开通功能：", "直接复制“触发”里的命令发送即可。", "", ] for idx, feature in enumerate(enabled_features, start=1): usage = self._extract_usage_from_description(feature["description"]) brief = self._extract_brief_from_description(feature["description"]) lines.append(f"{idx}. {feature['name']}") lines.append(f"触发：{usage}") lines.append(f"说明：{brief}") lines.append("") lines.append("快捷入口：") lines.append("发送“菜单”可再次查看；发送“菜单 状态”可看全部开关状态。") return "\n".join(lines).strip() def get_group_list(self) -> str: """返回所有启用了群机器人的群组清单""" group_list = list(GroupBotManager.local_cache["group_list"]) if not group_list: return "当前没有启用机器人的群组" return "\n".join(group_list) def at_list(self, xml): try: root = ET.fromstring(xml) atuserlist_element = root.find('.//atuserlist') atuserlist_content = (atuserlist_element.text if atuserlist_element is not None else '').strip() atuserlist_content_no_commas = atuserlist_content.strip(',') atuserlist_content_no_commas = re.sub(r'\s+', '', atuserlist_content_no_commas) atuserlist_set = set(atuserlist_content_no_commas.split(',')) atuserlist_set.discard('') return atuserlist_set except ET.ParseError: return set() @plugin_stats_decorator(plugin_name="机器人菜单") async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.debug(f"插件执行： {self.name}：{content}") sender = message.get("sender") roomid = message.get("roomid", "") is_at = bool(message.get("is_at", False)) command = content.split(" ")[0] gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") revoke: MessageAutoRevoke = message.get("revoke") wx_msg = message.get("full_wx_msg") xml = wx_msg.msg_source if wx_msg else "" if not bot: self.LOG.error("WechatAPIClient 未初始化") return False, "Bot 未初始化" # 检查权限 if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" target = roomid if roomid else sender is_at_menu_query = roomid and is_at and self._is_at_menu_query(content) # @语义菜单入口：用户@机器人直接问“功能清单/怎么用” if is_at_menu_query and command not in self._commands: menu_text = self.build_user_friendly_menu(roomid) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, menu_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 120) return True, "at菜单问答" # 检查命令格式 parts = content.split() if len(parts) == 1: # 显示功能菜单 group_for_menu = roomid if roomid else sender menu_text = self.build_user_friendly_menu(group_for_menu) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, menu_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 120) return True, "显示功能菜单" # 提取命令名 cmd_name = content[len(command):].strip() try: # 处理特殊命令 if cmd_name.upper() == "状态": # 显示所有功能状态 status_text = self.display_menu_status(roomid if roomid else sender) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, status_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "显示功能状态" # 处理群列表命令 if cmd_name.upper() == "群列表": group_list_text = self.get_group_list() client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, group_list_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "显示群列表" # 处理功能启用/关闭命令 # 格式：菜单 启用 功能名 或 菜单 关闭 功能名 if len(parts) >= 3 and parts[1] in ["启用", "关闭"]: # 检查管理员权限 group_id_for_perm = roomid if roomid else sender if not GroupBotManager.is_admin_for_group(sender, group_id_for_perm): await bot.send_at_message(target, "❌权限不足，只有管理员可以管理功能", [sender]) return True, "权限不足" action = parts[1] feature_identifier = " ".join(parts[2:]) # 支持带空格的功能名 # 构造命令字符串，调用 GroupBotManager.handle_command # 尝试作为序号 if feature_identifier.isdigit(): command_str = f"{feature_identifier}-{action}" else: # 尝试作为功能名 # 查找匹配的功能枚举 matched_feature = None for feature in Feature: # 尝试通过枚举名称匹配 if feature.name.lower() == feature_identifier.lower(): matched_feature = feature break # 尝试通过序号匹配 if str(feature.value) == feature_identifier: matched_feature = feature break if matched_feature: command_str = f"{matched_feature.value}-{action}" else: # 直接使用功能标识符（可能是序号或功能名） command_str = f"{feature_identifier}-{action}" # 调用 GroupBotManager 处理命令 result = gbm.handle_command(roomid if roomid else sender, command_str) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, result, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "处理功能命令" if len(parts) >= 2 and parts[1] == "管理员": if not GroupBotManager.is_admin(sender): await bot.send_at_message(target, "❌权限不足，只有全局管理员可以管理群管理员", [sender]) return True, "权限不足" group_id = roomid if roomid else sender if len(parts) < 3: help_text = "管理员命令：菜单 管理员 添加 wxid | 菜单 管理员 删除 wxid | 菜单 管理员 列表" client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, help_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "管理员命令帮助" admin_action = parts[2] if admin_action == "添加" and len(parts) >= 4: at_users = self.at_list(xml) if xml else set() if len(at_users) == 1: new_admin_id = next(iter(at_users)) GroupBotManager.add_group_admin(group_id, new_admin_id) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, "已添加群管理员", sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "添加群管理员" candidate = parts[3] resolved_id = candidate if roomid: members = ContactManager.get_instance().get_group_members(roomid) or {} for wxid, nick in members.items(): if nick == candidate: resolved_id = wxid break GroupBotManager.add_group_admin(group_id, resolved_id) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, "已添加群管理员", sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "添加群管理员" if admin_action == "删除" and len(parts) >= 4: at_users = self.at_list(xml) if xml else set() if len(at_users) == 1: del_admin_id = next(iter(at_users)) ok = GroupBotManager.remove_group_admin(group_id, del_admin_id) msg = "已删除群管理员" if ok else "该管理员不在清单中" client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, msg, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "删除群管理员" candidate = parts[3] resolved_id = candidate if roomid: members = ContactManager.get_instance().get_group_members(roomid) or {} for wxid, nick in members.items(): if nick == candidate: resolved_id = wxid break ok = GroupBotManager.remove_group_admin(group_id, resolved_id) msg = "已删除群管理员" if ok else "该管理员不在清单中" client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, msg, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "删除群管理员" if admin_action in ["列表", "清单"]: admins = GroupBotManager.get_group_admins(group_id) list_text = "群管理员列表为空" if not admins else "\n".join(admins) client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, list_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "列出群管理员" help_text = "管理员命令：菜单 管理员 添加 wxid | 菜单 管理员 删除 wxid | 菜单 管理员 列表" client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, help_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "管理员命令帮助" # 如果是其他未知命令，显示帮助 help_text = f"❌命令格式错误！\n{self.command_format}" client_msg_id, create_time, new_msg_id = await bot.send_text_message(target, help_text, sender) revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 90) return True, "命令格式错误" except Exception as e: self.LOG.error(f"处理功能请求出错: {e}") import traceback self.LOG.error(traceback.format_exc()) return False, f"处理出错: {e}"