mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
🚧 Implement superuser.bot_manage.
This commit is contained in:
BalconyJH
parent
893a235bf0
commit
b0a5371574
@ -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="",
|
||||
|
||||
51
zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py
Normal file
51
zhenxun/builtin_plugins/superuser/bot_manage/bot_switch.py
Normal file
@ -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()
|
||||
112
zhenxun/builtin_plugins/superuser/bot_manage/command.py
Normal file
112
zhenxun/builtin_plugins/superuser/bot_manage/command.py
Normal file
@ -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,
|
||||
)
|
||||
@ -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()
|
||||
83
zhenxun/builtin_plugins/superuser/bot_manage/plugin.py
Normal file
83
zhenxun/builtin_plugins/superuser/bot_manage/plugin.py
Normal file
@ -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()
|
||||
37
zhenxun/builtin_plugins/superuser/bot_manage/status.py
Normal file
37
zhenxun/builtin_plugins/superuser/bot_manage/status.py
Normal file
@ -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()
|
||||
75
zhenxun/builtin_plugins/superuser/bot_manage/task.py
Normal file
75
zhenxun/builtin_plugins/superuser/bot_manage/task.py
Normal file
@ -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()
|
||||
@ -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]:
|
||||
"""
|
||||
在 `<aaa,<bbb,<ccc,` 和 `["aaa", "bbb", "ccc"]` 之间进行相互转换。
|
||||
|
||||
@ -168,173 +176,221 @@ class BotConsole(Model):
|
||||
if isinstance(data, str):
|
||||
return [item.strip(",") for item in data.split("<") if item]
|
||||
elif isinstance(data, list):
|
||||
return "".join(self.format(item) for item in data)
|
||||
return "".join(cls.format(item) for item in data)
|
||||
|
||||
@classmethod
|
||||
async def toggle_field(
|
||||
async def _toggle_field(
|
||||
cls,
|
||||
bot_id: str,
|
||||
available_field: str,
|
||||
block_field: str,
|
||||
from_field: str,
|
||||
to_field: str,
|
||||
data: str,
|
||||
to_block: bool,
|
||||
) -> 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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user