From 0f91ce03e4c6d0197d55b96b42a6954912b15d81 Mon Sep 17 00:00:00 2001 From: Rumio <32546670+webjoin111@users.noreply.github.com> Date: Thu, 19 Mar 2026 16:08:13 +0800 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor(plugin=5Fswitch):?= =?UTF-8?q?=20=E9=87=8D=E6=9E=84=E5=8A=9F=E8=83=BD=E5=BC=80=E5=85=B3?= =?UTF-8?q?=E4=B8=8E=E5=B8=AE=E5=8A=A9=E7=B3=BB=E7=BB=9F=E5=B9=B6=E4=BC=98?= =?UTF-8?q?=E5=8C=96=20UI=20=E6=8A=A5=E8=A1=A8=E6=B8=B2=E6=9F=93=20(#2111)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * ♻️ refactor(plugin_switch): 重构功能开关与帮助系统并优化 UI 报表渲染 - 引入策略模式 (Strategy Pattern) 统一插件与被动任务的开关逻辑 - 新增超级用户强制管控功能，支持系统级禁用且群管无法自行开启 - 新增全服群组活跃状态报表与功能全局覆盖率可视化统计图表 - 优化指令解析系统，支持多功能批量开关及更智能的快捷词映射 - 重构帮助系统，支持在帮助页区分展示管理员与超级用户指令 - 细化插件列表状态显示，区分群控、系统管控与全局禁用状态 - UI 组件 UserInfoBlock 新增 extra 扩展插槽支持 - 更新 resources.spec 资源版本要求至 1.1.1 * :rotating_light: 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> --- resources.spec | 2 +- .../admin/plugin_switch/__init__.py | 662 ++++++++++-------- .../admin/plugin_switch/_data_source.py | 600 ---------------- .../admin/plugin_switch/command.py | 187 ++--- .../admin/plugin_switch/data_source.py | 325 +++++++++ .../admin/plugin_switch/strategy.py | 187 +++++ .../builtin_plugins/admin/plugin_switch/ui.py | 454 ++++++++++++ zhenxun/builtin_plugins/help/__init__.py | 19 +- zhenxun/builtin_plugins/help/_config.py | 17 - zhenxun/builtin_plugins/help/config.py | 3 + .../help/{_data_source.py => data_source.py} | 102 +-- .../help/{_utils.py => utils.py} | 0 zhenxun/ui/__init__.py | 2 + zhenxun/ui/models/components/display.py | 4 + zhenxun/ui/models/presets/plugin_menu.py | 4 +- 15 files changed, 1464 insertions(+), 1104 deletions(-) delete mode 100644 zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py create mode 100644 zhenxun/builtin_plugins/admin/plugin_switch/data_source.py create mode 100644 zhenxun/builtin_plugins/admin/plugin_switch/strategy.py create mode 100644 zhenxun/builtin_plugins/admin/plugin_switch/ui.py delete mode 100644 zhenxun/builtin_plugins/help/_config.py create mode 100644 zhenxun/builtin_plugins/help/config.py rename zhenxun/builtin_plugins/help/{_data_source.py => data_source.py} (85%) rename zhenxun/builtin_plugins/help/{_utils.py => utils.py} (100%) diff --git a/resources.spec b/resources.spec index 43998270..a45aa5f1 100644 --- a/resources.spec +++ b/resources.spec @@ -1 +1 @@ -require_resources_version: ">=1.1.0" +require_resources_version: ">=1.1.1" diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/__init__.py b/zhenxun/builtin_plugins/admin/plugin_switch/__init__.py index 9a638039..2fe9f2aa 100644 --- a/zhenxun/builtin_plugins/admin/plugin_switch/__init__.py +++ b/zhenxun/builtin_plugins/admin/plugin_switch/__init__.py @@ -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 + ) diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py b/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py deleted file mode 100644 index 89f11792..00000000 --- a/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py +++ /dev/null @@ -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 "没有找到这个功能喔..." diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/command.py b/zhenxun/builtin_plugins/admin/plugin_switch/command.py index 62056a2e..3460d098 100644 --- a/zhenxun/builtin_plugins/admin/plugin_switch/command.py +++ b/zhenxun/builtin_plugins/admin/plugin_switch/command.py @@ -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.+)", - command="switch", - arguments=["open", "{name}", "--task"], - prefix=True, -) - -_status_matcher.shortcut( - r"关闭群被动\s*(?P.+)", - command="switch", - arguments=["close", "{name}", "--task"], - prefix=True, -) - -_status_matcher.shortcut( - r"开启默认群被动\s*(?P.+)", - command="switch", - arguments=["open", "{name}", "--task", "-df"], - prefix=True, -) - -_status_matcher.shortcut( - r"关闭默认群被动\s*(?P.+)", - 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.+)", - command="switch", - arguments=["open", "{name}", "-df"], - prefix=True, -) - -_status_matcher.shortcut( - r"开启(?P.+)", - 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.+)", - command="switch", - arguments=["close", "{name}", "-df"], - prefix=True, -) - -_status_matcher.shortcut( - r"关闭(?P.+)", - command="switch", - arguments=["close", "{name}"], + r"^(?P开启|关闭)\s*(?P所有|全部)?\s*(?P默认)?\s*(?P群被动|被动|插件|功能)?\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, +) diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/data_source.py b/zhenxun/builtin_plugins/admin/plugin_switch/data_source.py new file mode 100644 index 00000000..7118e74d --- /dev/null +++ b/zhenxun/builtin_plugins/admin/plugin_switch/data_source.py @@ -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} 操作。" diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/strategy.py b/zhenxun/builtin_plugins/admin/plugin_switch/strategy.py new file mode 100644 index 00000000..62e3e7d7 --- /dev/null +++ b/zhenxun/builtin_plugins/admin/plugin_switch/strategy.py @@ -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() diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/ui.py b/zhenxun/builtin_plugins/admin/plugin_switch/ui.py new file mode 100644 index 00000000..4abff5c7 --- /dev/null +++ b/zhenxun/builtin_plugins/admin/plugin_switch/ui.py @@ -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}) diff --git a/zhenxun/builtin_plugins/help/__init__.py b/zhenxun/builtin_plugins/help/__init__.py index 37dbc0ba..b8857e94 100644 --- a/zhenxun/builtin_plugins/help/__init__.py +++ b/zhenxun/builtin_plugins/help/__init__.py @@ -19,7 +19,7 @@ from zhenxun.services.log import logger from zhenxun.utils.enum import PluginType from zhenxun.utils.message import MessageUtils -from ._data_source import create_help_img, get_llm_help, get_plugin_help +from .data_source import create_help_img, get_llm_help, get_plugin_help __plugin_meta__ = PluginMetadata( name="帮助", @@ -75,7 +75,6 @@ _matcher = on_alconna( Alconna( "功能", Args["name?", str], - Option("-s|--superuser", action=store_true, help_text="超级用户帮助"), Option("-d|--detail", action=store_true, help_text="详细帮助"), ), aliases={"help", "帮助", "菜单"}, @@ -98,15 +97,9 @@ async def _( bot: Bot, name: Match[str], session: Uninfo, - is_superuser: Query[bool] = AlconnaQuery("superuser.value", False), is_detail: Query[bool] = AlconnaQuery("detail.value", False), ): - _is_superuser = is_superuser.result if is_superuser.available else False - - if _is_superuser and session.user.id not in bot.config.superusers: - await MessageUtils.build_message("权限不足，无法查看超级用户帮助").finish( - reply_to=True - ) + _is_superuser = session.user.id in bot.config.superusers if name.available: help_style = Config.get_config("help", "HELP_STYLE") @@ -116,11 +109,7 @@ async def _( session.user.id, name.result, _is_superuser, variant=variant ) - is_plugin_found = not ( - isinstance(traditional_help_result, str) - and "没有查找到这个功能噢..." in traditional_help_result - ) - if is_plugin_found: + if traditional_help_result is not None: await MessageUtils.build_message(traditional_help_result).send( reply_to=True ) @@ -130,7 +119,7 @@ async def _( llm_answer = await get_llm_help(name.result, session.user.id) await MessageUtils.build_message(llm_answer).send(reply_to=True) else: - await MessageUtils.build_message(traditional_help_result).send( + await MessageUtils.build_message("没有查找到这个功能噢...").send( reply_to=True ) logger.info( diff --git a/zhenxun/builtin_plugins/help/_config.py b/zhenxun/builtin_plugins/help/_config.py deleted file mode 100644 index 586026c5..00000000 --- a/zhenxun/builtin_plugins/help/_config.py +++ /dev/null @@ -1,17 +0,0 @@ -from zhenxun.configs.config import Config -from zhenxun.configs.path_config import DATA_PATH, IMAGE_PATH - -GROUP_HELP_PATH = DATA_PATH / "group_help" -GROUP_HELP_PATH.mkdir(exist_ok=True, parents=True) -for f in GROUP_HELP_PATH.iterdir(): - f.unlink() - -SIMPLE_HELP_IMAGE = IMAGE_PATH / "SIMPLE_HELP.png" -if SIMPLE_HELP_IMAGE.exists(): - SIMPLE_HELP_IMAGE.unlink() - -SIMPLE_DETAIL_HELP_IMAGE = IMAGE_PATH / "SIMPLE_DETAIL_HELP.png" -if SIMPLE_DETAIL_HELP_IMAGE.exists(): - SIMPLE_DETAIL_HELP_IMAGE.unlink() - -base_config = Config.get("help") diff --git a/zhenxun/builtin_plugins/help/config.py b/zhenxun/builtin_plugins/help/config.py new file mode 100644 index 00000000..cb8f16d2 --- /dev/null +++ b/zhenxun/builtin_plugins/help/config.py @@ -0,0 +1,3 @@ +from zhenxun.configs.config import Config + +base_config = Config.get("help") diff --git a/zhenxun/builtin_plugins/help/_data_source.py b/zhenxun/builtin_plugins/help/data_source.py similarity index 85% rename from zhenxun/builtin_plugins/help/_data_source.py rename to zhenxun/builtin_plugins/help/data_source.py index 44315e20..8bc6b107 100644 --- a/zhenxun/builtin_plugins/help/_data_source.py +++ b/zhenxun/builtin_plugins/help/data_source.py @@ -3,7 +3,6 @@ from nonebot_plugin_uninfo import Uninfo from zhenxun import ui from zhenxun.configs.config import BotConfig, Config -from zhenxun.configs.path_config import IMAGE_PATH from zhenxun.configs.utils import PluginExtraData from zhenxun.models.bot_console import BotConsole from zhenxun.models.group_console import GroupConsole @@ -23,10 +22,7 @@ from zhenxun.utils.common_utils import format_usage_for_markdown from zhenxun.utils.enum import BlockType, PluginType from zhenxun.utils.platform import PlatformUtils -from ._utils import classify_plugin - -random_bk_path = IMAGE_PATH / "background" / "help" / "simple_help" -background = IMAGE_PATH / "background" / "0.png" +from .utils import classify_plugin driver = nonebot.get_driver() _HELP_MENU_IMAGE_CACHE = RenderResultMemoryCache( @@ -43,7 +39,7 @@ def _create_plugin_menu_item( is_detail: bool, ) -> dict: """为插件菜单构造一个插件菜单项数据字典""" - status = True + status_type = 0 has_superuser_help = False nb_plugin = nonebot.get_plugin_by_module_name(plugin.module_path) if nb_plugin and nb_plugin.metadata and nb_plugin.metadata.extra: @@ -51,17 +47,21 @@ def _create_plugin_menu_item( if extra_data.superuser_help: has_superuser_help = True + module_tag = f"<{plugin.module}," + if not plugin.status: if plugin.block_type == BlockType.ALL: - status = False + status_type = 3 elif group and plugin.block_type == BlockType.GROUP: - status = False + status_type = 3 elif not group and plugin.block_type == BlockType.PRIVATE: - status = False - elif group and f"{plugin.module}," in group.block_plugin: - status = False - elif bot and f"{plugin.module}," in bot.block_plugins: - status = False + status_type = 3 + elif group and module_tag in (group.superuser_block_plugin or ""): + status_type = 2 + elif bot and module_tag in (bot.block_plugins or ""): + status_type = 2 + elif group and module_tag in (group.block_plugin or ""): + status_type = 1 commands = [] if is_detail and nb_plugin and nb_plugin.metadata and nb_plugin.metadata.extra: @@ -71,7 +71,7 @@ def _create_plugin_menu_item( return { "id": str(plugin.id), "name": plugin.name, - "status": status, + "status": status_type, "has_superuser_help": has_superuser_help, "commands": commands, } @@ -98,13 +98,13 @@ async def create_help_img( main_category_name = "主要功能" if menu_key in ["normal", "功能"] else menu_key categories_for_model.append({"name": main_category_name, "items": max_data}) plugin_count += len(max_data) - active_count += sum(1 for item in max_data if item["status"]) + active_count += sum(1 for item in max_data if item["status"] == 0) for menu, value in sorted_categories.items(): category_name = "主要功能" if menu in ["normal", "功能"] else menu categories_for_model.append({"name": category_name, "items": value}) plugin_count += len(value) - active_count += sum(1 for item in value if item["status"]) + active_count += sum(1 for item in value if item["status"] == 0) platform = PlatformUtils.get_platform(session) bot_id = BotConfig.get_qbot_uid(session.self_id) or session.self_id @@ -117,7 +117,6 @@ async def create_help_img( PluginMenuCategory(name=category["name"], items=category["items"]) ) - # 直接实例化 Data Model menu_data = PluginMenuData( bot_name=BotConfig.self_nickname, bot_avatar_url=bot_avatar_url, @@ -169,29 +168,9 @@ async def get_user_allow_help(user_id: str) -> list[PluginType]: return type_list -def min_leading_spaces(str_list: list[str]) -> int: - min_spaces = 9999 - - for s in str_list: - leading_spaces = len(s) - len(s.lstrip(" ")) - - if leading_spaces < min_spaces: - min_spaces = leading_spaces - - return min_spaces if min_spaces != 9999 else 0 - - -def split_text(text: str): - split_text = text.split("\n") - min_spaces = min_leading_spaces(split_text) - if min_spaces > 0: - split_text = [s[min_spaces:] for s in split_text] - return [s.replace(" ", " ") for s in split_text] - - async def get_plugin_help( user_id: str, name: str, is_superuser: bool, variant: str | None = None -) -> str | bytes: +) -> bytes | None: """获取功能的帮助信息 参数: @@ -215,10 +194,6 @@ async def get_plugin_help( call_count = await Statistics.filter(plugin_name=plugin.module).count() usage = _plugin.metadata.usage - if is_superuser: - if not extra_data.superuser_help: - return "该功能没有超级用户帮助信息" - usage = extra_data.superuser_help metadata_items = [ {"label": "作者", "value": extra_data.author or "未知"}, @@ -226,15 +201,40 @@ async def get_plugin_help( {"label": "调用次数", "value": call_count}, ] - processed_description = format_usage_for_markdown( - _plugin.metadata.description.strip() + sections = [] + sections.append( + { + "title": "功能简介", + "content": [ + format_usage_for_markdown(_plugin.metadata.description.strip()) + ], + "is_admin": False, + } ) - processed_usage = format_usage_for_markdown(usage.strip()) - sections = [ - {"title": "简介", "content": [processed_description]}, - {"title": "使用方法", "content": [processed_usage]}, - ] + if usage and usage.strip(): + sections.append( + { + "title": "管理员指令", + "content": [format_usage_for_markdown(usage.strip())], + "is_admin": False, + } + ) + + if ( + is_superuser + and extra_data.superuser_help + and extra_data.superuser_help.strip() + ): + sections.append( + { + "title": "超级用户指令", + "content": [ + format_usage_for_markdown(extra_data.superuser_help.strip()) + ], + "is_admin": True, + } + ) page_data = { "title": _plugin.metadata.name, @@ -246,8 +246,8 @@ async def get_plugin_help( if variant: component.variant = variant return await ui.render(component, use_cache=True, device_scale_factor=2) - return "糟糕! 该功能没有帮助喔..." - return "没有查找到这个功能噢..." + return None + return None async def get_llm_help(question: str, user_id: str) -> str | bytes: diff --git a/zhenxun/builtin_plugins/help/_utils.py b/zhenxun/builtin_plugins/help/utils.py similarity index 100% rename from zhenxun/builtin_plugins/help/_utils.py rename to zhenxun/builtin_plugins/help/utils.py diff --git a/zhenxun/ui/__init__.py b/zhenxun/ui/__init__.py index c5b39f7d..c1989578 100644 --- a/zhenxun/ui/__init__.py +++ b/zhenxun/ui/__init__.py @@ -241,12 +241,14 @@ def user_info_block( avatar_url: str, subtitle: str | None = None, tags: list[str] | None = None, + extra: RenderableComponent | None = None, ) -> UserInfoBlock: return UserInfoBlock( name=name, avatar_url=avatar_url, subtitle=subtitle, tags=tags or [], + extra=extra, ) diff --git a/zhenxun/ui/models/components/display.py b/zhenxun/ui/models/components/display.py index 8ef44de2..4278a96c 100644 --- a/zhenxun/ui/models/components/display.py +++ b/zhenxun/ui/models/components/display.py @@ -96,6 +96,10 @@ class UserInfoBlock(RenderableComponent): """显示在名称下方的副标题 (如UID或角色)""" tags: list[str] = Field(default_factory=list, description="附加的标签列表") """附加的标签列表""" + extra: RenderableComponent | None = Field( + default=None, description="附加扩展插槽，支持传入任意UI组件(如徽章、按钮)" + ) + """附加扩展插槽""" @property def template_name(self) -> str: diff --git a/zhenxun/ui/models/presets/plugin_menu.py b/zhenxun/ui/models/presets/plugin_menu.py index 8fe9df56..d3a47aa6 100644 --- a/zhenxun/ui/models/presets/plugin_menu.py +++ b/zhenxun/ui/models/presets/plugin_menu.py @@ -16,8 +16,8 @@ class PluginMenuItem(BaseModel): """插件的唯一ID""" name: str """插件名称""" - status: bool - """插件在当前群组的开关状态""" + status: int + """插件状态码: 0:开启, 1:群关, 2:系统关, 3:全局关""" has_superuser_help: bool """插件是否有超级用户专属帮助""" commands: list[str] = Field(default_factory=list, description="插件的主要命令列表")