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

feat✨: 添加hook，更新其他

Browse Source
This commit is contained in:
HibiKier
2024-02-26 03:04:32 +08:00
parent eb0572ea77
commit 7b3793728a
22 changed files with 1064 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhenxun/builtin_plugins/admin/plugin_switch/__init__.py
+83 -49
View File
@@ -21,6 +21,7 @@ from zhenxun.utils.enum import BlockType, PluginType
from zhenxun.utils.rules import admin_check, ensure_group
from ._data_source import PluginManage, build_plugin, build_task
from .command import _group_status_matcher, _status_matcher
base_config = Config.get("admin_bot_manage")
@@ -53,60 +54,70 @@ __plugin_meta__ = PluginMetadata(
)
_status_matcher = on_alconna(
Alconna(
"switch",
Option("-t|--task", action=store_true, help_text="被动技能"),
Subcommand(
"open",
Args["name", str],
Option(
"-g|--group",
Args["group_id", str],
),
),
Subcommand(
"close",
Args["name", str],
Option(
"-t|--type",
Args["block_type", ["all", "a", "private", "p", "group", "g"]],
),
Option(
"-g|--group",
Args["group_id", str],
),
),
),
rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL"),
priority=5,
block=True,
)
# _status_matcher = on_alconna(
# Alconna(
# "switch",
# Option("-t|--task", action=store_true, help_text="被动技能"),
# Subcommand(
# "open",
# Args["name", str],
# Option(
# "-g|--group",
# Args["group_id", str],
# ),
# ),
# Subcommand(
# "close",
# Args["name", str],
# Option(
# "-t|--type",
# Args["block_type", ["all", "a", "private", "p", "group", "g"]],
# ),
# Option(
# "-g|--group",
# Args["group_id", str],
# ),
# ),
# ),
# rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL"),
# priority=5,
# block=True,
# )
# TODO: shortcut
# # TODO: shortcut
_group_status_matcher = on_alconna(
Alconna("group-status", Args["status", ["sleep", "wake"]]),
rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL") & ensure_group,
priority=5,
block=True,
)
# _group_status_matcher = on_alconna(
# Alconna("group-status", Args["status", ["sleep", "wake"]]),
# rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL") & ensure_group,
# priority=5,
# block=True,
# )
@_status_matcher.assign("$main")
async def _(bot: Bot, session: EventSession, arparma: Arparma):
image = None
if arparma.find("task"):
image = await build_task(session.id3 or session.id2)
elif session.id1 in bot.config.superusers:
if session.id1 in bot.config.superusers:
image = await build_plugin()
if image:
await Image(image.pic2bs4()).send(reply=True)
logger.info(
f"查看{'被动' if arparma.find('task') else '功能'}列表",
arparma.header_result,
session=session,
)
logger.info(
f"查看功能列表",
arparma.header_result,
session=session,
)
@_status_matcher.assign("task")
async def _(bot: Bot, session: EventSession, arparma: Arparma):
image = None
if image := await build_task(session.id3 or session.id2):
await Image(image.pic2bs4()).send(reply=True)
logger.info(
f"查看被动列表",
arparma.header_result,
session=session,
)
@_status_matcher.assign("open")
@@ -122,13 +133,14 @@ async def _(
await Text(result).send(reply=True)
logger.info(f"开启功能 {name}", arparma.header_result, session=session)
elif session.id1 in bot.config.superusers:
result = await PluginManage.superuser_block(name, None, group.result)
group_id = group.result if group.available else None
result = await PluginManage.superuser_block(name, None, group_id)
await Text(result).send(reply=True)
logger.info(
f"超级用户开启功能 {name}",
arparma.header_result,
session=session,
target=group.result,
target=group_id,
)
@@ -146,17 +158,39 @@ async def _(
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
_type = BlockType.ALL
if block_type.available:
if block_type.result in ["p", "private"]:
_type = BlockType.FRIEND
_type = BlockType.PRIVATE
elif block_type.result in ["g", "group"]:
_type = BlockType.GROUP
result = await PluginManage.superuser_block(name, _type, group.result)
result = await PluginManage.superuser_block(name, _type, group_id)
await Text(result).send(reply=True)
logger.info(
f"超级用户关闭功能 {name}, 禁用类型: {_type}",
arparma.header_result,
session=session,
target=group.result,
target=group_id,
)
@_group_status_matcher.handle()
async def _(
bot: Bot,
session: EventSession,
arparma: Arparma,
status: str,
):
if gid := session.id3 or session.id2:
if status == "sleep":
await PluginManage.sleep(gid)
logger.info("进行休眠", arparma.header_result, session=session)
await Text("那我先睡觉了...").finish()
else:
if PluginManage.is_wake(gid):
await Text("我还醒着呢！").finish()
await PluginManage.wake(gid)
logger.info("醒来", arparma.header_result, session=session)
await Text("呜..醒来了...").finish()
return Text("群组id为空...").send()
zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
+31 -8
View File
@@ -119,7 +119,7 @@ async def build_task(group_id: str | None) -> BuildImage:
task.name,
"开启" if task.module not in group.block_task else "关闭",
"开启" if task.status else "关闭",
task.run_time,
task.run_time or "-",
]
)
else:
@@ -129,7 +129,7 @@ async def build_task(group_id: str | None) -> BuildImage:
task.module,
task.name,
"开启" if task.status else "关闭",
task.run_time,
task.run_time or "-",
]
)
return await ImageTemplate.table_page(
@@ -143,6 +143,20 @@ async def build_task(group_id: str | None) -> BuildImage:
class PluginManage:
@classmethod
async def is_wake(cls, group_id: str) -> bool:
if c := await GroupConsole.get_or_none(group_id=group_id):
return c.status
return False
@classmethod
async def sleep(cls, group_id: str):
await GroupConsole.filter(group_id=group_id).update(status=False)
@classmethod
async def wake(cls, group_id: str):
await GroupConsole.filter(group_id=group_id).update(status=True)
@classmethod
async def block(cls, module: str):
await PluginInfo.filter(module=module).update(status=False)
@@ -191,8 +205,13 @@ class PluginManage:
返回:
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)
status_str = "开启" if status else "关闭"
if plugin := await PluginInfo.get_or_none(name=plugin_name):
if plugin:
group, _ = await GroupConsole.get_or_create(group_id=group_id)
if status:
if plugin.module in group.block_plugin:
@@ -200,12 +219,12 @@ class PluginManage:
f"{plugin.module},", ""
)
await group.save(update_fields=["block_plugin"])
return f"已成功{status_str} {plugin_name} 功能!"
return f"已成功{status_str} {plugin.name} 功能!"
else:
if plugin.module not in group.block_plugin:
group.block_plugin += f"{plugin.module},"
await group.save(update_fields=["block_plugin"])
return f"已成功{status_str} {plugin_name} 功能!"
return f"已成功{status_str} {plugin.name} 功能!"
return f"该功能已经{status_str}了喔，不要重复{status_str}..."
return "没有找到这个功能喔..."
@@ -223,7 +242,11 @@ class PluginManage:
返回:
str: 返回信息
"""
if plugin := await PluginInfo.get_or_none(name=plugin_name):
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:
if group_id:
if group := await GroupConsole.get_or_none(group_id=group_id):
if f"super:{plugin_name}," not in group.block_plugin:
@@ -238,7 +261,7 @@ class PluginManage:
plugin.status = not bool(block_type)
await plugin.save(update_fields=["status", "block_type"])
if not block_type:
return f"已成功将 {plugin_name} 全局启用!"
return f"已成功将 {plugin.name} 全局启用!"
else:
return f"已成功将 {plugin_name} 全局关闭!"
return f"已成功将 {plugin.name} 全局关闭!"
return "没有找到这个功能喔..."
zhenxun/builtin_plugins/admin/plugin_switch/command.py
+94
View File
@@ -0,0 +1,94 @@
from nonebot.rule import to_me
from nonebot_plugin_alconna import (
Alconna,
Args,
Option,
Subcommand,
on_alconna,
store_true,
)
from zhenxun.utils.rules import admin_check, ensure_group
_status_matcher = on_alconna(
Alconna(
"switch",
Option("-t|--task", action=store_true, help_text="被动技能"),
Subcommand(
"open",
Args["name", [str, int]],
Option(
"-g|--group",
Args["group_id", str],
),
),
Subcommand(
"close",
Args["name", [str, int]],
Option(
"-t|--type",
Args["block_type", ["all", "a", "private", "p", "group", "g"]],
),
Option(
"-g|--group",
Args["group_id", str],
),
),
),
rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL"),
priority=5,
block=True,
)
# TODO: shortcut
_group_status_matcher = on_alconna(
Alconna("group-status", Args["status", ["sleep", "wake"]]),
rule=admin_check("admin_bot_manage", "CHANGE_GROUP_SWITCH_LEVEL")
& ensure_group
& to_me(),
priority=5,
block=True,
)
_status_matcher.shortcut(
r"插件列表",
command="switch",
arguments=[],
prefix=True,
)
_status_matcher.shortcut(
r"群被动状态",
command="switch",
arguments=["--task"],
prefix=True,
)
_status_matcher.shortcut(
r"开启(?P<name>.+)",
command="switch",
arguments=["open", "{name}"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭(?P<name>.+)",
command="switch",
arguments=["close", "{name}"],
prefix=True,
)
_group_status_matcher.shortcut(
r"醒来",
command="group-status",
arguments=["wake"],
prefix=True,
)
_group_status_matcher.shortcut(
r"休息吧",
command="group-status",
arguments=["sleep"],
prefix=True,
)