From b0a53715746a9090c0726ee43532f185802adb72 Mon Sep 17 00:00:00 2001 From: BalconyJH Date: Sun, 3 Nov 2024 16:08:20 +0800 Subject: [PATCH] :construction: Implement superuser.bot_manage. --- .../superuser/bot_manage/__init__.py | 18 +- .../superuser/bot_manage/bot_switch.py | 51 +++++ .../superuser/bot_manage/command.py | 112 ++++++++++ .../superuser/bot_manage/full_function.py | 51 +++++ .../superuser/bot_manage/plugin.py | 83 +++++++ .../superuser/bot_manage/status.py | 37 +++ .../superuser/bot_manage/task.py | 75 +++++++ zhenxun/models/bot_console.py | 210 +++++++++++------- 8 files changed, 547 insertions(+), 90 deletions(-) create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/command.py create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/full_function.py create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/plugin.py create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/status.py create mode 100644 zhenxun/builtin_plugins/superuser/bot_manage/task.py diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/__init__.py b/zhenxun/builtin_plugins/superuser/bot_manage/__init__.py index f196c167..7b9fdffc 100644 --- a/zhenxun/builtin_plugins/superuser/bot_manage/__init__.py +++ b/zhenxun/builtin_plugins/superuser/bot_manage/__init__.py @@ -1,26 +1,18 @@ -import os -import time +from pathlib import Path -from nonebot.rule import to_me -from nonebot.utils import run_sync -from nonebot.permission import SUPERUSER +import nonebot from nonebot.plugin import PluginMetadata -from nonebot_plugin_session import EventSession -from nonebot_plugin_apscheduler import scheduler -from nonebot_plugin_alconna import Alconna, on_alconna -from zhenxun.services.log import logger from zhenxun.utils.enum import PluginType -from zhenxun.utils.message import MessageUtils -from zhenxun.configs.path_config import TEMP_PATH from zhenxun.configs.utils import PluginExtraData -from zhenxun.utils.utils import ResourceDirManager + +_sub_plugins = set() +_sub_plugins |= nonebot.load_plugins(str(Path(__file__).parent.resolve())) __plugin_meta__ = PluginMetadata( name="Bot管理", description="指定bot对象的功能/被动开关和状态", usage=""" - 清理临时数据 """.strip(), extra=PluginExtraData( author="", diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py b/zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py new file mode 100644 index 00000000..9048b736 --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py @@ -0,0 +1,51 @@ +from nonebot_plugin_uninfo import Uninfo +from nonebot_plugin_alconna import Match, AlconnaMatch + +from zhenxun.services.log import logger +from zhenxun.utils.message import MessageUtils +from zhenxun.models.bot_console import BotConsole +from zhenxun.builtin_plugins.superuser.bot_manage.command import bot_manage + + +@bot_manage.assign("bot_switch.enable") +async def enable_bot_switch( + session: Uninfo, + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + else: + logger.info( + f"开启 {bot_id.result} ", + "bot_manage.bot_switch.enable", + session=session, + ) + try: + await BotConsole.set_bot_status(True, bot_id.result) + except ValueError: + await MessageUtils.build_message(f"bot_id {bot_id.result} 不存在").finish() + + await MessageUtils.build_message(f"已开启 {bot_id.result} ").finish() + + +@bot_manage.assign("bot_switch.disable") +async def diasble_bot_switch( + session: Uninfo, + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + else: + logger.info( + f"禁用 {bot_id.result} ", + "bot_manage.bot_switch.disable", + session=session, + ) + try: + await BotConsole.set_bot_status(False, bot_id.result) + except ValueError: + await MessageUtils.build_message(f"bot_id {bot_id.result} 不存在").finish() + + await MessageUtils.build_message(f"已禁用 {bot_id.result} ").finish() diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/command.py b/zhenxun/builtin_plugins/superuser/bot_manage/command.py new file mode 100644 index 00000000..0183fe62 --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/command.py @@ -0,0 +1,112 @@ +from nonebot.permission import SUPERUSER +from arclet.alconna.action import store_true +from nonebot_plugin_alconna import on_alconna +from arclet.alconna import Args, Option, Alconna, Subcommand + +bot_manage = on_alconna( + Alconna( + "bot_manage", + Subcommand( + "task", + Option( + "list", + action=store_true, + default=False, + help_text="查看 bot_id 下的所有可用被动", + ), + Subcommand( + "enable", + Option( + "-a|--all", + action=store_true, + default=False, + help_text="可选开启全部", + ), + Args["feature_name?", str], + Args["bot_id?", str], + ), + Subcommand( + "disable", + Option( + "-a|--all", + action=store_true, + default=False, + help_text="可选关闭全部", + ), + Args["feature_name?", str], + Args["bot_id?", str], + ), + ), + Subcommand( + "plugin", + Option( + "list", + action=store_true, + default=False, + help_text="查看 bot_id 下的所有可用插件", + ), + Subcommand( + "enable", + Option( + "-a|--all", + action=store_true, + default=False, + help_text="可选开启全部", + ), + Args["plugin_name?", str], + Args["bot_id?", str], + ), + Subcommand( + "disable", + Option( + "-a|--all", + action=store_true, + default=False, + help_text="可选关闭全部", + ), + Args["plugin_name?", str], + Args["bot_id?", str], + ), + ), + Subcommand( + "status", + Subcommand( + "tasks", + Args["bot_id", str], + ), + Subcommand( + "plugins", + Args["bot_id", str], + ), + Subcommand( + "bots", + Args["bot_id", str], + ), + ), + Subcommand( + "full_function", + Subcommand( + "enable", + Args["bot_id?", str], + ), + Subcommand( + "disable", + Args["bot_id?", str], + ), + ), + Subcommand( + "bot_switch", + Subcommand( + "enable", + Args["bot_id?", str], + ), + Subcommand( + "disable", + Args["bot_id?", str], + ), + ), + ), + permission=SUPERUSER, + priority=5, + block=True, +) diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/full_function.py b/zhenxun/builtin_plugins/superuser/bot_manage/full_function.py new file mode 100644 index 00000000..0e5dcebe --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/full_function.py @@ -0,0 +1,51 @@ +from nonebot_plugin_uninfo import Uninfo +from nonebot_plugin_alconna import Match, AlconnaMatch + +from zhenxun.services.log import logger +from zhenxun.utils.message import MessageUtils +from zhenxun.models.bot_console import BotConsole +from zhenxun.builtin_plugins.superuser.bot_manage.command import bot_manage + + +@bot_manage.assign("full_function.enable") +async def enable_full_function( + session: Uninfo, + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + else: + logger.info( + f"开启 {bot_id.result} 的所有可用插件及被动", + "bot_manage.full_function.enable", + session=session, + ) + await BotConsole.enable_all(bot_id.result, "tasks") + await BotConsole.enable_all(bot_id.result, "plugins") + + await MessageUtils.build_message( + f"已开启 {bot_id.result} 的所有插件及被动" + ).finish() + + +@bot_manage.assign("full_function.disable") +async def diasble_full_function( + session: Uninfo, + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + else: + logger.info( + f"禁用 {bot_id.result} 的所有可用插件及被动", + "bot_manage.full_function.disable", + session=session, + ) + await BotConsole.disable_all(bot_id.result, "tasks") + await BotConsole.disable_all(bot_id.result, "plugins") + + await MessageUtils.build_message( + f"已禁用 {bot_id.result} 的所有插件及被动" + ).finish() diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/plugin.py b/zhenxun/builtin_plugins/superuser/bot_manage/plugin.py new file mode 100644 index 00000000..0e5a1d7b --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/plugin.py @@ -0,0 +1,83 @@ +from nonebot_plugin_uninfo import Uninfo +from nonebot_plugin_alconna import Match, Query, AlconnaMatch, AlconnaQuery + +from zhenxun.services.log import logger +from zhenxun.utils.message import MessageUtils +from zhenxun.models.bot_console import BotConsole +from zhenxun.builtin_plugins.superuser.bot_manage.command import bot_manage + + +@bot_manage.assign("plugin") +async def bot_plugin( + session: Uninfo, + plugin_list: Query[bool] = AlconnaQuery("plugin.list.value", default=False), +): + if plugin_list: + logger.info("获取全部 bot 的所有可用插件", "bot_manage.plugin", session=session) + data = await BotConsole.get_plugins() + for bot in data: + await MessageUtils.build_message(f"{bot[0]} : {bot[1]}").finish() + + +@bot_manage.assign("plugin.enable") +async def enable_plugin( + session: Uninfo, + all_flag: Query[bool] = AlconnaQuery("plugin.enable.all.value", default=False), + plugin_name: Match[str] = AlconnaMatch("plugin_name"), + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if all_flag and plugin_name.available: + await logger.info( + f"启用全部 bot 的 {plugin_name.result} ", + "bot_manage.plugin.enable", + session=session, + ) + await BotConsole.enable_plugin(None, plugin_name.result) + await MessageUtils.build_message( + f"已启用全部 bot 的 {plugin_name.result} " + ).finish() + + if bot_id.available and plugin_name.available: + logger.info( + f"启用 {bot_id.result} 的 {plugin_name.result}", + "bot_manage.plugin.enable", + session=session, + ) + await BotConsole.enable_plugin(bot_id.result, plugin_name.result) + await MessageUtils.build_message( + f"已启用 {bot_id.result} 的 {plugin_name.result}" + ).finish() + + await MessageUtils.build_message("缺失参数").finish() + + +@bot_manage.assign("plugin.disable") +async def disable_plugin( + session: Uninfo, + all_flag: Query[bool] = AlconnaQuery("plugin.disable.all.value", default=False), + plugin_name: Match[str] = AlconnaMatch("plugin_name"), + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if all_flag and plugin_name.available: + await logger.info( + f"禁用全部 bot 的 {plugin_name.result} ", + "bot_manage.plugin.disable", + session=session, + ) + await BotConsole.disable_plugin(None, plugin_name.result) + await MessageUtils.build_message( + f"已禁用全部 bot 的 {plugin_name.result} " + ).finish() + + if bot_id.available and plugin_name.available: + logger.info( + f"禁用 {bot_id.result} 的 {plugin_name.result}", + "bot_manage.plugin.disable", + session=session, + ) + await BotConsole.disable_plugin(bot_id.result, plugin_name.result) + await MessageUtils.build_message( + f"已禁用 {bot_id.result} 的 {plugin_name.result}" + ).finish() + + await MessageUtils.build_message("缺失参数").finish() diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/status.py b/zhenxun/builtin_plugins/superuser/bot_manage/status.py new file mode 100644 index 00000000..f50b1f3e --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/status.py @@ -0,0 +1,37 @@ +import asyncio + +from nonebot_plugin_alconna import Match, AlconnaMatch + +from zhenxun.utils.message import MessageUtils +from zhenxun.models.bot_console import BotConsole +from zhenxun.builtin_plugins.superuser.bot_manage.command import bot_manage + + +@bot_manage.assign("status.tasks") +async def handle_tasks_status(bot_id: Match[str] = AlconnaMatch("bot_id")): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + result = await asyncio.gather( + BotConsole.get_tasks(bot_id.result), + BotConsole.get_tasks(bot_id.result, False), + ) + + await MessageUtils.build_message( + f"可用被动: {result[0]}\n禁用被动: {result[1]}" + ).finish() + + +@bot_manage.assign("status.plugins") +async def handle_plugins_status(bot_id: Match[str] = AlconnaMatch("bot_id")): + if not bot_id.available: + await MessageUtils.build_message("bot_id 不能为空").finish() + + result = await asyncio.gather( + BotConsole.get_plugins(bot_id.result), + BotConsole.get_plugins(bot_id.result, False), + ) + + await MessageUtils.build_message( + f"可用插件: {result[0]}\n禁用插件: {result[1]}" + ).finish() diff --git a/zhenxun/builtin_plugins/superuser/bot_manage/task.py b/zhenxun/builtin_plugins/superuser/bot_manage/task.py new file mode 100644 index 00000000..624bf180 --- /dev/null +++ b/zhenxun/builtin_plugins/superuser/bot_manage/task.py @@ -0,0 +1,75 @@ +from nonebot_plugin_uninfo import Uninfo +from nonebot_plugin_alconna import Match, Query, AlconnaMatch, AlconnaQuery + +from zhenxun.services.log import logger +from zhenxun.utils.message import MessageUtils +from zhenxun.models.bot_console import BotConsole +from zhenxun.builtin_plugins.superuser.bot_manage.command import bot_manage + + +@bot_manage.assign("task") +async def bot_task( + session: Uninfo, + task_list: Query[bool] = AlconnaQuery("task.list.value", default=False), +): + if task_list: + logger.info("获取全部 bot 的所有可用被动", "bot_manage.task", session=session) + data = await BotConsole.get_tasks() + for bot in data: + await MessageUtils.build_message(f"{bot[0]} : {bot[1]}").finish() + + +@bot_manage.assign("task.enable") +async def enable_task( + session: Uninfo, + all_flag: Query[bool] = AlconnaQuery("task.enable.all.value", default=False), + task_name: Match[str] = AlconnaMatch("plugin_name"), + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if all_flag and task_name.available: + await logger.info( + "启用全部 bot 的所有可用被动", "bot_manage.task.enable", session=session + ) + await BotConsole.enable_task(None, task_name.result) + await MessageUtils.build_message("已启用全部 bot 的所有可用被动").finish() + + if bot_id.available and task_name.available: + await logger.info( + f"启用 {bot_id.result} 的 {task_name.result}", + "bot_manage.task.enable", + session=session, + ) + await BotConsole.enable_task(bot_id.result, task_name.result) + await MessageUtils.build_message( + f"已启用 {bot_id.result} 的 {task_name.result}" + ).finish() + + await MessageUtils.build_message("缺失参数").finish() + + +@bot_manage.assign("task.disable") +async def disable_task( + session: Uninfo, + all_flag: Query[bool] = AlconnaQuery("task.disable.all.value", default=False), + task_name: Match[str] = AlconnaMatch("plugin_name"), + bot_id: Match[str] = AlconnaMatch("bot_id"), +): + if all_flag and task_name.available: + await logger.info( + "禁用全部 bot 的所有可用被动", "bot_manage.task.disable", session=session + ) + await BotConsole.disable_task(None, task_name.result) + await MessageUtils.build_message("已禁用全部 bot 的所有可用被动").finish() + + if bot_id.available and task_name.available: + logger.info( + f"禁用 {bot_id.result} 的 {task_name.result}", + "bot_manage.task.disable", + session=session, + ) + await BotConsole.disable_task(bot_id.result, task_name.result) + await MessageUtils.build_message( + f"已禁用 {bot_id.result} 的 {task_name.result}" + ).finish() + + await MessageUtils.build_message("缺失参数").finish() diff --git a/zhenxun/models/bot_console.py b/zhenxun/models/bot_console.py index a39d6cad..d7a8d20a 100644 --- a/zhenxun/models/bot_console.py +++ b/zhenxun/models/bot_console.py @@ -1,4 +1,4 @@ -from typing import overload +from typing import Literal, overload from tortoise import fields @@ -16,9 +16,9 @@ class BotConsole(Model): """创建时间""" platform = fields.CharField(255, null=True, description="平台") """平台""" - block_plugin = fields.TextField(default="", description="禁用插件") + block_plugins = fields.TextField(default="", description="禁用插件") """禁用插件""" - block_task = fields.TextField(default="", description="禁用被动技能") + block_tasks = fields.TextField(default="", description="禁用被动技能") """禁用被动技能""" available_plugins = fields.TextField(default="", description="可用插件") # todo)) 计划任务 or on_startup 写入可用插件 @@ -90,12 +90,12 @@ class BotConsole(Model): list[tuple[str, str]] | str: 被动技能 """ if not bot_id: - task_field = "available_tasks" if status else "block_task" + task_field = "available_tasks" if status else "block_tasks" return await cls.all().values_list("bot_id", task_field) result = await cls.get_or_none(bot_id=bot_id) if result: - return result.available_tasks if status else result.block_task + return result.available_tasks if status else result.block_tasks return "" @overload @@ -127,12 +127,12 @@ class BotConsole(Model): list[tuple[str, str]] | str: 插件 """ if not bot_id: - plugin_field = "available_plugins" if status else "block_plugin" + plugin_field = "available_plugins" if status else "block_plugins" return await cls.all().values_list("bot_id", plugin_field) result = await cls.get_or_none(bot_id=bot_id) if result: - return result.available_plugins if status else result.block_plugin + return result.available_plugins if status else result.block_plugins return "" @classmethod @@ -143,19 +143,27 @@ class BotConsole(Model): Args: status (bool): 状态 bot_id (str, optional): bot_id. Defaults to None. + + Raises: + ValueError: 未找到 bot_id """ if bot_id: - await cls.filter(bot_id=bot_id).update(status=status) + affected_rows = await cls.filter(bot_id=bot_id).update(status=status) + if not affected_rows: + raise ValueError(f"未找到 bot_id: {bot_id}") else: await cls.all().update(status=status) @overload - def convert_module_format(self, data: str) -> list[str]: ... + @classmethod + def _convert_module_format(cls, data: str) -> list[str]: ... @overload - def convert_module_format(self, data: list[str]) -> str: ... + @classmethod + def _convert_module_format(cls, data: list[str]) -> str: ... - def convert_module_format(self, data: str | list[str]) -> str | list[str]: + @classmethod + def _convert_module_format(cls, data: str | list[str]) -> str | list[str]: """ 在 ` None: """ - 在 available_field 和 block_field 之间移动指定的 data + 在 from_field 和 to_field 之间移动指定的 data Args: bot_id (str): 目标 bot 的 ID - available_field (str): 可用的目标字段 - block_field (str): 禁用的目标字段 + from_field (str): 源字段名称 + to_field (str): 目标字段名称 data (str): 要插入的内容 - to_block (bool): 移动到 block_list 还是 available_list Raises: - ValueError: 如果 data 不在 block_list 与 available_list 中 + ValueError: 如果 data 不在 from_field 和 to_field 中 """ - - def __move_to_block(): - nonlocal available_list, block_list - setattr( - bot_data, available_field, available_list.replace(formatted_data, "") - ) - setattr(bot_data, block_field, block_list + formatted_data) - - def __move_to_available(): - nonlocal available_list, block_list - setattr(bot_data, block_field, block_list.replace(formatted_data, "")) - setattr(bot_data, available_field, available_list + formatted_data) - bot_data, _ = await cls.get_or_create(bot_id=bot_id) formatted_data = cls.format(data) - available_list: str = getattr(bot_data, available_field) - block_list: str = getattr(bot_data, block_field) + from_list: str = getattr(bot_data, from_field) + to_list: str = getattr(bot_data, to_field) - if formatted_data not in available_list and formatted_data not in block_list: - raise ValueError(f"{data} 不在block_list与available_list中") + if formatted_data not in (from_list + to_list): + raise ValueError(f"{data} 不在源字段和目标字段中") - if to_block and formatted_data in available_list: - __move_to_block() - elif not to_block and formatted_data in block_list: - __move_to_available() + if formatted_data in from_list: + from_list = from_list.replace(formatted_data, "", 1) + if formatted_data not in to_list: + to_list += formatted_data - await bot_data.save(update_fields=[available_field, block_field]) + setattr(bot_data, from_field, from_list) + setattr(bot_data, to_field, to_list) + + await bot_data.save(update_fields=[from_field, to_field]) @classmethod - async def set_block_plugin(cls, bot_id: str | None, module: str) -> None: + async def disable_plugin(cls, bot_id: str | None, plugin_name: str) -> None: """ 禁用插件 Args: bot_id (str | None): bot_id - module (str): 模块名称 + plugin_name (str): 插件名称 """ if bot_id: - await cls.toggle_field( - bot_id, "available_plugins", "block_plugin", module, True + await cls._toggle_field( + bot_id, + "available_plugins", + "block_plugins", + plugin_name, ) else: bot_list = await cls.all() for bot in bot_list: - await cls.toggle_field( + await cls._toggle_field( bot.bot_id, "available_plugins", - "block_plugin", - module, - True, + "block_plugins", + plugin_name, ) @classmethod - async def set_unblock_plugin(cls, bot_id: str | None, module: str) -> None: + async def enable_plugin(cls, bot_id: str | None, plugin_name: str) -> None: """ 启用插件 Args: bot_id (str | None): bot_id - module (str): 模块名称 + plugin_name (str): 插件名称 """ if bot_id: - await cls.toggle_field( - bot_id, "block_plugin", "available_plugins", module, False + await cls._toggle_field( + bot_id, + "block_plugins", + "available_plugins", + plugin_name, ) else: bot_list = await cls.all() for bot in bot_list: - await cls.toggle_field( + await cls._toggle_field( bot.bot_id, - "block_plugin", + "block_plugins", "available_plugins", - module, - False, + plugin_name, ) @classmethod - async def set_block_task(cls, bot_id: str | None, module: str) -> None: + async def disable_task(cls, bot_id: str | None, task_name: str) -> None: """ 禁用被动技能 Args: bot_id (str | None): bot_id - module (str): 模块名称 + task_name (str): 被动技能名称 """ if bot_id: - await cls.toggle_field( - bot_id, "available_tasks", "block_task", module, True + await cls._toggle_field( + bot_id, + "available_tasks", + "block_tasks", + task_name, ) else: bot_list = await cls.all() for bot in bot_list: - await cls.toggle_field( - bot.bot_id, "available_tasks", "block_task", module, True + await cls._toggle_field( + bot.bot_id, + "available_tasks", + "block_tasks", + task_name, ) @classmethod - async def set_unblock_task(cls, bot_id: str | None, module: str) -> None: + async def enable_task(cls, bot_id: str | None, task_name: str) -> None: """ 启用被动技能 Args: bot_id (str | None): bot_id - module (str): 模块名称 + task_name (str): 被动技能名称 """ if bot_id: - await cls.toggle_field( - bot_id, "block_task", "available_tasks", module, False + await cls._toggle_field( + bot_id, + "block_tasks", + "available_tasks", + task_name, ) else: bot_list = await cls.all() for bot in bot_list: - await cls.toggle_field( - bot.bot_id, "block_task", "available_tasks", module, False + await cls._toggle_field( + bot.bot_id, + "block_tasks", + "available_tasks", + task_name, ) @classmethod - async def is_block_plugin(cls, bot_id: str, module: str) -> bool: + async def disable_all( + cls, + bot_id: str, + feat: Literal["plugins", "tasks"], + ) -> None: + """ + 禁用全部插件或被动技能 + + Args: + bot_id (str): bot_id + feat (Literal["plugins", "tasks"]): 插件或被动技能 + """ + bot_data, _ = await cls.get_or_create(bot_id=bot_id) + if feat == "plugins": + bot_data.available_plugins = "" + # todo)) 使用初始化方法重新写入bot_data.block_plugins以保证插件列表完整 + elif feat == "tasks": + bot_data.available_tasks = "" + # todo)) 使用初始化方法重新写入bot_data.block_tasks以保证插件列表完整 + await bot_data.save() + + @classmethod + async def enable_all( + cls, + bot_id: str, + feat: Literal["plugins", "tasks"], + ) -> None: + """ + 启用全部插件或被动技能 + + Args: + bot_id (str): bot_id + feat (Literal["plugins", "tasks"]): 插件或被动技能 + """ + bot_data, _ = await cls.get_or_create(bot_id=bot_id) + if feat == "plugins": + bot_data.block_plugins = "" + # todo)) 使用初始化方法重新写入bot_data.available_plugins以保证插件列表完整 + elif feat == "tasks": + bot_data.block_tasks = "" + # todo)) 使用初始化方法重新写入bot_data.available_tasks以保证插件列表完整 + await bot_data.save() + + @classmethod + async def is_block_plugin(cls, bot_id: str, plugin_name: str) -> bool: """ 检查插件是否被禁用 Args: bot_id (str): bot_id - module (str): 插件名称 + plugin_name (str): 插件名称 Returns: bool: 是否被禁用 """ bot_data, _ = await cls.get_or_create(bot_id=bot_id) - return cls.format(module) in bot_data.block_plugin + return cls.format(plugin_name) in bot_data.block_plugins @classmethod - async def is_block_task(cls, bot_id: str, module: str) -> bool: + async def is_block_task(cls, bot_id: str, task_name: str) -> bool: """ 检查被动技能是否被禁用 Args: bot_id (str): bot_id - module (str): 被动技能名称 + task_name (str): 被动技能名称 Returns: bool: 是否被禁用 """ bot_data, _ = await cls.get_or_create(bot_id=bot_id) - return cls.format(module) in bot_data.block_task + return cls.format(task_name) in bot_data.block_tasks