9f08a7ad96
变更项： 1. 新增“菜单 指令清单、菜单 指令、菜单 命令清单”入口，返回本群可用触发指令。 2. 从功能描述方括号中自动提取命令并去重，生成可直接复制的指令列表。 3. 优化主菜单文案，增加“常用指令清单”区块，帮助用户快速上手。 4. 扩展@语义关键词，支持“指令清单、命令清单”等问法触发菜单。
500 lines
22 KiB
Python
500 lines
22 KiB
Python
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_command_candidates(description: str) -> List[str]:
|
||
"""
|
||
从 Feature.description 中提取可触发命令列表
|
||
例如: [菜单 - 显示功能菜单 | 菜单 状态 - 显示功能状态]
|
||
-> ["菜单", "菜单 状态"]
|
||
"""
|
||
desc = str(description or "")
|
||
match = re.search(r"\[(.*?)\]", desc)
|
||
if not match:
|
||
return []
|
||
inner = match.group(1).strip()
|
||
if not inner:
|
||
return []
|
||
commands: List[str] = []
|
||
for part in inner.split("|"):
|
||
item = part.strip()
|
||
if not item:
|
||
continue
|
||
if "-" in item:
|
||
item = item.split("-", 1)[0].strip()
|
||
if item:
|
||
commands.append(item)
|
||
return commands
|
||
|
||
@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 _collect_group_commands(self, group_id: str) -> List[str]:
|
||
enabled_features = self._get_enabled_feature_items(group_id)
|
||
if not enabled_features:
|
||
return []
|
||
ordered: List[str] = []
|
||
seen = set()
|
||
for feature in enabled_features:
|
||
for cmd in self._extract_command_candidates(feature.get("description", "")):
|
||
key = self._normalize_text(cmd)
|
||
if not key or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
ordered.append(cmd)
|
||
return ordered
|
||
|
||
def build_command_list_text(self, group_id: str) -> str:
|
||
"""仅输出本群可用的触发指令清单"""
|
||
if group_id not in GroupBotManager.local_cache["group_list"]:
|
||
return "当前群未开通机器人功能，请联系管理员开启。"
|
||
|
||
commands = self._collect_group_commands(group_id)
|
||
if not commands:
|
||
return "当前群暂无可用指令。"
|
||
|
||
lines = ["本群指令清单：", "复制以下任一指令发送即可触发。", ""]
|
||
for idx, cmd in enumerate(commands, start=1):
|
||
lines.append(f"{idx}. {cmd}")
|
||
lines.append("")
|
||
lines.append("提示：发送“菜单”查看功能说明。")
|
||
return "\n".join(lines).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 "当前群暂无可用功能。"
|
||
|
||
command_list = self._collect_group_commands(group_id)
|
||
|
||
lines = [
|
||
"本群已开通功能：",
|
||
"直接复制“触发”里的命令发送即可。",
|
||
"",
|
||
]
|
||
if command_list:
|
||
lines.append("常用指令清单：")
|
||
for idx, cmd in enumerate(command_list, start=1):
|
||
lines.append(f"{idx}. {cmd}")
|
||
lines.append("")
|
||
|
||
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 in ["指令", "指令清单", "命令", "命令清单"]:
|
||
list_text = self.build_command_list_text(roomid if roomid else sender)
|
||
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, 120)
|
||
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}"
|