Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-10-06 20:29:59 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat✨: 插件管理添加统一开关命令

Browse Source
This commit is contained in:
HibiKier
2024-07-18 23:16:29 +08:00
parent 15aba0bea9
commit 4b48fc2557
9 changed files with 359 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhenxun/builtin_plugins/admin/plugin_switch/__init__.py
+111 -8
View File
@@ -43,6 +43,12 @@ __plugin_meta__ = PluginMetadata(
插件列表
开启/关闭[功能名称] ?[-t ["private", "p", "group", "g"](关闭类型)] ?[-g 群组Id]
开启/关闭插件df[功能名称]: 开启/关闭指定插件进群默认状态
开启/关闭所有插件df: 开启/关闭所有插件进群默认状态
开启/关闭所有插件:
私聊中: 开启/关闭所有插件全局状态
群组中: 开启/关闭当前群组所有插件状态
私聊下:
示例:
开启签到 : 全局开启签到
@@ -90,19 +96,69 @@ async def _(
bot: Bot,
session: EventSession,
arparma: Arparma,
name: str,
plugin_name: Match[str],
group: Match[str],
task: Query[bool] = AlconnaQuery("task.value", False),
default_status: Query[bool] = AlconnaQuery("default.value", False),
all: Query[bool] = AlconnaQuery("all.value", False),
):
if gid := session.id3 or session.id2:
if not all.result and not plugin_name.available:
await Text("请输入功能名称").finish(reply=True)
name = plugin_name.result
gid = session.id3 or session.id2
if gid:
"""群组中使用命令"""
if task.result:
result = await PluginManage.unblock_group_task(name, gid)
logger.info(f"开启群组被动 {name}", arparma.header_result, session=session)
else:
result = await PluginManage.block_group_plugin(name, gid)
if session.id1 in bot.config.superusers and (
all.result or default_status.result
):
if all.result:
"""所有插件"""
result = await PluginManage.set_all_plugin_status(
True, default_status.result, gid
)
logger.info(
f"超级用户开启群组中全部功能",
arparma.header_result,
session=session,
)
else:
"""单个插件的进群默认修改"""
result = await PluginManage.set_default_status(name, True)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
)
else:
result = await PluginManage.block_group_plugin(name, gid)
logger.info(f"开启功能 {name}", arparma.header_result, session=session)
await Text(result).send(reply=True)
logger.info(f"开启功能 {name}", arparma.header_result, session=session)
elif session.id1 in bot.config.superusers:
"""私聊"""
group_id = group.result if group.available else None
if all.result:
result = await PluginManage.set_all_plugin_status(
True, default_status.result, group_id
)
logger.info(
f"超级用户开启全部功能全局开关 {f'指定群组: {group_id}' if group_id else ''}",
arparma.header_result,
session=session,
)
await Text(result).finish(reply=True)
if default_status.result:
result = await PluginManage.set_default_status(name, True)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
target=group_id,
)
await Text(result).finish(reply=True)
if task.result:
result = await PluginManage.superuser_task_handle(name, group_id, True)
await Text(result).send(reply=True)
@@ -128,20 +184,67 @@ async def _(
bot: Bot,
session: EventSession,
arparma: Arparma,
name: str,
plugin_name: Match[str],
block_type: Match[str],
group: Match[str],
task: Query[bool] = AlconnaQuery("task.value", False),
default_status: Query[bool] = AlconnaQuery("default.value", False),
all: Query[bool] = AlconnaQuery("all.value", False),
):
if gid := session.id3 or session.id2:
if not all.result and not plugin_name.available:
await Text("请输入功能名称").finish(reply=True)
name = plugin_name.result
gid = session.id3 or session.id2
if gid:
if task.result:
result = await PluginManage.block_group_task(name, gid)
else:
result = await PluginManage.unblock_group_plugin(name, gid)
if session.id1 in bot.config.superusers and (
all.result or default_status.result
):
if all.result:
"""所有插件"""
result = await PluginManage.set_all_plugin_status(
False, default_status.result, gid
)
logger.info(
f"超级用户开启群组中全部功能",
arparma.header_result,
session=session,
)
else:
"""单个插件的进群默认修改"""
result = await PluginManage.set_default_status(name, False)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
)
else:
result = await PluginManage.unblock_group_plugin(name, gid)
logger.info(f"关闭功能 {name}", arparma.header_result, session=session)
await Text(result).send(reply=True)
logger.info(f"关闭功能 {name}", arparma.header_result, session=session)
elif session.id1 in bot.config.superusers:
group_id = group.result if group.available else None
if all.result:
result = await PluginManage.set_all_plugin_status(
False, default_status.result, group_id
)
logger.info(
f"超级用户关闭全部功能全局开关 {f'指定群组: {group_id}' if group_id else ''}",
arparma.header_result,
session=session,
)
await Text(result).finish(reply=True)
if default_status.result:
result = await PluginManage.set_default_status(name, False)
logger.info(
f"超级用户关闭 {name} 功能进群默认开关",
arparma.header_result,
session=session,
target=group_id,
)
await Text(result).finish(reply=True)
if task.result:
result = await PluginManage.superuser_task_handle(name, group_id, False)
await Text(result).send(reply=True)
zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
+87
View File
@@ -145,8 +145,75 @@ async def build_task(group_id: str | None) -> BuildImage:
class PluginManage:
@classmethod
async def set_default_status(cls, plugin_name: str, status: bool) -> str:
"""设置插件进群默认状态
参数:
plugin_name: 插件名称
status: 状态
返回:
str: 返回信息
"""
if plugin_name.isdigit():
plugin = await PluginInfo.get_or_none(id=int(plugin_name))
else:
plugin = await PluginInfo.get_or_none(name=plugin_name)
if plugin:
plugin.default_status = status
await plugin.save(update_fields=["default_status"])
return f'成功将 {plugin.name} 进群默认状态修改为: {"开启" if status else "关闭"}'
return f"没有找到这个功能喔..."
@classmethod
async def set_all_plugin_status(
cls, status: bool, is_default: bool = False, group_id: str | None = None
) -> str:
"""修改所有插件状态
参数:
status: 状态
is_default: 是否进群默认.
group_id: 指定群组id.
返回:
str: 返回信息
"""
if is_default:
await PluginInfo.filter(plugin_type=PluginType.NORMAL).update(
default_status=status
)
return f'成功将所有功能进群默认状态修改为: {"开启" if status else "关闭"}'
if group_id:
if group := await GroupConsole.get_or_none(
group_id=group_id, channel_id__isnull=True
):
if status:
group.block_plugin = ""
else:
module_list = await PluginInfo.filter(
plugin_type=PluginType.NORMAL
).values_list("module", flat=True)
group.block_plugin = ",".join(module_list) + "," # type: ignore
await group.save(update_fields=["block_plugin"])
return f'成功将此群组所有功能状态修改为: {"开启" if status else "关闭"}'
return "获取群组失败..."
await PluginInfo.filter(plugin_type=PluginType.NORMAL).update(
status=status, block_type=BlockType.ALL if not status else None
)
return f'成功将所有功能全局状态修改为: {"开启" if status else "关闭"}'
@classmethod
async def is_wake(cls, group_id: str) -> bool:
"""是否醒来
参数:
group_id: 群组id
返回:
bool: 是否醒来
"""
if c := await GroupConsole.get_or_none(
group_id=group_id, channel_id__isnull=True
):
@@ -155,22 +222,42 @@ class PluginManage:
@classmethod
async def sleep(cls, group_id: str):
"""休眠
参数:
group_id: 群组id
"""
await GroupConsole.filter(group_id=group_id, channel_id__isnull=True).update(
status=False
)
@classmethod
async def wake(cls, group_id: str):
"""醒来
参数:
group_id: 群组id
"""
await GroupConsole.filter(group_id=group_id, channel_id__isnull=True).update(
status=True
)
@classmethod
async def block(cls, module: str):
"""禁用
参数:
module: 模块名
"""
await PluginInfo.filter(module=module).update(status=False)
@classmethod
async def unblock(cls, module: str):
"""启用
参数:
module: 模块名
"""
await PluginInfo.filter(module=module).update(status=True)
@classmethod
zhenxun/builtin_plugins/admin/plugin_switch/command.py
+48 -3
View File
@@ -14,9 +14,11 @@ _status_matcher = on_alconna(
Alconna(
"switch",
Option("-t|--task", action=store_true, help_text="被动技能"),
Option("-df|--default", action=store_true, help_text="进群默认开关"),
Option("--all", action=store_true, help_text="全部插件"),
Subcommand(
"open",
Args["name", [str, int]],
Args["plugin_name?", [str, int]],
Option(
"-g|--group",
Args["group", str],
@@ -24,7 +26,7 @@ _status_matcher = on_alconna(
),
Subcommand(
"close",
Args["name", [str, int]],
Args["plugin_name?", [str, int]],
Option(
"-t|--type",
Args["block_type", ["all", "a", "private", "p", "group", "g"]],
@@ -72,6 +74,27 @@ _status_matcher.shortcut(
)
_status_matcher.shortcut(
r"开启所有插件",
command="switch",
arguments=["open", "s", "--all"],
prefix=True,
)
_status_matcher.shortcut(
r"开启所有插件df",
command="switch",
arguments=["open", "s", "-df", "--all"],
prefix=True,
)
_status_matcher.shortcut(
r"开启插件df(?P<name>.+)",
command="switch",
arguments=["open", "{name}", "-df"],
prefix=True,
)
_status_matcher.shortcut(
r"开启(?P<name>.+)",
command="switch",
@@ -79,6 +102,7 @@ _status_matcher.shortcut(
prefix=True,
)
_status_matcher.shortcut(
r"关闭群被动(?P<name>.+)",
command="switch",
@@ -86,6 +110,28 @@ _status_matcher.shortcut(
prefix=True,
)
_status_matcher.shortcut(
r"关闭所有插件",
command="switch",
arguments=["close", "s", "--all"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭所有插件df",
command="switch",
arguments=["close", "s", "-df", "--all"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭插件df(?P<name>.+)",
command="switch",
arguments=["close", "{name}", "-df"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭(?P<name>.+)",
command="switch",
@@ -93,7 +139,6 @@ _status_matcher.shortcut(
prefix=True,
)
_group_status_matcher.shortcut(
r"醒来",
command="group-status",