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

♻️ refactor(plugin_switch): 重构功能开关与帮助系统并优化 UI 报表渲染 (#2111)

Browse Source 
* ♻️ refactor(plugin_switch): 重构功能开关与帮助系统并优化 UI 报表渲染

- 引入策略模式 (Strategy Pattern) 统一插件与被动任务的开关逻辑
- 新增超级用户强制管控功能，支持系统级禁用且群管无法自行开启
- 新增全服群组活跃状态报表与功能全局覆盖率可视化统计图表
- 优化指令解析系统，支持多功能批量开关及更智能的快捷词映射
- 重构帮助系统，支持在帮助页区分展示管理员与超级用户指令
- 细化插件列表状态显示，区分群控、系统管控与全局禁用状态
- UI 组件 UserInfoBlock 新增 extra 扩展插槽支持
- 更新 resources.spec 资源版本要求至 1.1.1

* 🚨 auto fix by pre-commit hooks

---------

Co-authored-by: webjoin111 <455457521@qq.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Rumio
2026-03-19 16:08:13 +08:00
committed by GitHub
co-authored by webjoin111 pre-commit-ci[bot]
parent ce94f63d9a
commit 0f91ce03e4
15 changed files with 1464 additions and 1104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhenxun/builtin_plugins/admin/plugin_switch/__init__.py
+365 -297
View File
@@ -1,16 +1,26 @@
from nonebot.adapters import Bot
from nonebot.adapters import Bot, Event
from nonebot.exception import FinishedException
from nonebot.permission import SUPERUSER as SUPERUSER_PERM
from nonebot.plugin import PluginMetadata
from nonebot_plugin_alconna import AlconnaQuery, Arparma, Match, Query
from nonebot_plugin_alconna import AlconnaMatch, AlconnaQuery, Arparma, Match, Query
from nonebot_plugin_uninfo import Uninfo
from zhenxun.configs.config import Config
from zhenxun.configs.utils import PluginExtraData, RegisterConfig
from zhenxun.services.log import logger
from zhenxun.services.tags import tag_manager
from zhenxun.utils.enum import BlockType, PluginType
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.platform import PlatformUtils
from ._data_source import PluginManager, build_plugin, build_task
from .command import _group_status_matcher, _status_matcher
from .data_source import PluginManager
from .ui import (
build_plugin,
build_task,
render_global_status,
render_group_active_status,
)
base_config = Config.get("plugin_switch")
@@ -18,64 +28,57 @@ base_config = Config.get("plugin_switch")
__plugin_meta__ = PluginMetadata(
name="功能开关",
description="对群组内的功能限制，超级用户可以对群组以及全局的功能被动开关限制",
usage="""
普通管理员
格式:
开启/关闭[功能名称] : 开关功能
开启/关闭群被动[被动名称] : 群被动开关
开启/关闭所有插件 : 开启/关闭当前群组所有插件状态
开启/关闭所有群被动 : 开启/关闭当前群组所有群被动
群被动状态 : 查看被动技能开关状态
醒来 : 结束休眠
休息吧 : 群组休眠, 不会再响应命令
usage="""### 基础开关控制
- `开启/关闭 [功能名...]`：在当前群开启/关闭指定功能
- `开启/关闭被动 [被动名...]`：在当前群开启/关闭指定被动
- `开启/关闭所有功能`：在当前群开启/关闭所有功能
- `开启/关闭所有被动`：在当前群开启/关闭所有被动
示例:
开启签到 : 开启签到
关闭签到 : 关闭签到
开启群被动早晚安 : 关闭被动任务早晚安
**操作示例：**
- `关闭 签到 抽卡 色图`：在当前群批量关闭指定功能
""".strip(),
### 机器人状态控制
- `醒来`：让机器人在当前群恢复工作
- `休息吧`：让机器人在当前群进入休眠状态
""",
extra=PluginExtraData(
author="HibiKier",
version="0.1",
version="1.0",
plugin_type=PluginType.SUPER_AND_ADMIN,
superuser_help="""
格式:
插件列表
开启/关闭[功能名称] ?[-t ["private", "p", "group", "g"](关闭类型)] ?[-g 群组Id]
开启/关闭插件df[功能名称]: 开启/关闭指定插件进群默认状态
开启/关闭所有插件df: 开启/关闭所有插件进群默认状态
开启/关闭所有插件:
私聊中: 开启/关闭所有插件全局状态
群组中: 开启/关闭当前群组所有插件状态
开启/关闭群被动[name] ?[-g [group_id]]
私聊中: 开启/关闭全局指定的被动状态
群组中: 开启/关闭当前群组指定的被动状态
示例:
关闭群被动早晚安
关闭群被动早晚安 -g 12355555
开启/关闭默认群被动 [被动名称]
私聊下: 开启/关闭群被动默认状态
示例:
关闭默认群被动 早晚安
开启/关闭所有群被动 ?[-g [group_id]]
私聊中: 开启/关闭全局或指定群组被动状态
示例:
开启所有群被动: 开启全局所有被动
开启所有群被动 -g 12345678: 开启群组12345678所有被动
私聊下:
示例:
开启签到 : 全局开启签到
关闭签到 : 全局关闭签到
关闭签到 p : 全局私聊关闭签到
关闭签到 -g 12345678 : 关闭群组12345678的签到功能(普通管理员无法开启)
""",
admin_level=base_config.get("CHANGE_GROUP_SWITCH_LEVEL", 2),
superuser_help="""### 状态查询
- `插件列表`：查看所有插件的全局状态、群聊状态
- `被动状态`：查看所有被动技能的状态
- `查看功能状态 [功能名]`：查看指定功能在所有群组中的开关状态
- `查看被动状态 [被动名]`：查看指定被动在所有群组中的开关状态
- `查看群状态`：查看所有群组的休眠/工作状态
### 高级开关控制 (跨群/全局)
支持在指令后追加以下参数进行批量操作：
- `-g <群号>`：指定操作目标群（可多个）
- `-t <标签>`：指定操作带有特定标签的群
- `--all`：操作所有群组
- `--only`：白名单模式，仅在指定群组开启，其他群组自动关闭
- `-s`：**强制管控**。使用系统级字段禁用功能，群管理员无法通过普通指令自行开启
**操作示例：**
- `关闭 签到 抽卡 -t 游戏群`：关闭所有带有"游戏群"标签的群的签到和抽卡功能
- `开启 色图 --only -g 123456 654321`：仅在这两个群开启色图，其余群全部关闭
- `关闭 色图 -s`：在当前群强制锁定关闭色图，群管无法开启
### 系统级开关
追加 `--type [范围]` 或使用特定快捷词实现系统级控制。
范围：`p` (私聊), `g` (所有群聊), `a` (全局)
- `关闭 签到 --type a`：全局彻底禁用签到功能
- `开启/关闭默认 [功能名]`：修改功能进群时的默认开关状态
- `开启/关闭所有默认功能`：批量修改所有功能的进群默认状态
### 强制唤醒/休眠
同样支持高级目标参数。
- `休息吧 --all`：所有群组进入休眠
- `醒来 -t 内部测试群`：唤醒带有该标签的群组
""",
configs=[
RegisterConfig(
key="CHANGE_GROUP_SWITCH_LEVEL",
@@ -103,276 +106,313 @@ async def _(
session=session,
)
await MessageUtils.build_message(image).finish(reply_to=True)
else:
await MessageUtils.build_message("权限不足捏...").finish(reply_to=True)
async def get_target_groups(
bot: Bot,
event: Event,
session: Uninfo,
tag: str | None,
groups: tuple[str, ...] | None,
all_scope: bool,
) -> set[str] | None:
"""解析目标群组列表，包含标签、群号和全量选项。"""
targets: set[str] = set()
is_superuser = await SUPERUSER_PERM(bot, event)
if (tag or groups or all_scope) and not is_superuser:
return None
if groups:
targets.update(str(group_id) for group_id in groups if group_id)
if tag:
tag_groups = await tag_manager.resolve_tag_to_group_ids(tag, bot=bot)
targets.update(str(group_id) for group_id in tag_groups)
if all_scope:
all_groups, _ = await PlatformUtils.get_group_list(bot)
targets.update(str(group.group_id) for group in all_groups if group.group_id)
if not targets and session.group:
targets.add(str(session.group.id))
return targets
async def _handle_switch_command(
status: bool,
bot: Bot,
event: Event,
session: Uninfo,
arparma: Arparma,
plugin_names: Match[tuple[str, ...]],
groups: Match[tuple[str, ...]] = AlconnaMatch("groups"),
tag: Match[str] = AlconnaMatch("tag"),
task: Query[bool] = AlconnaQuery("task.value", False),
default_status: Query[bool] = AlconnaQuery("default.value", False),
all_groups_flag: Query[bool] = AlconnaQuery("all.value", False),
all_plugins_flag: Query[bool] = AlconnaQuery("all-plugins.value", False),
only_flag: Query[bool] | None = None,
use_su_field: Query[bool] = AlconnaQuery("su.value", False),
):
is_superuser = await SUPERUSER_PERM(bot, event)
only_flag_value = only_flag.result if only_flag else False
is_remote = bool(
tag.available or groups.available or all_groups_flag.result or only_flag_value
)
use_su_field_final = is_remote or use_su_field.result
sub_name = "open" if status else "close"
block_type_val = arparma.query(f"{sub_name}.type.block_type")
if block_type_val is not None:
if not is_superuser:
return
if task.result:
await MessageUtils.build_message(
"被动技能不支持指定禁用范围，请直接使用 开启/关闭"
).finish(reply_to=True)
return
if not all_plugins_flag.result and not plugin_names.available:
await MessageUtils.build_message("请输入功能/被动名称").finish(reply_to=True)
return
targets = await get_target_groups(
bot,
event,
session,
tag.result if tag and tag.available else None,
groups.result if groups and groups.available else None,
all_groups_flag.result,
)
if targets is None:
return
if all_plugins_flag.result:
if targets:
messages = []
for gid in targets:
messages.append(
await PluginManager.set_all_plugin_status(
status=status,
is_default=default_status.result if is_superuser else False,
group_id=gid,
is_task=task.result,
is_superuser=is_superuser,
use_su_field=use_su_field_final,
)
)
await MessageUtils.build_message("\n".join(messages)).finish(reply_to=True)
return
if is_superuser and not session.group:
result = await PluginManager.set_all_plugin_status(
status=status,
is_default=default_status.result,
group_id=None,
is_task=task.result,
is_superuser=is_superuser,
use_su_field=use_su_field_final,
)
await MessageUtils.build_message(result).finish(reply_to=True)
return
await MessageUtils.build_message("请输入目标群组").finish(reply_to=True)
return
names = plugin_names.result if plugin_names.available else ()
if isinstance(names, str):
names = (names,)
if (
not targets
and not (is_superuser and not session.group)
and not default_status.result
and block_type_val is None
):
await MessageUtils.build_message("请选择一个目标群组").finish(reply_to=True)
return
messages = []
for name in names:
name_str = str(name)
if is_superuser and default_status.result:
result = await PluginManager.set_default_status(
name_str, status, is_task=task.result
)
messages.append(result)
continue
if block_type_val is not None:
_type = BlockType.ALL
if block_type_val in ["p", "private"]:
_type = BlockType.PRIVATE
elif block_type_val in ["g", "group"]:
_type = BlockType.GROUP
result = await PluginManager.superuser_set_status(
name_str, status, _type, None, is_task=task.result
)
messages.append(result)
continue
if not targets:
if is_superuser and not session.group:
target_block_type = BlockType.ALL if not status else None
result = await PluginManager.superuser_set_status(
name_str, status, target_block_type, None, is_task=task.result
)
messages.append(result)
continue
messages.append(f"{name_str}: 请选择一个目标群组")
continue
msg = await PluginManager.batch_update_status(
name_str,
targets,
status=status,
is_task=task.result,
is_superuser=is_superuser,
is_whitelist_mode=only_flag_value,
use_su_field=use_su_field_final,
bot=bot,
)
action_name = "开启" if status else "关闭"
logger.info(
f"{action_name}操作: {name_str}, targets={targets}",
arparma.header_result,
session=session,
)
messages.append(msg)
await MessageUtils.build_message("\n".join(messages)).finish(reply_to=True)
@_status_matcher.assign("open")
async def _(
bot: Bot,
event: Event,
session: Uninfo,
arparma: Arparma,
plugin_name: Match[str],
group: Match[str],
plugin_names: Match[tuple[str, ...]],
groups: Match[tuple[str, ...]] = AlconnaMatch("groups"),
tag: Match[str] = AlconnaMatch("tag"),
task: Query[bool] = AlconnaQuery("task.value", False),
default_status: Query[bool] = AlconnaQuery("default.value", False),
all: Query[bool] = AlconnaQuery("all.value", False),
all_groups_flag: Query[bool] = AlconnaQuery("all.value", False),
all_plugins_flag: Query[bool] = AlconnaQuery("all-plugins.value", False),
only_flag: Query[bool] = AlconnaQuery("only.value", False),
use_su_field: Query[bool] = AlconnaQuery("su.value", False),
):
if not all.result and not plugin_name.available:
await MessageUtils.build_message("请输入功能/被动名称").finish(reply_to=True)
name = plugin_name.result.strip()
if session.group:
group_id = session.group.id
"""修改当前群组的数据"""
if task.result:
if all.result:
result = await PluginManager.unblock_group_all_task(group_id)
logger.info("开启所有群组被动", arparma.header_result, session=session)
else:
result = await PluginManager.unblock_group_task(name, group_id)
logger.info(
f"开启群组被动 {name}", arparma.header_result, session=session
)
elif session.user.id in bot.config.superusers and default_status.result:
"""单个插件的进群默认修改"""
result = await PluginManager.set_default_status(name, True)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
)
elif all.result:
"""所有插件"""
result = await PluginManager.set_all_plugin_status(
True, default_status.result, group_id
)
logger.info(
"开启群组中全部功能",
arparma.header_result,
session=session,
)
else:
result = await PluginManager.unblock_group_plugin(name, group_id)
logger.info(f"开启功能 {name}", arparma.header_result, session=session)
await MessageUtils.build_message(result).finish(reply_to=True)
elif session.user.id in bot.config.superusers:
"""私聊"""
group_id = group.result if group.available else None
if all.result:
if task.result:
"""关闭全局或指定群全部被动"""
if group_id:
result = await PluginManager.unblock_group_all_task(group_id)
else:
result = await PluginManager.unblock_global_all_task(
default_status.result
)
else:
result = await PluginManager.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 MessageUtils.build_message(result).finish(reply_to=True)
if default_status.result and not task.result:
result = await PluginManager.set_default_status(name, True)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
target=group_id,
)
await MessageUtils.build_message(result).finish(reply_to=True)
if task.result:
split_list = name.split()
if len(split_list) > 1:
name = split_list[0]
group_id = split_list[1]
if group_id:
result = await PluginManager.superuser_task_handle(name, group_id, True)
logger.info(
f"超级用户开启被动技能 {name}",
arparma.header_result,
session=session,
target=group_id,
)
else:
result = await PluginManager.unblock_global_task(
name, default_status.result
)
logger.info(
f"超级用户开启全局被动技能 {name}",
arparma.header_result,
session=session,
)
else:
result = await PluginManager.superuser_unblock(name, None, group_id)
logger.info(
f"超级用户开启功能 {name}",
arparma.header_result,
session=session,
target=group_id,
)
await MessageUtils.build_message(result).finish(reply_to=True)
await _handle_switch_command(
True,
bot,
event,
session,
arparma,
plugin_names,
groups,
tag,
task,
default_status,
all_groups_flag,
all_plugins_flag,
only_flag=only_flag,
use_su_field=use_su_field,
)
@_status_matcher.assign("close")
async def _(
bot: Bot,
event: Event,
session: Uninfo,
arparma: Arparma,
plugin_name: Match[str],
block_type: Match[str],
group: Match[str],
plugin_names: Match[tuple[str, ...]],
groups: Match[tuple[str, ...]] = AlconnaMatch("groups"),
tag: Match[str] = AlconnaMatch("tag"),
task: Query[bool] = AlconnaQuery("task.value", False),
default_status: Query[bool] = AlconnaQuery("default.value", False),
all: Query[bool] = AlconnaQuery("all.value", False),
all_groups_flag: Query[bool] = AlconnaQuery("all.value", False),
all_plugins_flag: Query[bool] = AlconnaQuery("all-plugins.value", False),
use_su_field: Query[bool] = AlconnaQuery("su.value", False),
):
if not all.result and not plugin_name.available:
await MessageUtils.build_message("请输入功能/被动名称").finish(reply_to=True)
name = plugin_name.result.strip()
if session.group:
group_id = session.group.id
"""修改当前群组的数据"""
if task.result:
if all.result:
result = await PluginManager.block_group_all_task(group_id)
logger.info("开启所有群组被动", arparma.header_result, session=session)
else:
result = await PluginManager.block_group_task(name, group_id)
logger.info(
f"关闭群组被动 {name}", arparma.header_result, session=session
)
elif session.user.id in bot.config.superusers and default_status.result:
"""单个插件的进群默认修改"""
result = await PluginManager.set_default_status(name, False)
logger.info(
f"超级用户开启 {name} 功能进群默认开关",
arparma.header_result,
session=session,
)
elif all.result:
"""所有插件"""
result = await PluginManager.set_all_plugin_status(
False, default_status.result, group_id
)
logger.info("关闭群组中全部功能", arparma.header_result, session=session)
else:
result = await PluginManager.block_group_plugin(name, group_id)
logger.info(f"关闭功能 {name}", arparma.header_result, session=session)
await MessageUtils.build_message(result).finish(reply_to=True)
elif session.user.id in bot.config.superusers:
group_id = group.result if group.available else None
if all.result:
if task.result:
"""关闭全局或指定群全部被动"""
if group_id:
result = await PluginManager.block_group_all_task(group_id)
else:
result = await PluginManager.block_global_all_task(
default_status.result
)
else:
result = await PluginManager.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 MessageUtils.build_message(result).finish(reply_to=True)
if default_status.result and not task.result:
result = await PluginManager.set_default_status(name, False)
logger.info(
f"超级用户关闭 {name} 功能进群默认开关",
arparma.header_result,
session=session,
target=group_id,
)
await MessageUtils.build_message(result).finish(reply_to=True)
if task.result:
split_list = name.split()
if len(split_list) > 1:
name = split_list[0]
group_id = split_list[1]
if group_id:
result = await PluginManager.superuser_task_handle(
name, group_id, False
)
logger.info(
f"超级用户关闭被动技能 {name}",
arparma.header_result,
session=session,
target=group_id,
)
else:
result = await PluginManager.block_global_task(
name, default_status.result
)
logger.info(
f"超级用户关闭全局被动技能 {name}",
arparma.header_result,
session=session,
)
else:
parsed_block_type = (
block_type.result.lower()
if block_type.available and block_type.result
else ""
)
# 兼容中文快捷命令：`关闭功能名 p/g/a`
if not parsed_block_type and " " in name:
split_name = name.rsplit(maxsplit=1)
if len(split_name) == 2 and split_name[1].lower() in {
"a",
"all",
"g",
"group",
"p",
"private",
}:
name = split_name[0].strip()
parsed_block_type = split_name[1].lower()
_type = BlockType.ALL
if parsed_block_type in {"p", "private"}:
_type = BlockType.PRIVATE
elif parsed_block_type in {"g", "group"}:
_type = BlockType.GROUP
result = await PluginManager.superuser_block(name, _type, group_id)
logger.info(
f"超级用户关闭功能 {name}, 禁用类型: {_type}",
arparma.header_result,
session=session,
target=group_id,
)
await MessageUtils.build_message(result).finish(reply_to=True)
await _handle_switch_command(
False,
bot,
event,
session,
arparma,
plugin_names,
groups,
tag,
task,
default_status,
all_groups_flag,
all_plugins_flag,
use_su_field=use_su_field,
)
@_group_status_matcher.handle()
async def _(
bot: Bot,
event: Event,
session: Uninfo,
arparma: Arparma,
status: str,
groups: Match[tuple[str, ...]] = AlconnaMatch("groups"),
tag: Match[str] = AlconnaMatch("tag"),
all_flag: Query[bool] = AlconnaQuery("all.value", False),
only_flag: Query[bool] = AlconnaQuery("only.value", False),
):
if session.group:
group_id = session.group.id
if status == "sleep":
await PluginManager.sleep(group_id)
logger.info("进行休眠", arparma.header_result, session=session)
await MessageUtils.build_message("那我先睡觉了...").finish()
else:
if await PluginManager.is_wake(group_id):
await MessageUtils.build_message("我还醒着呢！").finish()
await PluginManager.wake(group_id)
logger.info("醒来", arparma.header_result, session=session)
await MessageUtils.build_message("呜..醒来了...").finish()
return MessageUtils.build_message("群组id为空...").send()
is_wake = status == "wake"
if status == "check":
if not await SUPERUSER_PERM(bot, event):
return
try:
image = await render_group_active_status(bot)
logger.info(
"查看全服群组工作状态报表", arparma.header_result, session=session
)
await MessageUtils.build_message(image).finish(reply_to=True)
except FinishedException:
raise
except Exception as e:
logger.error(f"渲染群组激活状态报表失败: {e}", e=e)
await MessageUtils.build_message("生成状态报表失败，请检查日志").finish(
reply_to=True
)
return
targets = await get_target_groups(
bot,
event,
session,
tag.result if tag and tag.available else None,
groups.result if groups and groups.available else None,
all_flag.result,
)
if not targets:
await MessageUtils.build_message("请指定目标群组或在群聊中使用").finish(
reply_to=True
)
return
msg = await PluginManager.batch_set_group_active_status(
targets, status=is_wake, is_whitelist_mode=only_flag.result, bot=bot
)
action_name = "醒来" if is_wake else "进行休眠"
reply_msg = "呜..醒来了..." if is_wake else "那我先睡觉了..."
if len(targets) > 1 or only_flag.result:
reply_msg = msg
logger.info(action_name, arparma.header_result, session=session)
await MessageUtils.build_message(reply_msg).finish(reply_to=True)
@_status_matcher.assign("task")
@@ -380,9 +420,37 @@ async def _(
session: Uninfo,
arparma: Arparma,
):
if arparma.find("check") or arparma.find("open") or arparma.find("close"):
return
image = await build_task(session.group.id if session.group else None)
if image:
logger.info("查看群被动列表", arparma.header_result, session=session)
await MessageUtils.build_message(image).finish(reply_to=True)
else:
await MessageUtils.build_message("获取群被动任务失败...").finish(reply_to=True)
@_status_matcher.assign("check")
async def _(
bot: Bot,
event: Event,
plugin_name: Match[str],
task: Query[bool] = AlconnaQuery("task.value", False),
):
if not await SUPERUSER_PERM(bot, event):
return
name = plugin_name.result
try:
img = await render_global_status(name, is_task=task.result, bot=bot)
await MessageUtils.build_message(img).finish(reply_to=True)
except FinishedException:
raise
except ValueError as e:
await MessageUtils.build_message(str(e)).finish(reply_to=True)
except Exception as e:
logger.error(f"渲染状态图表失败: {e}", e=e)
await MessageUtils.build_message("生成状态报表失败，请检查日志").finish(
reply_to=True
)
zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
-600
View File
@@ -1,600 +0,0 @@
from typing import cast
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.models.task_info import TaskInfo
from zhenxun.services.cache import CacheRoot
from zhenxun.services.cache.runtime_cache import (
PluginInfoMemoryCache,
TaskInfoMemoryCache,
)
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.enum import BlockType, CacheType, PluginType
from zhenxun.utils.exception import GroupInfoNotFound
from zhenxun.utils.image_utils import BuildImage, ImageTemplate, RowStyle
def plugin_row_style(column: str, text: str) -> RowStyle:
"""被动技能文本风格
参数:
column: 表头
text: 文本内容
返回:
RowStyle: RowStyle
"""
style = RowStyle()
if (column == "全局状态" and text == "开启") or (
column != "全局状态" and column == "加载状态" and text == "SUCCESS"
):
style.font_color = "#67C23A"
elif column in {"全局状态", "加载状态"}:
style.font_color = "#F56C6C"
return style
async def build_plugin() -> BuildImage:
column_name = [
"ID",
"模块",
"名称",
"全局状态",
"禁用类型",
"加载状态",
"菜单分类",
"作者",
"版本",
"金币花费",
]
plugin_list = await PluginInfo.filter(plugin_type__not=PluginType.HIDDEN).all()
column_data = [
[
plugin.id,
plugin.module,
plugin.name,
"开启" if plugin.status else "关闭",
plugin.block_type,
"SUCCESS" if plugin.load_status else "ERROR",
plugin.menu_type,
plugin.author,
plugin.version,
plugin.cost_gold,
]
for plugin in plugin_list
]
return await ImageTemplate.table_page(
"Plugin",
"插件状态",
column_name,
column_data,
text_style=plugin_row_style,
)
def task_row_style(column: str, text: str) -> RowStyle:
"""被动技能文本风格
参数:
column: 表头
text: 文本内容
返回:
RowStyle: RowStyle
"""
style = RowStyle()
if column in {"群组状态", "全局状态"}:
style.font_color = "#67C23A" if text == "开启" else "#F56C6C"
return style
async def build_task(group_id: str | None) -> BuildImage:
"""构造被动技能状态图片
参数:
group_id: 群组id
异常:
GroupInfoNotFound: 未找到群组
返回:
BuildImage: 被动技能状态图片
"""
task_list = await TaskInfo.all()
column_name = ["ID", "模块", "名称", "群组状态", "全局状态", "运行时间"]
group = None
if group_id:
group = await GroupConsole.get_group_db(group_id=group_id)
if not group:
raise GroupInfoNotFound()
else:
column_name.remove("群组状态")
column_data = []
for task in task_list:
if group:
column_data.append(
[
task.id,
task.module,
task.name,
"开启" if f"<{task.module}," not in group.block_task else "关闭",
"开启" if task.status else "关闭",
task.run_time or "-",
]
)
else:
column_data.append(
[
task.id,
task.module,
task.name,
"开启" if task.status else "关闭",
task.run_time or "-",
]
)
return await ImageTemplate.table_page(
"Task",
"被动技能状态",
column_name,
column_data,
text_style=task_row_style,
)
class PluginManager:
@staticmethod
async def _get_plugin_by_name_or_module(plugin_name: str) -> PluginInfo | None:
plugin_name = plugin_name.strip()
if not plugin_name:
return None
if plugin_name.isdigit():
return await PluginInfo.get_or_none(id=int(plugin_name))
plugin = await PluginInfo.get_or_none(
name=plugin_name, load_status=True, plugin_type__not=PluginType.PARENT
)
if plugin:
return plugin
return await PluginInfo.get_or_none(
module=plugin_name, load_status=True, plugin_type__not=PluginType.PARENT
)
@classmethod
async def set_default_status(cls, plugin_name: str, status: bool) -> str:
"""设置插件进群默认状态
参数:
plugin_name: 插件名称
status: 状态
返回:
str: 返回信息
"""
plugin = await cls._get_plugin_by_name_or_module(plugin_name)
if plugin:
plugin.default_status = status
await plugin.save(update_fields=["default_status"])
status_text = "开启" if status else "关闭"
return f"成功将 {plugin.name} 进群默认状态修改为: {status_text}"
return "没有找到这个功能喔..."
@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
)
await PluginInfoMemoryCache.refresh()
return f"成功将所有功能进群默认状态修改为: {'开启' if status else '关闭'}"
if group_id:
if group := await GroupConsole.get_group_db(group_id=group_id):
module_list = cast(
list[str],
await PluginInfo.filter(plugin_type=PluginType.NORMAL).values_list(
"module", flat=True
),
)
if status:
# 开启所有功能 - 清空禁用列表
group.block_plugin = ""
else:
# 关闭所有功能 - 将模块列表转换为禁用格式
group.block_plugin = CommonUtils.convert_module_format(module_list)
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=None if status else BlockType.ALL
)
await CacheRoot.invalidate_cache(CacheType.PLUGINS)
await PluginInfoMemoryCache.refresh()
return f"成功将所有功能全局状态修改为: {'开启' if status else '关闭'}"
@classmethod
async def is_wake(cls, group_id: str) -> bool:
"""是否醒来
参数:
group_id: 群组id
返回:
bool: 是否醒来
"""
if c := await GroupConsole.get_group_db(group_id=group_id):
return c.status
return False
@classmethod
async def sleep(cls, group_id: str):
"""休眠
参数:
group_id: 群组id
"""
group, _ = await GroupConsole.get_or_create(
group_id=group_id, channel_id__isnull=True
)
group.status = False
await group.save(update_fields=["status"])
@classmethod
async def wake(cls, group_id: str):
"""醒来
参数:
group_id: 群组id
"""
group, _ = await GroupConsole.get_or_create(
group_id=group_id, channel_id__isnull=True
)
group.status = True
await group.save(update_fields=["status"])
@classmethod
async def block(cls, module: str):
"""禁用
参数:
module: 模块名
"""
if plugin := await PluginInfo.get_plugin(module=module):
plugin.status = False
await plugin.save(update_fields=["status"])
@classmethod
async def unblock(cls, module: str):
"""启用
参数:
module: 模块名
"""
if plugin := await PluginInfo.get_plugin(module=module):
plugin.status = True
await plugin.save(update_fields=["status"])
@classmethod
async def block_group_plugin(cls, plugin_name: str, group_id: str) -> str:
"""禁用群组插件
参数:
plugin_name: 插件名称
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_plugin(plugin_name, group_id, False)
@classmethod
async def unblock_group_task(cls, task_name: str, group_id: str) -> str:
"""启用被动技能
参数:
task_name: 被动技能名称
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_task(task_name, group_id, False)
@classmethod
async def unblock_group_all_task(cls, group_id: str) -> str:
"""启用被动技能
参数:
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_task("", group_id, False, True)
@classmethod
async def block_group_task(cls, task_name: str, group_id: str) -> str:
"""禁用被动技能
参数:
task_name: 被动技能名称
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_task(task_name, group_id, True)
@classmethod
async def block_group_all_task(cls, group_id: str) -> str:
"""禁用被动技能
参数:
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_task("", group_id, True, True)
@classmethod
async def block_global_all_task(cls, is_default: bool) -> str:
"""禁用全局被动技能
返回:
str: 返回信息
"""
if is_default:
await TaskInfo.all().update(default_status=False)
await TaskInfoMemoryCache.refresh()
return "已禁用所有被动进群默认状态"
else:
await TaskInfo.all().update(status=False)
await TaskInfoMemoryCache.refresh()
return "已全局禁用所有被动状态"
@classmethod
async def block_global_task(cls, name: str, is_default: bool = False) -> str:
"""禁用全局被动技能
参数:
name: 被动技能名称
返回:
str: 返回信息
"""
if is_default:
await TaskInfo.filter(name=name).update(default_status=False)
await TaskInfoMemoryCache.refresh()
return f"已禁用被动进群默认状态 {name}"
else:
await TaskInfo.filter(name=name).update(status=False)
await TaskInfoMemoryCache.refresh()
return f"已全局禁用被动状态 {name}"
@classmethod
async def unblock_global_all_task(cls, is_default: bool) -> str:
"""开启全局被动技能
参数:
is_default: 是否为默认状态
返回:
str: 返回信息
"""
if is_default:
await TaskInfo.all().update(default_status=True)
await TaskInfoMemoryCache.refresh()
return "已开启所有被动进群默认状态"
else:
await TaskInfo.all().update(status=True)
await TaskInfoMemoryCache.refresh()
return "已全局开启所有被动状态"
@classmethod
async def unblock_global_task(cls, name: str, is_default: bool = False) -> str:
"""开启全局被动技能
参数:
name: 被动技能名称
is_default: 是否为默认状态
返回:
str: 返回信息
"""
if is_default:
await TaskInfo.filter(name=name).update(default_status=True)
await TaskInfoMemoryCache.refresh()
return f"已开启被动进群默认状态 {name}"
else:
await TaskInfo.filter(name=name).update(status=True)
await TaskInfoMemoryCache.refresh()
return f"已全局开启被动状态 {name}"
@classmethod
async def unblock_group_plugin(cls, plugin_name: str, group_id: str) -> str:
"""启用群组插件
参数:
plugin_name: 插件名称
group_id: 群组id
返回:
str: 返回信息
"""
return await cls._change_group_plugin(plugin_name, group_id, True)
@classmethod
async def _change_group_task(
cls, task_name: str, group_id: str, status: bool, is_all: bool = False
) -> str:
"""改变群组被动技能状态
参数:
task_name: 被动技能名称
group_id: 群组Id
status: 状态，为True时是关闭
is_all: 所有群被动
返回:
str: 返回信息
"""
status_str = "关闭" if status else "开启"
if is_all:
module_list = cast(
list[str], await TaskInfo.annotate().values_list("module", flat=True)
)
if module_list:
group, _ = await GroupConsole.get_or_create(
group_id=group_id, channel_id__isnull=True
)
if status:
group.block_task = CommonUtils.convert_module_format(module_list)
else:
# 开启所有模块 - 清空禁用列表
group.block_task = ""
await group.save(update_fields=["block_task"])
return f"已成功{status_str}全部被动技能!"
elif task := await TaskInfo.get_or_none(name=task_name):
if status:
await GroupConsole.set_block_task(group_id, task.module)
elif await GroupConsole.is_superuser_block_task(group_id, task.module):
return f"{status_str} {task_name} 被动技能失败，当前群组该被动已被管理员禁用" # noqa: E501
else:
await GroupConsole.set_unblock_task(group_id, task.module)
return f"已成功{status_str} {task_name} 被动技能!"
return "没有找到这个被动技能喔..."
@classmethod
async def _change_group_plugin(
cls, plugin_name: str, group_id: str, status: bool
) -> str:
"""修改群组插件状态
参数:
plugin_name: 插件名称
group_id: 群组id
status: 插件状态
返回:
str: 返回信息
"""
plugin = await cls._get_plugin_by_name_or_module(plugin_name)
if plugin:
status_str = "开启" if status else "关闭"
if status:
if await GroupConsole.is_normal_block_plugin(group_id, plugin.module):
await GroupConsole.set_unblock_plugin(group_id, plugin.module)
return f"已成功{status_str} {plugin.name} 功能!"
elif not await GroupConsole.is_normal_block_plugin(group_id, plugin.module):
await GroupConsole.set_block_plugin(group_id, plugin.module)
return f"已成功{status_str} {plugin.name} 功能!"
return f"该功能已经{status_str}了喔，不要重复{status_str}..."
return "没有找到这个功能喔..."
@classmethod
async def superuser_task_handle(
cls, task_name: str, group_id: str | None, status: bool
) -> str:
"""超级用户禁用被动技能
参数:
task_name: 被动技能名称
group_id: 群组id
status: 状态
返回:
str: 返回信息
"""
if not (task := await TaskInfo.get_or_none(name=task_name)):
return "没有找到这个功能喔..."
if group_id:
if status:
await GroupConsole.set_unblock_task(group_id, task.module, True)
else:
await GroupConsole.set_block_task(group_id, task.module, True)
status_str = "开启" if status else "关闭"
return f"已成功将群组 {group_id} 被动技能 {task_name} {status_str}!"
return "没有找到这个群组喔..."
@classmethod
async def superuser_block(
cls, plugin_name: str, block_type: BlockType | None, group_id: str | None
) -> str:
"""超级用户禁用插件
参数:
plugin_name: 插件名称
block_type: 禁用类型
group_id: 群组id
返回:
str: 返回信息
"""
plugin = await cls._get_plugin_by_name_or_module(plugin_name)
if plugin:
if group_id:
if not await GroupConsole.is_superuser_block_plugin(
group_id, plugin.module
):
await GroupConsole.set_block_plugin(group_id, plugin.module, True)
return f"已成功关闭群组 {group_id} 的 {plugin_name} 功能!"
return "此群组该功能已被超级用户关闭，不要重复关闭..."
plugin.block_type = block_type
plugin.status = not bool(block_type)
await plugin.save(update_fields=["status", "block_type"])
if not block_type:
return f"已成功将 {plugin.name} 全局启用!"
if block_type == BlockType.ALL:
return f"已成功将 {plugin.name} 全局关闭!"
if block_type == BlockType.GROUP:
return f"已成功将 {plugin.name} 全局群组关闭!"
if block_type == BlockType.PRIVATE:
return f"已成功将 {plugin.name} 全局私聊关闭!"
return "没有找到这个功能喔..."
@classmethod
async def superuser_unblock(
cls, plugin_name: str, block_type: BlockType | None, group_id: str | None
) -> str:
"""超级用户开启插件
参数:
plugin_name: 插件名称
block_type: 禁用类型
group_id: 群组id
返回:
str: 返回信息
"""
plugin = await cls._get_plugin_by_name_or_module(plugin_name)
if plugin:
if group_id:
if await GroupConsole.is_superuser_block_plugin(
group_id, plugin.module
):
await GroupConsole.set_unblock_plugin(group_id, plugin.module, True)
return f"已成功开启群组 {group_id} 的 {plugin_name} 功能!"
return "此群组该功能已被超级用户开启，不要重复开启..."
plugin.block_type = block_type
plugin.status = not bool(block_type)
await plugin.save(update_fields=["status", "block_type"])
if not block_type:
return f"已成功将 {plugin.name} 全局启用!"
if block_type == BlockType.ALL:
return f"已成功将 {plugin.name} 全局开启!"
if block_type == BlockType.GROUP:
return f"已成功将 {plugin.name} 全局群组开启!"
if block_type == BlockType.PRIVATE:
return f"已成功将 {plugin.name} 全局私聊开启!"
return "没有找到这个功能喔..."
zhenxun/builtin_plugins/admin/plugin_switch/command.py
+66 -121
View File
@@ -2,31 +2,46 @@ from nonebot.rule import to_me
from nonebot_plugin_alconna import (
Alconna,
Args,
MultiVar,
Option,
Subcommand,
on_alconna,
store_true,
)
from zhenxun.utils.rules import admin_check, ensure_group
from zhenxun.utils.rules import admin_check
_status_matcher = on_alconna(
Alconna(
"switch",
Option("-t|--task", action=store_true, help_text="被动技能"),
Option("--task", action=store_true, help_text="被动技能"),
Option("-df|--default", action=store_true, help_text="进群默认开关"),
Option("--all", action=store_true, help_text="全部插件/被动"),
Option("-g|--group", Args["group?", str], help_text="指定群组"),
Option("--all-plugins", action=store_true, help_text="所有插件/功能"),
Option("--all", action=store_true, help_text="所有群组 (超级用户专用)"),
Option("-g|--group", Args["groups", MultiVar(str)], help_text="指定群组"),
Option("-t|--tag", Args["tag", str], help_text="指定标签"),
Option("-o|--only", action=store_true, help_text="白名单模式(仅在目标群开启)"),
Option("-s|--su", action=store_true, help_text="操作超级用户专用字段"),
Subcommand(
"check",
Args["plugin_name", [str, int]],
),
Subcommand(
"open",
Args["plugin_name?", [str, int]],
Args["plugin_names?", MultiVar(str)],
Option(
"--type",
Args["block_type?", ["all", "a", "private", "p", "group", "g"]],
help_text="全局禁用范围",
),
),
Subcommand(
"close",
Args["plugin_name?", [str, int]],
Args["plugin_names?", MultiVar(str)],
Option(
"-t|--type",
"--type",
Args["block_type?", ["all", "a", "private", "p", "group", "g"]],
help_text="全局禁用范围",
),
),
),
@@ -36,10 +51,15 @@ _status_matcher = on_alconna(
)
_group_status_matcher = on_alconna(
Alconna("group-status", Args["status", ["sleep", "wake"]]),
rule=admin_check("plugin_switch", "CHANGE_GROUP_SWITCH_LEVEL")
& ensure_group
& to_me(),
Alconna(
"group-status",
Args["status", ["sleep", "wake", "check"]],
Option("-g|--group", Args["groups", MultiVar(str)], help_text="指定群组"),
Option("-t|--tag", Args["tag", str], help_text="指定标签"),
Option("--all", action=store_true, help_text="所有群组"),
Option("-o|--only", action=store_true, help_text="白名单模式(仅在目标群醒来)"),
),
rule=admin_check("plugin_switch", "CHANGE_GROUP_SWITCH_LEVEL") & to_me(),
priority=5,
block=True,
)
@@ -52,124 +72,42 @@ _status_matcher.shortcut(
)
_status_matcher.shortcut(
r"群被动状态",
r"查看(功能|插件)?状态",
command="switch check {*}",
prefix=True,
)
_status_matcher.shortcut(
r"查看(群)?被动状态",
command="switch check {*} --task",
prefix=True,
)
_status_matcher.shortcut(
r"(群)?被动状态",
command="switch",
arguments=["--task"],
prefix=True,
)
_status_matcher.shortcut(
r"开启(所有|全部)默认群被动",
command="switch",
arguments=["open", "--task", "--all", "-df"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭(所有|全部)默认群被动",
command="switch",
arguments=["close", "--task", "--all", "-df"],
prefix=True,
)
_status_matcher.shortcut(
r"开启群被动\s*(?P<name>.+)",
command="switch",
arguments=["open", "{name}", "--task"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭群被动\s*(?P<name>.+)",
command="switch",
arguments=["close", "{name}", "--task"],
prefix=True,
)
_status_matcher.shortcut(
r"开启默认群被动\s*(?P<name>.+)",
command="switch",
arguments=["open", "{name}", "--task", "-df"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭默认群被动\s*(?P<name>.+)",
command="switch",
arguments=["close", "{name}", "--task", "-df"],
prefix=True,
)
def _switch_wrapper(slot: str, content: str | None, context: dict) -> str:
"""动态映射转换函数"""
if slot == "action":
return "open" if content == "开启" else "close"
if slot == "all" and content:
return "--all-plugins"
if slot == "default" and content:
return "-df"
if slot == "type" and content:
return "--task" if "被动" in content else ""
return ""
_status_matcher.shortcut(
r"开启(所有|全部)群被动",
command="switch",
arguments=["open", "--task", "--all"],
prefix=True,
)
_status_matcher.shortcut(
r"关闭(所有|全部)群被动",
command="switch",
arguments=["close", "--task", "--all"],
prefix=True,
)
_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",
arguments=["open", "{name}"],
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",
arguments=["close", "{name}"],
r"^(?P<action>开启|关闭)\s*(?P<all>所有|全部)?\s*(?P<default>默认)?\s*(?P<type>群被动|被动|插件|功能)?\s*",
command="switch {all} {default} {type} {action} {* }",
wrapper=_switch_wrapper, # type: ignore
prefix=True,
)
@@ -182,8 +120,15 @@ _group_status_matcher.shortcut(
)
_group_status_matcher.shortcut(
r"休息吧",
r"休息(吧)?",
command="group-status",
arguments=["sleep"],
prefix=True,
)
_group_status_matcher.shortcut(
r"查看群(状态|信息)",
command="group-status",
arguments=["check"],
prefix=True,
)
zhenxun/builtin_plugins/admin/plugin_switch/data_source.py
+325
View File
@@ -0,0 +1,325 @@
from nonebot.adapters import Bot
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.cache import CacheRoot
from zhenxun.services.cache.runtime_cache import GroupMemoryCache
from zhenxun.utils.common_utils import CommonUtils
from zhenxun.utils.enum import BlockType, CacheType
from zhenxun.utils.platform import PlatformUtils
from .strategy import get_strategy
class PluginManager:
@staticmethod
def _modify_block_string(current_str: str, module: str, add: bool) -> str:
"""辅助: 添加或移除禁用模块字符串"""
items = CommonUtils.convert_module_format(current_str)
if add:
if module not in items:
items.append(module)
else:
if module in items:
items.remove(module)
return CommonUtils.convert_module_format(items)
@classmethod
async def _calculate_affected_groups(
cls,
target_groups: set[str],
status: bool,
is_whitelist_mode: bool,
bot: Bot | None,
) -> tuple[set[str], set[str]]:
"""提取公用的目标群组计算逻辑（白名单/普通模式交并集）"""
groups_to_open = set()
groups_to_close = set()
clean_targets = {str(gid) for gid in target_groups if gid}
if is_whitelist_mode and status:
if bot:
active_groups, _ = await PlatformUtils.get_group_list(
bot, only_group=True
)
all_group_set = {str(g.group_id) for g in active_groups if g.group_id}
else:
all_group_ids = await GroupConsole.all().values_list(
"group_id", flat=True
)
all_group_set = {str(gid) for gid in all_group_ids}
groups_to_open = clean_targets
groups_to_close = all_group_set - clean_targets
else:
if status:
groups_to_open = clean_targets
else:
groups_to_close = clean_targets
return groups_to_open, groups_to_close
@classmethod
async def batch_update_status(
cls,
name: str,
target_groups: set[str],
status: bool,
is_task: bool = False,
is_superuser: bool = False,
is_whitelist_mode: bool = False,
bot: Bot | None = None,
use_su_field: bool = False,
) -> str:
"""批量更新状态 (已用策略模式完全重构)"""
strategy = get_strategy(is_task)
entity = await strategy.get_entity(name)
if not entity:
return f"未找到{strategy.entity_type_name}: {name}"
module_name = entity.module
norm_field = strategy.norm_field
su_field = strategy.su_field
groups_to_open, groups_to_close = await cls._calculate_affected_groups(
target_groups, status, is_whitelist_mode, bot
)
affected_ids = groups_to_open | groups_to_close
if not affected_ids:
return "没有目标群组需要操作。"
for gid in groups_to_open | groups_to_close:
platform = bot.adapter.get_name() if bot else "qq"
await GroupConsole.get_or_create(
group_id=gid, defaults={"platform": platform}
)
groups_obj = await GroupConsole.filter(group_id__in=list(affected_ids)).all()
update_list = []
opened_groups: set[str] = set()
closed_groups: set[str] = set()
for group in groups_obj:
gid = str(group.group_id)
norm_val = getattr(group, norm_field)
su_val = getattr(group, su_field)
new_norm_val, new_su_val = norm_val, su_val
is_changed = False
change_type = None
if gid in groups_to_open:
new_norm_val = cls._modify_block_string(norm_val, module_name, False)
if is_superuser:
new_su_val = cls._modify_block_string(su_val, module_name, False)
if norm_val != new_norm_val or su_val != new_su_val:
is_changed = True
change_type = "open"
elif gid in groups_to_close:
if is_superuser and use_su_field:
new_su_val = cls._modify_block_string(su_val, module_name, True)
else:
new_norm_val = cls._modify_block_string(norm_val, module_name, True)
if norm_val != new_norm_val or su_val != new_su_val:
is_changed = True
change_type = "close"
if is_changed:
setattr(group, norm_field, new_norm_val)
setattr(group, su_field, new_su_val)
update_list.append(group)
if change_type == "open":
opened_groups.add(gid)
elif change_type == "close":
closed_groups.add(gid)
if update_list:
await GroupConsole.bulk_update(
update_list, [norm_field, su_field], batch_size=500
)
await CacheRoot.clear(CacheType.GROUPS)
for group in update_list:
await GroupMemoryCache.upsert_from_model(group)
item_str = strategy.entity_type_name
mode_str = "(白名单模式)" if is_whitelist_mode else ""
if not update_list:
if is_whitelist_mode:
return f"目标群组的 {item_str} {name} 已符合白名单配置，无需重复操作。"
status_desc = "开启" if status else ("系统禁用" if use_su_field else "关闭")
return (
f"目标群组的 {item_str} {name} 均已处于 {status_desc} 状态，"
"无需重复操作。"
)
opened_count, closed_count = len(opened_groups), len(closed_groups)
if status:
su_hint = " (已同步解除系统禁用)" if is_superuser else ""
success_msg = f"已开启 {opened_count} 个群组的 {item_str} {name}{su_hint}"
else:
if is_superuser and use_su_field:
success_msg = f"已系统级禁用 {closed_count} 个群组的 {item_str} {name}"
else:
success_msg = f"已在 {closed_count} 个群组中关闭了 {item_str} {name}"
if is_whitelist_mode:
msg_parts = []
if opened_count > 0:
msg_parts.append(f"已开启 {opened_count} 个群组")
if closed_count > 0:
msg_parts.append(f"已关闭 {closed_count} 个群组")
return f"{'，'.join(msg_parts)} 的 {item_str} {name} {mode_str}。"
return f"{success_msg}。"
@classmethod
async def set_default_status(
cls, plugin_name: str, status: bool, is_task: bool = False
) -> str:
strategy = get_strategy(is_task)
entity = await strategy.get_entity(plugin_name)
if entity:
await strategy.set_default_status(entity, status)
status_text = "开启" if status else "关闭"
return (
f"成功将 {getattr(entity, 'name', plugin_name)} "
f"进群默认状态修改为: {status_text}"
)
return "没有找到这个功能喔..."
@classmethod
async def set_all_plugin_status(
cls,
status: bool,
is_default: bool = False,
group_id: str | None = None,
is_task: bool = False,
is_superuser: bool = False,
use_su_field: bool = False,
) -> str:
strategy = get_strategy(is_task)
type_str = strategy.entity_type_name
if is_default:
await strategy.set_all_default_status(status)
return (
f"成功将所有{type_str}进群默认状态修改为: "
f"{'开启' if status else '关闭'}"
)
if group_id:
if group := await GroupConsole.get_group_db(group_id=group_id):
norm_field = strategy.norm_field
su_field = strategy.su_field
module_list = await strategy.get_all_modules()
all_modules_str = CommonUtils.convert_module_format(module_list)
update_fields = []
if status:
if is_superuser:
setattr(group, norm_field, "")
setattr(group, su_field, "")
update_fields.extend([norm_field, su_field])
msg = f"成功将此群组所有{type_str}完全开启 (包括解除系统禁用)"
else:
setattr(group, norm_field, "")
update_fields.append(norm_field)
msg = f"成功开启此群组所有{type_str}"
else:
if is_superuser and use_su_field:
setattr(group, su_field, all_modules_str)
update_fields.append(su_field)
msg = f"已由超级用户系统级禁用此群组所有{type_str}"
else:
setattr(group, norm_field, all_modules_str)
update_fields.append(norm_field)
msg = f"成功关闭此群组所有{type_str}"
await group.save(update_fields=update_fields)
return f"{msg}。"
return "获取群组失败..."
await strategy.set_all_global_status(status)
return f"成功将所有{type_str}全局状态修改为: {'开启' if status else '关闭'}"
@classmethod
async def superuser_set_status(
cls,
plugin_name: str,
status: bool,
block_type: BlockType | None,
group_id: str | None,
is_task: bool = False,
) -> str:
strategy = get_strategy(is_task)
entity = await strategy.get_entity(plugin_name)
action_cn = "开启" if status else "关闭"
if entity:
if group_id:
is_su_blocked, _ = await strategy.check_block_status(
group_id, entity.module
)
if status and is_su_blocked:
await cls.batch_update_status(
plugin_name,
{group_id},
True,
is_task=is_task,
is_superuser=True,
)
return f"已成功{action_cn}群组 {group_id} 的 {plugin_name} 功能!"
if not status and not is_su_blocked:
await cls.batch_update_status(
plugin_name,
{group_id},
False,
is_task=is_task,
is_superuser=True,
use_su_field=True,
)
return f"已成功{action_cn}群组 {group_id} 的 {plugin_name} 功能!"
return f"此群组该功能已被超级用户{action_cn}，不要重复操作..."
await strategy.set_global_status(entity, status, block_type)
await strategy.refresh_cache()
if not block_type or block_type == BlockType.ALL:
return f"已成功将 {entity.name} 全局{action_cn}!"
if block_type == BlockType.GROUP:
return f"已成功将 {entity.name} 全局群组{action_cn}!"
if block_type == BlockType.PRIVATE:
return f"已成功将 {entity.name} 全局私聊{action_cn}!"
return "没有找到这个功能喔..."
@classmethod
async def batch_set_group_active_status(
cls,
target_groups: set[str],
status: bool,
is_whitelist_mode: bool = False,
bot: Bot | None = None,
) -> str:
"""批量设置群组激活状态 (休眠/醒来) - 采用与插件相同的目标计算逻辑"""
groups_to_wake, groups_to_sleep = await cls._calculate_affected_groups(
target_groups, status, is_whitelist_mode, bot
)
affected_ids = groups_to_wake | groups_to_sleep
if not affected_ids:
return "没有目标群组需要操作。"
if groups_to_wake:
await GroupConsole.filter(group_id__in=list(groups_to_wake)).update(
status=True
)
if groups_to_sleep:
await GroupConsole.filter(group_id__in=list(groups_to_sleep)).update(
status=False
)
await CacheRoot.clear(CacheType.GROUPS)
await GroupMemoryCache.refresh()
action_str = "醒来" if status else "休眠"
return f"已完成目标群组的 {action_str} 操作。"
zhenxun/builtin_plugins/admin/plugin_switch/strategy.py
+187
View File
@@ -0,0 +1,187 @@
from abc import ABC, abstractmethod
from typing import Any, cast
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.models.task_info import TaskInfo
from zhenxun.services.cache import CacheRoot
from zhenxun.services.cache.runtime_cache import (
PluginInfoMemoryCache,
TaskInfoMemoryCache,
)
from zhenxun.utils.enum import BlockType, CacheType, PluginType
class SwitchStrategy(ABC):
"""插件与被动技能切换策略基类"""
@property
@abstractmethod
def entity_type_name(self) -> str:
pass
@property
@abstractmethod
def norm_field(self) -> str:
"""普通的群组禁用字段名"""
pass
@property
@abstractmethod
def su_field(self) -> str:
"""超级用户群组禁用字段名"""
pass
@abstractmethod
async def get_entity(self, name: str) -> Any:
"""通过名称获取实体信息"""
pass
@abstractmethod
async def check_block_status(self, group_id: str, module: str) -> tuple[bool, bool]:
"""检查目标群组的禁用状态，返回 (is_su_blocked, is_norm_blocked)"""
pass
@abstractmethod
async def get_all_modules(self) -> list[str]:
"""获取所有模块的名称列表"""
pass
@abstractmethod
async def set_default_status(self, entity: Any, status: bool) -> None:
"""设置单个实体的进群默认状态"""
pass
@abstractmethod
async def set_global_status(
self, entity: Any, status: bool, block_type: BlockType | None = None
) -> None:
"""设置单个实体的全局状态"""
pass
@abstractmethod
async def set_all_default_status(self, status: bool) -> None:
"""设置所有实体的进群默认状态"""
pass
@abstractmethod
async def set_all_global_status(self, status: bool) -> None:
"""设置所有实体的全局状态"""
pass
@abstractmethod
async def refresh_cache(self) -> None:
"""刷新相关的内存缓存"""
pass
class PluginStrategy(SwitchStrategy):
@property
def entity_type_name(self) -> str:
return "功能"
@property
def norm_field(self) -> str:
return "block_plugin"
@property
def su_field(self) -> str:
return "superuser_block_plugin"
async def get_entity(self, name: str) -> Any:
if name.isdigit():
return await PluginInfo.get_or_none(id=int(name))
return await PluginInfo.get_or_none(
name=name, load_status=True, plugin_type__not=PluginType.PARENT
)
async def check_block_status(self, group_id: str, module: str) -> tuple[bool, bool]:
is_su_blocked = await GroupConsole.is_superuser_block_plugin(group_id, module)
is_norm_blocked = await GroupConsole.is_normal_block_plugin(group_id, module)
return is_su_blocked, is_norm_blocked
async def get_all_modules(self) -> list[str]:
return cast(
list[str],
await PluginInfo.filter(plugin_type=PluginType.NORMAL).values_list(
"module", flat=True
),
)
async def set_default_status(self, entity: PluginInfo, status: bool) -> None:
entity.default_status = status
await entity.save(update_fields=["default_status"])
async def set_global_status(
self, entity: PluginInfo, status: bool, block_type: BlockType | None = None
) -> None:
entity.block_type = block_type
entity.status = not bool(block_type)
await entity.save(update_fields=["status", "block_type"])
async def set_all_default_status(self, status: bool) -> None:
await PluginInfo.filter(plugin_type=PluginType.NORMAL).update(
default_status=status
)
await self.refresh_cache()
async def set_all_global_status(self, status: bool) -> None:
await PluginInfo.filter(plugin_type=PluginType.NORMAL).update(
status=status, block_type=None if status else BlockType.ALL
)
await self.refresh_cache()
async def refresh_cache(self) -> None:
await CacheRoot.invalidate_cache(CacheType.PLUGINS)
await PluginInfoMemoryCache.refresh()
class TaskStrategy(SwitchStrategy):
@property
def entity_type_name(self) -> str:
return "被动"
@property
def norm_field(self) -> str:
return "block_task"
@property
def su_field(self) -> str:
return "superuser_block_task"
async def get_entity(self, name: str) -> Any:
return await TaskInfo.get_or_none(name=name)
async def check_block_status(self, group_id: str, module: str) -> tuple[bool, bool]:
is_su_blocked = await GroupConsole.is_superuser_block_task(group_id, module)
is_norm_blocked = await GroupConsole.is_block_task(group_id, module)
return is_su_blocked, is_norm_blocked
async def get_all_modules(self) -> list[str]:
return cast(list[str], await TaskInfo.all().values_list("module", flat=True))
async def set_default_status(self, entity: TaskInfo, status: bool) -> None:
entity.default_status = status
await entity.save(update_fields=["default_status"])
async def set_global_status(
self, entity: TaskInfo, status: bool, block_type: BlockType | None = None
) -> None:
entity.status = status
await entity.save(update_fields=["status"])
async def set_all_default_status(self, status: bool) -> None:
await TaskInfo.all().update(default_status=status)
await self.refresh_cache()
async def set_all_global_status(self, status: bool) -> None:
await TaskInfo.all().update(status=status)
await self.refresh_cache()
async def refresh_cache(self) -> None:
await TaskInfoMemoryCache.refresh()
def get_strategy(is_task: bool) -> SwitchStrategy:
"""工厂方法：获取对应的处理策略"""
return TaskStrategy() if is_task else PluginStrategy()
zhenxun/builtin_plugins/admin/plugin_switch/ui.py
+454
View File
@@ -0,0 +1,454 @@
from typing import Any
from nonebot.adapters import Bot
from zhenxun import ui
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.models.task_info import TaskInfo
from zhenxun.ui.models import LayoutData, StatusBadgeCell, TextCell
from zhenxun.utils.enum import PluginType
from zhenxun.utils.exception import GroupInfoNotFound
from zhenxun.utils.platform import PlatformUtils
from .strategy import get_strategy
async def build_plugin() -> bytes:
"""构造插件状态图片"""
column_name = [
"ID",
"模块",
"名称",
"全局状态",
"禁用类型",
"加载状态",
"菜单分类",
"作者",
"版本",
"金币花费",
]
plugin_list = await PluginInfo.filter(plugin_type__not=PluginType.HIDDEN).all()
rows = []
for plugin in plugin_list:
status_cell = StatusBadgeCell(
text="开启" if plugin.status else "关闭",
status_type="ok" if plugin.status else "error",
)
load_cell = StatusBadgeCell(
text="SUCCESS" if plugin.load_status else "ERROR",
status_type="ok" if plugin.load_status else "error",
)
rows.append(
[
plugin.id,
plugin.module,
plugin.name,
status_cell,
plugin.block_type.value if plugin.block_type else "-",
load_cell,
plugin.menu_type or "-",
plugin.author or "-",
plugin.version or "-",
plugin.cost_gold,
]
)
table = ui.table("Plugin List", "插件状态概览")
table.set_headers(column_name)
table.add_rows(rows)
table.set_column_widths(
[
"60px",
"150px",
"150px",
"80px",
"100px",
"100px",
"100px",
"100px",
"80px",
"80px",
]
)
return await ui.render(table, viewport={"width": 1400, "height": 10})
async def build_task(group_id: str | None) -> bytes:
"""构造被动技能状态图片"""
task_list = await TaskInfo.all()
column_name = ["ID", "模块", "名称", "群组状态", "全局状态", "运行时间"]
group = None
if group_id:
group = await GroupConsole.get_group_db(group_id=group_id)
if not group:
raise GroupInfoNotFound()
else:
column_name.remove("群组状态")
rows = []
for task in task_list:
global_status_cell = StatusBadgeCell(
text="开启" if task.status else "关闭",
status_type="ok" if task.status else "error",
)
row = [task.id, task.module, task.name]
if group:
is_group_open = f"<{task.module}," not in group.block_task
group_status_cell = StatusBadgeCell(
text="开启" if is_group_open else "关闭",
status_type="ok" if is_group_open else "error",
)
row.append(group_status_cell)
row.extend([global_status_cell, task.run_time or "-"])
rows.append(row)
table = ui.table("Task List", "被动技能状态概览")
table.set_headers(column_name)
table.add_rows(rows)
if group:
table.set_column_widths(["60px", "150px", "150px", "100px", "100px", "auto"])
viewport_width = 1200
else:
table.set_column_widths(["60px", "150px", "150px", "100px", "auto"])
viewport_width = 1000
return await ui.render(table, viewport={"width": viewport_width, "height": 10})
async def render_global_status(name: str, is_task: bool, bot: Bot) -> bytes:
"""渲染全局状态报表，含差异化过滤和双栏展示"""
strategy = get_strategy(is_task)
info = await strategy.get_entity(name)
if not info:
raise ValueError(f"未找到{strategy.entity_type_name}: {name}")
module = info.module
default_status = info.status
online_groups, _ = await PlatformUtils.get_group_list(bot)
valid_keys = {(str(g.group_id), g.channel_id) for g in online_groups}
all_db_groups = await GroupConsole.all()
target_groups = [
g for g in all_db_groups if (str(g.group_id), g.channel_id) in valid_keys
]
total_count = len(target_groups)
status_data = []
for group in target_groups:
gid = str(group.group_id)
is_su_blocked, is_norm_blocked = await strategy.check_block_status(gid, module)
is_open = bool(default_status) and not is_su_blocked and not is_norm_blocked
if not default_status:
status_text, badge_color = "全局关闭", "error"
elif is_su_blocked:
status_text, badge_color = "系统禁用", "error"
elif is_norm_blocked:
status_text, badge_color = "群内关闭", "warning"
else:
status_text, badge_color = "开启", "success"
status_data.append(
{
"id": str(group.group_id),
"name": group.group_name,
"status": is_open,
"status_text": status_text,
"badge_color": badge_color,
}
)
open_list = [item for item in status_data if item["status"]]
close_list = [item for item in status_data if not item["status"]]
open_count = len(open_list)
open_rate = open_count / total_count if total_count > 0 else 0
global_alert = None
if not default_status:
global_alert = ui.alert(
"全局已禁用",
f"{strategy.entity_type_name} [{name}] 当前处于全局关闭状态。",
type="error",
)
display_list = []
list_title = "群组状态详情"
if total_count > 0 and default_status:
if open_rate > 0.9:
display_list, list_title = (
close_list,
f"异常状态列表 (其余 {open_count} 个群均正常开启)",
)
elif open_rate < 0.1:
display_list, list_title = (
open_list,
f"异常状态列表 (其余 {len(close_list)} 个群均已禁用)",
)
else:
display_list = sorted(status_data, key=lambda x: not x["status"])
return await build_dashboard_report(
page_title=f"{strategy.entity_type_name}状态报告: {name}",
total_count=total_count,
active_count=open_count,
inactive_count=len(close_list),
active_rate=open_rate,
active_label="已开启",
active_color="var(--color-accent-green)",
inactive_label="已关闭",
inactive_color="var(--color-accent-red)",
progress_label=f"功能 [{name}] 全局覆盖率",
summary_tip=(
f"总群数: {total_count} | 🟢 开启: {open_count} | "
f"🔴 关闭: {len(close_list)}"
),
display_list=display_list,
list_title=list_title,
global_alert=global_alert,
perfect_state_alert=ui.alert(
"状态完美", f"所有 {total_count} 个群组状态一致。", type="success"
)
if not global_alert
else None,
)
async def render_group_active_status(bot: Bot) -> bytes:
"""渲染群组醒来/休眠状态报表"""
online_groups, _ = await PlatformUtils.get_group_list(bot)
valid_keys = {(str(g.group_id), g.channel_id) for g in online_groups}
all_db_groups = await GroupConsole.all()
target_groups = [
g for g in all_db_groups if (str(g.group_id), g.channel_id) in valid_keys
]
total_count = len(target_groups)
status_data = [
{
"id": str(group.group_id),
"name": group.group_name,
"status": group.status,
"status_text": "工作中" if group.status else "休息中",
"badge_color": "success" if group.status else "info",
}
for group in target_groups
]
wake_list = [item for item in status_data if item["status"]]
sleep_list = [item for item in status_data if not item["status"]]
wake_rate = len(wake_list) / total_count if total_count > 0 else 0
display_list, list_title = status_data, "群组状态详情"
if wake_rate > 0.9:
display_list, list_title = (
sleep_list,
f"休息中的群组 (其余 {len(wake_list)} 个群正常工作中)",
)
elif wake_rate < 0.1:
display_list, list_title = (
wake_list,
f"工作中/已醒来的群组 (其余 {len(sleep_list)} 个群休息中)",
)
return await build_dashboard_report(
page_title="真寻工作状态统计",
total_count=total_count,
active_count=len(wake_list),
inactive_count=len(sleep_list),
active_rate=wake_rate,
active_label="当前工作中",
active_color="var(--color-accent-green)",
inactive_label="当前休息中",
inactive_color="var(--color-text-muted)",
progress_label="全服群组活跃覆盖率",
display_list=display_list,
list_title=list_title,
no_record_alert=ui.alert("无记录", "当前没有已加入的群组记录。", type="info"),
perfect_state_alert=ui.alert(
"状态统一",
(
f"所有 {total_count} 个群组当前均处于 "
f"{'工作中' if wake_rate > 0.5 else '休息中'} 状态。"
),
type="success",
),
)
async def build_dashboard_report(
page_title: str,
total_count: int,
active_count: int,
inactive_count: int,
active_rate: float,
active_label: str,
active_color: str,
inactive_label: str,
inactive_color: str,
progress_label: str,
display_list: list[dict],
list_title: str,
summary_tip: str = "",
global_alert: Any = None,
no_record_alert: Any = None,
perfect_state_alert: Any = None,
) -> bytes:
"""通用的 Dashboard 报表构建器，用于替代原先冗余的 UI 代码"""
kpi_row = LayoutData.row(gap="12px", align_items="stretch")
def _build_kpi_card(title: str, value: str, val_color: str):
header = LayoutData.row(justify_content="space-between", width="100%")
header.add_item(
ui.text(title, font_size="13px", color="var(--color-text-muted)")
)
if title != "总群数" and title != "管理群总数":
rate_str = (
f"{active_rate:.1%}"
if "已开启" in title or "当前工作" in title
else f"{(1 - active_rate):.1%}"
)
header.add_item(
ui.text(rate_str, font_size="13px", bold=True, color=val_color)
)
content = ui.vstack(
[
header.build()
if "已开启" in title or "已关闭" in title
else ui.text(title, font_size="13px", color="var(--color-text-muted)"),
ui.text(value, font_size="24px", bold=True, color=val_color),
],
gap="2px",
align_items="start" if "总" in title else "stretch",
padding="0",
)
return ui.card(content).with_inline_style({"--card-padding": "12px 16px"})
kpi_row.add_item(
_build_kpi_card(
"总群数" if "功能" in progress_label else "管理群总数",
str(total_count),
"var(--color-text-dark)",
),
metadata={"flex": True},
)
kpi_row.add_item(
_build_kpi_card(active_label, str(active_count), active_color),
metadata={"flex": True},
)
kpi_row.add_item(
_build_kpi_card(inactive_label, str(inactive_count), inactive_color),
metadata={"flex": True},
)
progress_scheme = "primary" if "功能" in progress_label else "success"
progress_section = ui.vstack(
[
ui.text(progress_label, font_size="14px", color="var(--color-text-muted)"),
ui.progress_bar(
progress=active_rate * 100,
label=f"{active_count}/{total_count}",
color_scheme=progress_scheme,
),
],
gap="8px",
)
content_area = None
if not display_list:
if total_count == 0 and no_record_alert:
content_area = no_record_alert
elif global_alert and "功能" in progress_label:
content_area = global_alert
elif perfect_state_alert:
content_area = perfect_state_alert
elif len(display_list) <= 15:
rows = []
for item in display_list:
status_cell = StatusBadgeCell(
text=item["status_text"], status_type=item["badge_color"]
)
rows.append(
[
TextCell(content=str(item["id"])),
TextCell(content=str(item["name"])),
status_cell,
]
)
content_area = (
ui.table(list_title, None)
.set_headers(["群号", "群名", "状态"])
.set_column_widths(["160px", "auto", "100px"])
.add_rows(rows)
)
else:
grid = LayoutData.grid(columns=3, gap="15px")
MAX_SHOW = 60
for item in display_list[:MAX_SHOW]:
card_content = ui.vstack(
[
ui.text(str(item["name"]), bold=True, font_size="15px"),
LayoutData.row(justify_content="space-between", width="100%")
.add_item(ui.text(str(item["id"]), font_size="12px", color="#999"))
.add_item(
ui.badge(item["status_text"], color_scheme=item["badge_color"])
),
],
gap="8px",
align_items="start",
)
grid.add_item(ui.card(card_content))
container = LayoutData.column(gap="10px")
container.add_item(grid.build())
if len(display_list) > MAX_SHOW:
container.add_item(
ui.text(
f"... 还有 {len(display_list) - MAX_SHOW} 个群组未显示 ...",
align="center",
color="#ccc",
)
)
content_area = container.build()
main_layout = LayoutData.column(padding="40px", gap="30px")
main_layout.add_item(
ui.text(
page_title,
font_size="32px",
bold=True,
align="center",
color="var(--color-primary)",
)
)
stats_items = []
if global_alert and "功能" in progress_label:
stats_items.append(global_alert)
stats_items.extend(
[
kpi_row.build(),
ui.divider(margin="15px 0"),
progress_section,
]
)
if summary_tip:
stats_items.append(
ui.text(
summary_tip,
font_size="13px",
color="var(--color-text-muted)",
align="center",
)
)
main_layout.add_item(ui.card(ui.vstack(stats_items)))
if content_area:
main_layout.add_item(content_area)
return await ui.render(main_layout.build(), viewport={"width": 900, "height": 10})