Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev
This commit is contained in:
Windpicker-owo
@@ -371,7 +371,7 @@ class PermissionCommand(PlusCommand):
|
||||
@register_plugin
|
||||
class PermissionManagerPlugin(BasePlugin):
|
||||
plugin_name: str = "permission_manager_plugin"
|
||||
enable_plugin: bool = True
|
||||
enable_plugin: bool = False
|
||||
dependencies: ClassVar[list[str]] = []
|
||||
python_dependencies: ClassVar[list[str]] = []
|
||||
config_file_name: str = "config.toml"
|
||||
|
||||
@@ -487,7 +487,7 @@ class ManagementCommand(PlusCommand):
|
||||
@register_plugin
|
||||
class PluginManagementPlugin(BasePlugin):
|
||||
plugin_name: str = "plugin_management_plugin"
|
||||
enable_plugin: bool = True
|
||||
enable_plugin: bool = False
|
||||
dependencies: ClassVar[list[str]] = []
|
||||
python_dependencies: ClassVar[list[str]] = []
|
||||
config_file_name: str = "config.toml"
|
||||
|
||||
16
src/plugins/built_in/system_management/__init__.py
Normal file
16
src/plugins/built_in/system_management/__init__.py
Normal file
@@ -0,0 +1,16 @@
|
||||
from src.plugin_system.base.plugin_metadata import PluginMetadata
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="SystemManagement",
|
||||
description="一个统一的系统管理插件，集成了权限、插件和定时任务的管理功能。",
|
||||
usage="/system <permission|plugin|schedule> [...]",
|
||||
version="1.0.0",
|
||||
author="MoFox-Studio",
|
||||
license="GPL-v3.0-or-later",
|
||||
repository_url="https://github.com/MoFox-Studio",
|
||||
keywords=["plugins", "permission", "management", "built-in"],
|
||||
extra={
|
||||
"is_built_in": True,
|
||||
"plugin_type": "permission",
|
||||
},
|
||||
)
|
||||
531
src/plugins/built_in/system_management/plugin.py
Normal file
531
src/plugins/built_in/system_management/plugin.py
Normal file
@@ -0,0 +1,531 @@
|
||||
"""
|
||||
统一系统管理插件
|
||||
|
||||
提供权限、插件和定时任务的统一管理命令。
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import ClassVar
|
||||
|
||||
from src.plugin_system.apis import (
|
||||
plugin_manage_api,
|
||||
)
|
||||
from src.plugin_system.apis.logging_api import get_logger
|
||||
from src.plugin_system.apis.permission_api import permission_api
|
||||
from src.plugin_system.apis.plugin_register_api import register_plugin
|
||||
from src.plugin_system.base.base_plugin import BasePlugin
|
||||
from src.plugin_system.base.command_args import CommandArgs
|
||||
from src.plugin_system.base.component_types import (
|
||||
ChatType,
|
||||
PermissionNodeField,
|
||||
PlusCommandInfo,
|
||||
)
|
||||
from src.plugin_system.base.config_types import ConfigField
|
||||
from src.plugin_system.base.plus_command import PlusCommand
|
||||
from src.plugin_system.utils.permission_decorators import require_permission
|
||||
from src.schedule.unified_scheduler import TriggerType, unified_scheduler
|
||||
|
||||
logger = get_logger("SystemManagement")
|
||||
|
||||
|
||||
class SystemCommand(PlusCommand):
|
||||
"""系统管理命令 - 使用PlusCommand系统"""
|
||||
|
||||
command_name = "system"
|
||||
command_description = "统一系统管理命令，支持权限、插件、定时任务等管理功能"
|
||||
command_aliases: ClassVar[list[str]] = ["sys", "系统管理"]
|
||||
priority = 10
|
||||
chat_type_allow = ChatType.ALL
|
||||
intercept_message = True
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
async def execute(self, args: CommandArgs) -> tuple[bool, str | None, bool]:
|
||||
"""执行系统管理命令"""
|
||||
if args.is_empty:
|
||||
await self._show_help("all")
|
||||
return True, "显示帮助信息", True
|
||||
|
||||
subcommand = args.get_first.lower()
|
||||
remaining_args = args.get_args()[1:]
|
||||
|
||||
if subcommand in ["permission", "perm", "权限"]:
|
||||
await self._handle_permission_commands(remaining_args)
|
||||
elif subcommand in ["plugin", "插件"]:
|
||||
await self._handle_plugin_commands(remaining_args)
|
||||
elif subcommand in ["schedule", "定时任务"]:
|
||||
await self._handle_schedule_commands(remaining_args)
|
||||
elif subcommand in ["help", "帮助"]:
|
||||
await self._show_help("all")
|
||||
else:
|
||||
await self.send_text(f"❌ 未知的子命令: {subcommand}\n使用 /system help 查看帮助")
|
||||
|
||||
return True, "命令执行完成", True
|
||||
|
||||
async def _show_help(self, target: str):
|
||||
"""显示帮助信息"""
|
||||
help_text = "帮助文档待施工..."
|
||||
if target == "all":
|
||||
help_text = """📋 系统管理命令帮助 (`/system`)
|
||||
🔧 主要功能：
|
||||
• `/system help` - 显示此帮助
|
||||
• `/system permission` - 权限管理
|
||||
• `/system plugin` - 插件管理
|
||||
• `/system schedule` - 定时任务管理
|
||||
"""
|
||||
elif target == "schedule":
|
||||
help_text = """📅 定时任务管理帮助
|
||||
📋 查看命令:
|
||||
• `/system schedule list` - 列出所有定时任务
|
||||
• `/system schedule list <类型>` - 列出指定类型的任务 (time, event, custom)
|
||||
• `/system schedule info <任务ID>` - 查看任务详情
|
||||
|
||||
⚙️ 控制命令:
|
||||
• `/system schedule pause <任务ID>` - 暂停一个任务
|
||||
• `/system schedule resume <任务ID>` - 恢复一个任务
|
||||
"""
|
||||
elif target == "plugin":
|
||||
help_text = """🔌 插件管理命令帮助
|
||||
📋 基本操作：
|
||||
• `/system plugin help` - 显示插件管理帮助
|
||||
• `/system plugin list` - 列出所有注册的插件
|
||||
• `/system plugin list_enabled` - 列出所有加载（启用）的插件
|
||||
• `/system plugin rescan` - 重新扫描所有插件目录
|
||||
|
||||
⚙️ 插件控制：
|
||||
• `/system plugin load <插件名>` - 加载指定插件
|
||||
• `/system plugin unload <插件名>` - 卸载指定插件
|
||||
• `/system plugin reload <插件名>` - 重新加载指定插件
|
||||
• `/system plugin force_reload <插件名>` - 强制重载指定插件
|
||||
• `/system plugin add_dir <目录路径>` - 添加插件目录
|
||||
"""
|
||||
elif target == "permission":
|
||||
help_text = """📋 权限管理命令帮助
|
||||
🔐 管理命令 (需要 `system.permission.manage` 权限):
|
||||
• /system permission grant <@用户|QQ号> <权限节点> - 授权
|
||||
• /system permission revoke <@用户|QQ号> <权限节点> - 撤销
|
||||
|
||||
👀 查看命令 (需要 `system.permission.view` 权限):
|
||||
• /system permission list [用户] - 查看用户权限
|
||||
• /system permission check <@用户|QQ号> <权限节点> - 检查权限
|
||||
• /system permission nodes [插件名] - 查看权限节点
|
||||
• /system permission allnodes - 查看所有权限节点详情
|
||||
"""
|
||||
await self.send_text(help_text)
|
||||
|
||||
|
||||
# =================================================================
|
||||
# Plugin Management Section
|
||||
# =================================================================
|
||||
async def _handle_plugin_commands(self, args: list[str]):
|
||||
"""处理插件管理相关命令"""
|
||||
if not args:
|
||||
await self._show_help("plugin")
|
||||
return
|
||||
|
||||
action = args[0].lower()
|
||||
remaining_args = args[1:]
|
||||
|
||||
if action in ["help", "帮助"]:
|
||||
await self._show_help("plugin")
|
||||
elif action in ["list", "列表"]:
|
||||
await self._list_registered_plugins()
|
||||
elif action in ["list_enabled", "已启用"]:
|
||||
await self._list_loaded_plugins()
|
||||
elif action in ["rescan", "重扫"]:
|
||||
await self._rescan_plugin_dirs()
|
||||
elif action in ["load", "加载"] and len(remaining_args) > 0:
|
||||
await self._load_plugin(remaining_args[0])
|
||||
elif action in ["unload", "卸载"] and len(remaining_args) > 0:
|
||||
await self._unload_plugin(remaining_args[0])
|
||||
elif action in ["reload", "重载"] and len(remaining_args) > 0:
|
||||
await self._reload_plugin(remaining_args[0])
|
||||
elif action in ["force_reload", "强制重载"] and len(remaining_args) > 0:
|
||||
await self._force_reload_plugin(remaining_args[0])
|
||||
else:
|
||||
await self.send_text("❌ 插件管理命令不合法\n使用 /system plugin help 查看帮助")
|
||||
|
||||
|
||||
# =================================================================
|
||||
# Schedule Management Section
|
||||
# =================================================================
|
||||
async def _handle_schedule_commands(self, args: list[str]):
|
||||
"""处理定时任务管理相关命令"""
|
||||
if not args:
|
||||
await self._show_help("schedule")
|
||||
return
|
||||
|
||||
action = args[0].lower()
|
||||
remaining_args = args[1:]
|
||||
|
||||
if action in ["list", "列表"]:
|
||||
target_type = remaining_args[0] if remaining_args else None
|
||||
await self._list_schedules(target_type)
|
||||
elif action in ["info", "详情"] and remaining_args:
|
||||
await self._get_schedule_info(remaining_args[0])
|
||||
elif action in ["pause", "暂停"] and remaining_args:
|
||||
await self._pause_schedule(remaining_args[0])
|
||||
elif action in ["resume", "恢复"] and remaining_args:
|
||||
await self._resume_schedule(remaining_args[0])
|
||||
else:
|
||||
await self.send_text("❌ 定时任务管理命令不合法\n使用 /system schedule help 查看帮助")
|
||||
|
||||
@require_permission("system.schedule.view", "❌ 你没有查看定时任务的权限")
|
||||
async def _list_schedules(self, trigger_type_str: str | None):
|
||||
"""列出定时任务"""
|
||||
trigger_type = None
|
||||
if trigger_type_str:
|
||||
try:
|
||||
trigger_type = TriggerType(trigger_type_str.lower())
|
||||
except ValueError:
|
||||
await self.send_text(f"❌ 无效的任务类型: {trigger_type_str}")
|
||||
return
|
||||
|
||||
tasks = await unified_scheduler.list_tasks(trigger_type)
|
||||
if not tasks:
|
||||
await self.send_text("📅 当前没有定时任务")
|
||||
return
|
||||
|
||||
response_parts = [f"📅 定时任务列表 (共 {len(tasks)} 个):"]
|
||||
for task in tasks:
|
||||
status = "▶️" if task["is_active"] else "⏸️"
|
||||
recurring = "🔁" if task["is_recurring"] else "➡️"
|
||||
response_parts.append(
|
||||
f"{status} `{task['task_name']}` ({task['trigger_type']}) {recurring}\n"
|
||||
f" ID: `{task['schedule_id'][:8]}...`"
|
||||
)
|
||||
await self.send_text("\n".join(response_parts))
|
||||
|
||||
@require_permission("system.schedule.view", "❌ 你没有查看定时任务详情的权限")
|
||||
async def _get_schedule_info(self, schedule_id: str):
|
||||
"""获取任务详情"""
|
||||
task_info = await unified_scheduler.get_task_info(schedule_id)
|
||||
if not task_info:
|
||||
await self.send_text(f"❌ 找不到ID为 `{schedule_id}` 的任务")
|
||||
return
|
||||
|
||||
info_str = f"📅 任务详情: `{task_info['task_name']}`\n"
|
||||
for key, value in task_info.items():
|
||||
info_str += f" • {key}: `{value}`\n"
|
||||
await self.send_text(info_str)
|
||||
|
||||
@require_permission("system.schedule.manage", "❌ 你没有管理定时任务的权限")
|
||||
async def _pause_schedule(self, schedule_id: str):
|
||||
"""暂停任务"""
|
||||
success = await unified_scheduler.pause_schedule(schedule_id)
|
||||
if success:
|
||||
await self.send_text(f"⏸️ 已暂停任务: `{schedule_id}`")
|
||||
else:
|
||||
await self.send_text(f"❌ 暂停任务失败: `{schedule_id}`")
|
||||
|
||||
@require_permission("system.schedule.manage", "❌ 你没有管理定时任务的权限")
|
||||
async def _resume_schedule(self, schedule_id: str):
|
||||
"""恢复任务"""
|
||||
success = await unified_scheduler.resume_schedule(schedule_id)
|
||||
if success:
|
||||
await self.send_text(f"▶️ 已恢复任务: `{schedule_id}`")
|
||||
else:
|
||||
await self.send_text(f"❌ 恢复任务失败: `{schedule_id}`")
|
||||
|
||||
# =================================================================
|
||||
# Permission Management Section
|
||||
# =================================================================
|
||||
|
||||
async def _list_loaded_plugins(self):
|
||||
"""列出已加载的插件"""
|
||||
plugins = plugin_manage_api.list_loaded_plugins()
|
||||
await self.send_text(f"📦 已加载的插件: {', '.join(plugins) if plugins else '无'}")
|
||||
|
||||
async def _list_registered_plugins(self):
|
||||
"""列出已注册的插件"""
|
||||
plugins = plugin_manage_api.list_registered_plugins()
|
||||
await self.send_text(f"📋 已注册的插件: {', '.join(plugins) if plugins else '无'}")
|
||||
|
||||
async def _rescan_plugin_dirs(self):
|
||||
"""重新扫描插件目录"""
|
||||
plugin_manage_api.rescan_plugin_directory()
|
||||
await self.send_text("🔄 插件目录重新扫描已启动")
|
||||
|
||||
async def _load_plugin(self, plugin_name: str):
|
||||
"""加载指定插件"""
|
||||
success, count = plugin_manage_api.load_plugin(plugin_name)
|
||||
if success:
|
||||
await self.send_text(f"✅ 插件加载成功: `{plugin_name}`")
|
||||
else:
|
||||
if count == 0:
|
||||
await self.send_text(f"⚠️ 插件 `{plugin_name}` 为禁用状态")
|
||||
else:
|
||||
await self.send_text(f"❌ 插件加载失败: `{plugin_name}`")
|
||||
|
||||
async def _unload_plugin(self, plugin_name: str):
|
||||
"""卸载指定插件"""
|
||||
success = await plugin_manage_api.remove_plugin(plugin_name)
|
||||
if success:
|
||||
await self.send_text(f"✅ 插件卸载成功: `{plugin_name}`")
|
||||
else:
|
||||
await self.send_text(f"❌ 插件卸载失败: `{plugin_name}`")
|
||||
|
||||
async def _reload_plugin(self, plugin_name: str):
|
||||
"""重新加载指定插件"""
|
||||
success = await plugin_manage_api.reload_plugin(plugin_name)
|
||||
if success:
|
||||
await self.send_text(f"✅ 插件重新加载成功: `{plugin_name}`")
|
||||
else:
|
||||
await self.send_text(f"❌ 插件重新加载失败: `{plugin_name}`")
|
||||
|
||||
async def _force_reload_plugin(self, plugin_name: str):
|
||||
"""强制重载指定插件（深度清理）"""
|
||||
await self.send_text(f"🔄 开始强制重载插件: `{plugin_name}`... (注意: 实际执行reload)")
|
||||
try:
|
||||
success = await plugin_manage_api.reload_plugin(plugin_name)
|
||||
if success:
|
||||
await self.send_text(f"✅ 插件重载成功: `{plugin_name}`")
|
||||
else:
|
||||
await self.send_text(f"❌ 插件重载失败: `{plugin_name}`")
|
||||
except Exception as e:
|
||||
await self.send_text(f"❌ 重载过程中发生错误: {e!s}")
|
||||
|
||||
|
||||
|
||||
# =================================================================
|
||||
# Permission Management Section
|
||||
# =================================================================
|
||||
async def _handle_permission_commands(self, args: list[str]):
|
||||
"""处理权限管理相关命令"""
|
||||
if not args:
|
||||
await self._show_help("permission")
|
||||
return
|
||||
|
||||
action = args[0].lower()
|
||||
remaining_args = args[1:]
|
||||
chat_stream = self.message.chat_info.stream_id
|
||||
|
||||
if action in ["grant", "授权", "give"]:
|
||||
await self._grant_permission(chat_stream, remaining_args)
|
||||
elif action in ["revoke", "撤销", "remove"]:
|
||||
await self._revoke_permission(chat_stream, remaining_args)
|
||||
elif action in ["list", "列表", "ls"]:
|
||||
await self._list_permissions(chat_stream, remaining_args)
|
||||
elif action in ["check", "检查"]:
|
||||
await self._check_permission(chat_stream, remaining_args)
|
||||
elif action in ["nodes", "节点"]:
|
||||
await self._list_nodes(chat_stream, remaining_args)
|
||||
elif action in ["allnodes", "全部节点", "all"]:
|
||||
await self._list_all_nodes_with_description(chat_stream)
|
||||
else:
|
||||
await self.send_text(f"❌ 未知的权限子命令: {action}")
|
||||
|
||||
@staticmethod
|
||||
def _parse_user_mention(mention: str) -> str | None:
|
||||
"""解析用户提及，提取QQ号"""
|
||||
at_match = re.search(r"@<[^:]+:(\d+)>", mention)
|
||||
if at_match:
|
||||
return at_match.group(1)
|
||||
if mention.isdigit():
|
||||
return mention
|
||||
return None
|
||||
|
||||
@require_permission("system.permission.manage", "❌ 你没有权限管理的权限")
|
||||
async def _grant_permission(self, chat_stream, args: list[str]):
|
||||
"""授权用户权限"""
|
||||
if len(args) < 2:
|
||||
await self.send_text("❌ 用法: /system permission grant <@用户|QQ号> <权限节点>")
|
||||
return
|
||||
|
||||
user_id = self._parse_user_mention(args[0])
|
||||
if not user_id:
|
||||
await self.send_text("❌ 无效的用户格式")
|
||||
return
|
||||
|
||||
permission_node = args[1]
|
||||
success = await permission_api.grant_permission(chat_stream.platform, user_id, permission_node)
|
||||
if success:
|
||||
await self.send_text(f"✅ 已授权用户 {user_id} 权限节点 `{permission_node}`")
|
||||
else:
|
||||
await self.send_text("❌ 授权失败")
|
||||
|
||||
@require_permission("system.permission.manage", "❌ 你没有权限管理的权限")
|
||||
async def _revoke_permission(self, chat_stream, args: list[str]):
|
||||
"""撤销用户权限"""
|
||||
if len(args) < 2:
|
||||
await self.send_text("❌ 用法: /system permission revoke <@用户|QQ号> <权限节点>")
|
||||
return
|
||||
|
||||
user_id = self._parse_user_mention(args[0])
|
||||
if not user_id:
|
||||
await self.send_text("❌ 无效的用户格式")
|
||||
return
|
||||
|
||||
permission_node = args[1]
|
||||
success = await permission_api.revoke_permission(chat_stream.platform, user_id, permission_node)
|
||||
if success:
|
||||
await self.send_text(f"✅ 已撤销用户 {user_id} 权限节点 `{permission_node}`")
|
||||
else:
|
||||
await self.send_text("❌ 撤销失败")
|
||||
|
||||
@require_permission("system.permission.view", "❌ 你没有查看权限的权限")
|
||||
async def _list_permissions(self, chat_stream, args: list[str]):
|
||||
"""列出用户权限"""
|
||||
target_user_id = None
|
||||
if args:
|
||||
target_user_id = self._parse_user_mention(args[0])
|
||||
if not target_user_id:
|
||||
await self.send_text("❌ 无效的用户格式")
|
||||
return
|
||||
else:
|
||||
target_user_id = chat_stream.user_info.user_id
|
||||
|
||||
is_master = await permission_api.is_master(chat_stream.platform, target_user_id)
|
||||
permissions = await permission_api.get_user_permissions(chat_stream.platform, target_user_id)
|
||||
|
||||
if is_master:
|
||||
response = f"👑 用户 `{target_user_id}` 是Master用户，拥有所有权限"
|
||||
else:
|
||||
if permissions:
|
||||
perm_list = "\n".join([f"• `{perm}`" for perm in permissions])
|
||||
response = f"📋 用户 `{target_user_id}` 拥有的权限：\n{perm_list}"
|
||||
else:
|
||||
response = f"📋 用户 `{target_user_id}` 没有任何权限"
|
||||
await self.send_text(response)
|
||||
|
||||
@require_permission("system.permission.view", "❌ 你没有查看权限的权限")
|
||||
async def _check_permission(self, chat_stream, args: list[str]):
|
||||
"""检查用户权限"""
|
||||
if len(args) < 2:
|
||||
await self.send_text("❌ 用法: /system permission check <@用户|QQ号> <权限节点>")
|
||||
return
|
||||
|
||||
user_id = self._parse_user_mention(args[0])
|
||||
if not user_id:
|
||||
await self.send_text("❌ 无效的用户格式")
|
||||
return
|
||||
|
||||
permission_node = args[1]
|
||||
has_permission = await permission_api.check_permission(chat_stream.platform, user_id, permission_node)
|
||||
is_master = await permission_api.is_master(chat_stream.platform, user_id)
|
||||
|
||||
if has_permission:
|
||||
response = f"✅ 用户 `{user_id}` 拥有权限 `{permission_node}`"
|
||||
if is_master:
|
||||
response += "（Master用户）"
|
||||
else:
|
||||
response = f"❌ 用户 `{user_id}` 没有权限 `{permission_node}`"
|
||||
await self.send_text(response)
|
||||
|
||||
@require_permission("system.permission.view", "❌ 你没有查看权限的权限")
|
||||
async def _list_nodes(self, chat_stream, args: list[str]):
|
||||
"""列出权限节点"""
|
||||
plugin_name = args[0] if args else None
|
||||
if plugin_name:
|
||||
nodes = await permission_api.get_plugin_permission_nodes(plugin_name)
|
||||
title = f"📋 插件 {plugin_name} 的权限节点："
|
||||
else:
|
||||
nodes = await permission_api.get_all_permission_nodes()
|
||||
title = "📋 所有权限节点："
|
||||
|
||||
if not nodes:
|
||||
response = f"📋 插件 {plugin_name} 没有注册任何权限节点" if plugin_name else "📋 系统中没有任何权限节点"
|
||||
else:
|
||||
node_list = []
|
||||
for node in nodes:
|
||||
default_text = "（默认授权）" if node["default_granted"] else "（默认拒绝）"
|
||||
node_list.append(f"• {node['node_name']} {default_text}")
|
||||
node_list.append(f" 📄 {node['description']}")
|
||||
if not plugin_name:
|
||||
node_list.append(f" 🔌 插件: {node['plugin_name']}")
|
||||
node_list.append("")
|
||||
response = title + "\n" + "\n".join(node_list)
|
||||
await self.send_text(response)
|
||||
|
||||
@require_permission("system.permission.view", "❌ 你没有查看权限的权限")
|
||||
async def _list_all_nodes_with_description(self, chat_stream):
|
||||
"""列出所有插件的权限节点（带详细描述）"""
|
||||
all_nodes = await permission_api.get_all_permission_nodes()
|
||||
if not all_nodes:
|
||||
await self.send_text("📋 系统中没有任何权限节点")
|
||||
return
|
||||
|
||||
plugins_dict = {}
|
||||
for node in all_nodes:
|
||||
plugin_name = node["plugin_name"]
|
||||
if plugin_name not in plugins_dict:
|
||||
plugins_dict[plugin_name] = []
|
||||
plugins_dict[plugin_name].append(node)
|
||||
|
||||
response_parts = ["📋 所有插件权限节点详情：\n"]
|
||||
for plugin_name in sorted(plugins_dict.keys()):
|
||||
nodes = plugins_dict[plugin_name]
|
||||
response_parts.append(f"🔌 **{plugin_name}** ({len(nodes)}个节点)：")
|
||||
for node in nodes:
|
||||
default_text = "✅默认授权" if node["default_granted"] else "❌默认拒绝"
|
||||
response_parts.append(f" • `{node['node_name']}` - {default_text}")
|
||||
response_parts.append(f" 📄 {node['description']}")
|
||||
response_parts.append("")
|
||||
|
||||
total_nodes = len(all_nodes)
|
||||
total_plugins = len(plugins_dict)
|
||||
response_parts.append(f"📊 统计：共 {total_plugins} 个插件，{total_nodes} 个权限节点")
|
||||
response = "\n".join(response_parts)
|
||||
|
||||
if len(response) > 4000:
|
||||
await self._send_long_message(response)
|
||||
else:
|
||||
await self.send_text(response)
|
||||
|
||||
async def _send_long_message(self, message: str):
|
||||
"""发送长消息，自动分段"""
|
||||
lines = message.split("\n")
|
||||
current_chunk = []
|
||||
current_length = 0
|
||||
for line in lines:
|
||||
line_length = len(line) + 1
|
||||
if current_length + line_length > 3500 and current_chunk:
|
||||
await self.send_text("\n".join(current_chunk))
|
||||
current_chunk = []
|
||||
current_length = 0
|
||||
current_chunk.append(line)
|
||||
current_length += line_length
|
||||
if current_chunk:
|
||||
await self.send_text("\n".join(current_chunk))
|
||||
|
||||
|
||||
@register_plugin
|
||||
class SystemManagementPlugin(BasePlugin):
|
||||
plugin_name: str = "system_management"
|
||||
enable_plugin: bool = True
|
||||
dependencies: ClassVar[list[str]] = [] # 插件依赖列表
|
||||
python_dependencies: ClassVar[list[str]] = [] # Python包依赖列表，现在使用内置API
|
||||
config_file_name: str = "config.toml" # 配置文件名
|
||||
config_schema: ClassVar[dict] = {
|
||||
"plugin": {
|
||||
"enabled": ConfigField(bool, default=True, description="是否启用插件"),
|
||||
}
|
||||
}
|
||||
|
||||
def get_plugin_components(self) -> list[tuple[PlusCommandInfo, type[PlusCommand]]]:
|
||||
"""返回插件的PlusCommand组件"""
|
||||
return [(SystemCommand.get_plus_command_info(), SystemCommand)]
|
||||
|
||||
permission_nodes: ClassVar[list[PermissionNodeField]] = [
|
||||
PermissionNodeField(
|
||||
node_name="permission.manage",
|
||||
description="权限管理：授权和撤销权限",
|
||||
),
|
||||
PermissionNodeField(
|
||||
node_name="permission.view",
|
||||
description="权限查看：查看权限信息",
|
||||
),
|
||||
PermissionNodeField(
|
||||
node_name="plugin.manage",
|
||||
description="插件管理：管理插件的加载、卸载、重载等",
|
||||
),
|
||||
PermissionNodeField(
|
||||
node_name="schedule.view",
|
||||
description="定时任务查看：查看定时任务列表和详情",
|
||||
),
|
||||
PermissionNodeField(
|
||||
node_name="schedule.manage",
|
||||
description="定时任务管理：暂停和恢复定时任务",
|
||||
),
|
||||
]
|
||||
Reference in New Issue
Block a user