172 lines
6.7 KiB
Python
172 lines
6.7 KiB
Python
import base64
|
||
from loguru import logger
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
|
||
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.revoke.message_auto_revoke import MessageAutoRevoke
|
||
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
|
||
from wechat_ipad import WechatAPIClient
|
||
|
||
|
||
class RobotMenuPlugin(MessagePluginInterface):
|
||
"""功能菜单插件"""
|
||
|
||
@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 "Trae AI"
|
||
|
||
@property
|
||
def command_prefix(self) -> Optional[str]:
|
||
return "" # 不需要前缀，直接匹配命令
|
||
|
||
@property
|
||
def commands(self) -> List[str]:
|
||
return self._commands
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
|
||
def initialize(self, context: Dict[str, Any]) -> bool:
|
||
"""初始化插件"""
|
||
self.LOG = logger
|
||
self.LOG.info(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.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}")
|
||
return True
|
||
|
||
def start(self) -> bool:
|
||
"""启动插件"""
|
||
self.LOG.info(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]
|
||
|
||
return command in self._commands
|
||
|
||
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_features(self, group_id: str) -> str:
|
||
"""获取某个群已启用的功能列表及其描述，并返回格式化的字符串
|
||
只返回描述中包含指令(方括号[])的功能
|
||
|
||
Args:
|
||
group_id: 群ID
|
||
|
||
Returns:
|
||
str: 格式化的已启用功能列表字符串
|
||
"""
|
||
enabled_features = []
|
||
|
||
# 检查群是否在列表中
|
||
if group_id not in GroupBotManager.local_cache["group_list"]:
|
||
return "该群未启用机器人功能"
|
||
|
||
# 遍历所有功能，检查哪些已启用且包含指令
|
||
for feature in Feature:
|
||
status = GroupBotManager.get_group_permission(group_id, feature)
|
||
# 只添加已启用且描述中包含方括号的功能
|
||
if status == PermissionStatus.ENABLED and "[" in feature.description and "]" in feature.description:
|
||
enabled_features.append({
|
||
"id": feature.value,
|
||
"name": feature.name,
|
||
"description": feature.description
|
||
})
|
||
|
||
# 如果没有启用任何带指令的功能
|
||
if not enabled_features:
|
||
return "该群未启用任何带指令的功能"
|
||
|
||
# 构建格式化的字符串
|
||
result = f"不支持@交互，请通过指令触发\n 群功能菜单：\n"
|
||
for feature in enabled_features:
|
||
result += f"{feature['id']}.{feature['description']}\n"
|
||
|
||
return result
|
||
|
||
def get_group_list(self) -> str:
|
||
"""返回所有启用了群机器人的群组清单"""
|
||
group_list = list(GroupBotManager.local_cache["group_list"])
|
||
if not group_list:
|
||
return "当前没有启用机器人的群组"
|
||
return "\n".join(group_list)
|
||
|
||
@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}")
|
||
command = content.split(" ")[0]
|
||
sender = message.get("sender")
|
||
roomid = message.get("roomid", "")
|
||
bot: WechatAPIClient = message.get("bot")
|
||
revoke: MessageAutoRevoke = message.get("revoke")
|
||
|
||
# 检查命令格式
|
||
if len(content.split(" ")) == 1:
|
||
# 显示功能菜单
|
||
menu_text = self.get_enabled_features(roomid if roomid else sender)
|
||
|
||
client_msg_id, create_time, new_msg_id = await bot.send_text_message((roomid if roomid else sender),
|
||
menu_text, sender)
|
||
revoke.add_message_to_revoke((roomid if roomid else sender), client_msg_id, create_time, new_msg_id, 90)
|
||
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((roomid if roomid else sender),
|
||
status_text, sender)
|
||
revoke.add_message_to_revoke((roomid if roomid else sender), client_msg_id, create_time, new_msg_id, 90)
|
||
return True, "显示功能状态"
|
||
|
||
except Exception as e:
|
||
self.LOG.error(f"处理功能请求出错: {e}")
|
||
return False, f"处理出错: {e}"
|