Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-14 21:52:56 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

🐛 fix bot console init (#1760)

Browse Source
This commit is contained in:
BalconyJH 2024-12-11 09:31:51 +08:00 committed by GitHub
parent 5590445679
commit 64b5316570
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 56 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
zhenxun/builtin_plugins/superuser/bot_manage/__init__.py
View File

@ -1,4 +1,5 @@
from pathlib import Path
from typing import cast
import nonebot
from nonebot.adapters import Bot
@ -31,36 +32,50 @@ __plugin_meta__ = PluginMetadata(
@driver.on_bot_connect
async def _(bot: Bot):
async def init_bot_console(bot: Bot):
"""初始化Bot管理
参数:
bot: Bot
"""
plugin_list = await PluginInfo.get_plugins(
plugin_type__in=[PluginType.NORMAL, PluginType.DEPENDANT, PluginType.ADMIN]
async def _filter_blocked_items(
items_list: list[str], block_list: list[str]
) -> list[str]:
"""过滤被block的项目
参数:
items_list: 需要过滤的项目列表
block_list: block列表
返回:
list: 过滤后且经过格式化的项目列表
"""
return [item for item in items_list if item not in block_list]
plugin_list = [
plugin.module
for plugin in await PluginInfo.get_plugins(
plugin_type__in=[PluginType.NORMAL, PluginType.DEPENDANT, PluginType.ADMIN]
)
]
task_list = cast(
list[str], await TaskInfo.filter(status=True).values_list("module", flat=True)
)
available_tasks: list[str] = await TaskInfo.filter(status=True).values_list(
"module", flat=True
) # type: ignore
available_plugins = [p.module for p in plugin_list]
# for _, bot in nonebot.get_bots().items():
platform = PlatformUtils.get_platform(bot)
bot_data, is_create = await BotConsole.get_or_create(
bot_data, created = await BotConsole.get_or_create(
bot_id=bot.self_id, platform=platform
)
if not is_create:
block_plugins = await bot_data.get_plugins(bot.self_id, False)
block_plugins = BotConsole._convert_module_format(block_plugins)
for module in available_plugins.copy():
if module in block_plugins:
available_plugins.remove(module)
block_tasks = await bot_data.get_tasks(bot.self_id, False)
block_tasks = BotConsole._convert_module_format(block_tasks)
for module in available_tasks.copy():
if module in block_plugins:
available_tasks.remove(module)
bot_data.available_plugins = BotConsole._convert_module_format(available_plugins)
bot_data.available_tasks = BotConsole._convert_module_format(available_tasks)
if not created:
task_list = await _filter_blocked_items(
task_list, await bot_data.get_tasks(bot.self_id, False)
)
plugin_list = await _filter_blocked_items(
plugin_list, await bot_data.get_plugins(bot.self_id, False)
)
bot_data.available_plugins = BotConsole.convert_module_format(plugin_list)
bot_data.available_tasks = BotConsole.convert_module_format(task_list)
await bot_data.save(update_fields=["available_plugins", "available_tasks"])
logger.info("初始化Bot管理完成...")

40
zhenxun/models/bot_console.py
View File

@ -21,10 +21,8 @@ class BotConsole(Model):
block_tasks = fields.TextField(default="", description="禁用被动技能")
"""禁用被动技能"""
available_plugins = fields.TextField(default="", description="可用插件")
# todo)) 计划任务 or on_startup 写入可用插件
"""可用插件"""
available_tasks = fields.TextField(default="", description="可用被动技能")
# todo)) 计划任务 or on_startup 写入可用被动技能
"""可用被动技能"""
class Meta: # type: ignore
@ -94,11 +92,11 @@ class BotConsole(Model):
"available_tasks" if status else "block_tasks"
)
data_list = await cls.all().values_list("bot_id", task_field)
return {k: cls._convert_module_format(v) for k, v in data_list}
return {k: cls.convert_module_format(v) for k, v in data_list}
result = await cls.get_or_none(bot_id=bot_id)
if result:
tasks = result.available_tasks if status else result.block_tasks
return cls._convert_module_format(tasks)
return cls.convert_module_format(tasks)
return []
@overload
@ -132,12 +130,12 @@ class BotConsole(Model):
if not bot_id:
plugin_field = "available_plugins" if status else "block_plugins"
data_list = await cls.all().values_list("bot_id", plugin_field)
return {k: cls._convert_module_format(v) for k, v in data_list}
return {k: cls.convert_module_format(v) for k, v in data_list}
result = await cls.get_or_none(bot_id=bot_id)
if result:
plugins = result.available_plugins if status else result.block_plugins
return cls._convert_module_format(plugins)
return cls.convert_module_format(plugins)
return []
@classmethod
@ -161,14 +159,14 @@ class BotConsole(Model):
@overload
@classmethod
def _convert_module_format(cls, data: str) -> list[str]: ...
def convert_module_format(cls, data: str) -> list[str]: ...
@overload
@classmethod
def _convert_module_format(cls, data: list[str]) -> str: ...
def convert_module_format(cls, data: list[str]) -> str: ...
@classmethod
def _convert_module_format(cls, data: str | list[str]) -> str | list[str]:
def convert_module_format(cls, data: str | list[str]) -> str | list[str]:
"""
在 `<aaa,<bbb,<ccc,` 和 `["aaa", "bbb", "ccc"]` 之间进行相互转换。
@ -341,16 +339,16 @@ class BotConsole(Model):
"""
bot_data, _ = await cls.get_or_create(bot_id=bot_id)
if feat == "plugins":
available_plugins = cls._convert_module_format(bot_data.available_plugins)
block_plugins = cls._convert_module_format(bot_data.block_plugins)
bot_data.block_plugins = cls._convert_module_format(
available_plugins = cls.convert_module_format(bot_data.available_plugins)
block_plugins = cls.convert_module_format(bot_data.block_plugins)
bot_data.block_plugins = cls.convert_module_format(
available_plugins + block_plugins
)
bot_data.available_plugins = ""
elif feat == "tasks":
available_tasks = cls._convert_module_format(bot_data.available_tasks)
block_tasks = cls._convert_module_format(bot_data.block_tasks)
bot_data.block_tasks = cls._convert_module_format(
available_tasks = cls.convert_module_format(bot_data.available_tasks)
block_tasks = cls.convert_module_format(bot_data.block_tasks)
bot_data.block_tasks = cls.convert_module_format(
available_tasks + block_tasks
)
bot_data.available_tasks = ""
@ -378,16 +376,16 @@ class BotConsole(Model):
"""
bot_data, _ = await cls.get_or_create(bot_id=bot_id)
if feat == "plugins":
available_plugins = cls._convert_module_format(bot_data.available_plugins)
block_plugins = cls._convert_module_format(bot_data.block_plugins)
bot_data.available_plugins = cls._convert_module_format(
available_plugins = cls.convert_module_format(bot_data.available_plugins)
block_plugins = cls.convert_module_format(bot_data.block_plugins)
bot_data.available_plugins = cls.convert_module_format(
available_plugins + block_plugins
)
bot_data.block_plugins = ""
elif feat == "tasks":
available_tasks = cls._convert_module_format(bot_data.available_tasks)
block_tasks = cls._convert_module_format(bot_data.block_tasks)
bot_data.available_tasks = cls._convert_module_format(
available_tasks = cls.convert_module_format(bot_data.available_tasks)
block_tasks = cls.convert_module_format(bot_data.block_tasks)
bot_data.available_tasks = cls.convert_module_format(
available_tasks + block_tasks
)
bot_data.block_tasks = ""