@@ -7,7 +7,6 @@ from typing import Any, Optional, Tuple
from loguru import logger as default_logger
from base . plugin_common . plugin_manager import PluginManager
from utils . markdown_to_image import convert_md_str_to_image , html_to_image
from utils . revoke . message_auto_revoke import MessageAutoRevoke
from utils . robot_cmd . robot_command import Feature , GroupBotManager , PermissionStatus
@@ -190,276 +189,6 @@ class RobotMenuRenderTool:
} ,
)
@staticmethod
def _get_plugin_manager ( ) - > PluginManager :
""" 获取当前运行中的插件管理器单例。 """
return PluginManager . get_instance ( )
@staticmethod
def _resolve_snapshot_group_status ( snapshot : dict , group_id : str ) - > dict :
""" 解析插件在当前群里的可用状态。
规则说明:
1. 插件必须先处于 RUNNING, 才可能被认为“可用”;
2. 若插件支持群级开关,则继续读取该群的 feature 权限;
3. 若插件没有群级开关,则视为“运行即全局可用”。
"""
normalized_snapshot = dict ( snapshot or { } )
status = str ( normalized_snapshot . get ( " status " ) or " " ) . strip ( ) . upper ( )
supports_group_switch = bool ( normalized_snapshot . get ( " supports_group_switch " ) )
feature_key = str ( normalized_snapshot . get ( " feature_key " ) or " " ) . strip ( )
if status != " RUNNING " :
return {
" available " : False ,
" reason " : " 插件未运行 " ,
" reason_code " : " plugin_not_running " ,
}
if not group_id or not supports_group_switch or not feature_key :
return {
" available " : True ,
" reason " : " 全局可用 " ,
" reason_code " : " global_available " ,
}
feature = Feature . get_feature ( feature_key )
if feature is None :
return {
" available " : True ,
" reason " : " 未绑定群级开关,按运行中处理 " ,
" reason_code " : " feature_not_registered " ,
}
permission = GroupBotManager . get_group_permission ( group_id , feature )
if permission == PermissionStatus . ENABLED :
return {
" available " : True ,
" reason " : " 本群已启用 " ,
" reason_code " : " group_enabled " ,
}
return {
" available " : False ,
" reason " : " 本群未启用 " ,
" reason_code " : " group_disabled " ,
}
@staticmethod
def _format_plugin_command ( example_command : str , command_prefix : str ) - > str :
""" 把插件命令和前缀拼成最终展示文本。 """
prefix = str ( command_prefix or " " ) . strip ( )
command = str ( example_command or " " ) . strip ( )
if not prefix :
return command
return f " { prefix } { command } "
def _build_plugin_command_entry ( self , snapshot : dict , group_id : str ) - > Optional [ dict ] :
""" 把插件快照转换为菜单可展示的命令项。 """
normalized_snapshot = dict ( snapshot or { } )
commands = list ( normalized_snapshot . get ( " commands " , [ ] ) or [ ] )
plugin_types = list ( normalized_snapshot . get ( " plugin_types " , [ ] ) or [ ] )
if not commands and " scheduled " not in plugin_types :
return None
availability = self . _resolve_snapshot_group_status ( normalized_snapshot , group_id )
command_prefix = str ( normalized_snapshot . get ( " command_prefix " ) or " " ) . strip ( )
primary_command = self . _format_plugin_command ( commands [ 0 ] , command_prefix ) if commands else " "
alias_commands = [
self . _format_plugin_command ( command_text , command_prefix )
for command_text in commands [ 1 : 4 ]
if str ( command_text or " " ) . strip ( )
]
if " message " in plugin_types :
category = " message "
category_label = " 消息指令 "
elif " scheduled " in plugin_types :
category = " scheduled "
category_label = " 自动任务 "
else :
category = " generic "
category_label = " 通用能力 "
return {
" name " : str ( normalized_snapshot . get ( " name " ) or " " ) . strip ( ) ,
" module_name " : str ( normalized_snapshot . get ( " module_name " ) or " " ) . strip ( ) ,
" description " : str ( normalized_snapshot . get ( " description " ) or " " ) . strip ( ) or " 暂无描述 " ,
" category " : category ,
" category_label " : category_label ,
" commands " : commands ,
" primary_command " : primary_command ,
" alias_commands " : alias_commands ,
" supports_group_switch " : bool ( normalized_snapshot . get ( " supports_group_switch " ) ) ,
" feature_key " : str ( normalized_snapshot . get ( " feature_key " ) or " " ) . strip ( ) ,
" available " : bool ( availability . get ( " available " ) ) ,
" availability_reason " : str ( availability . get ( " reason " ) or " " ) . strip ( ) ,
" availability_code " : str ( availability . get ( " reason_code " ) or " " ) . strip ( ) ,
" status_label " : str ( normalized_snapshot . get ( " status_label " ) or " " ) . strip ( ) ,
}
def _collect_command_catalog ( self , group_id : str , requester_id : str ) - > dict :
""" 采集当前群和当前身份视角下的命令清单。
输出结构分三层:
1. 普通用户可直接用的命令;
2. 自动/定时能力;
3. 管理员附加能力与未启用项。
"""
plugin_manager = self . _get_plugin_manager ( )
snapshots = plugin_manager . get_plugin_snapshots ( )
is_admin = bool ( GroupBotManager . is_admin_for_group ( requester_id , group_id ) ) if group_id else bool ( GroupBotManager . is_admin ( requester_id ) )
available_manual = [ ]
available_auto = [ ]
unavailable_manual = [ ]
for snapshot in snapshots :
entry = self . _build_plugin_command_entry ( snapshot , group_id )
if not entry :
continue
if entry [ " category " ] == " scheduled " :
if entry [ " available " ] :
available_auto . append ( entry )
continue
if entry [ " available " ] :
available_manual . append ( entry )
else :
unavailable_manual . append ( entry )
available_manual . sort ( key = lambda item : ( item [ " category " ] , item [ " name " ] , item [ " primary_command " ] ) )
available_auto . sort ( key = lambda item : ( item [ " name " ] , item [ " primary_command " ] ) )
unavailable_manual . sort ( key = lambda item : ( item [ " availability_code " ] , item [ " name " ] ) )
admin_commands = [ ]
if is_admin :
admin_commands = [
{ " title " : " 查看功能状态 " , " example " : " 菜单 状态 " , " description " : " 查看当前群所有功能开关状态 " } ,
{ " title " : " 启用某个功能 " , " example " : " 菜单 启用 功能序号 " , " description " : " 按菜单序号启用某项功能 " } ,
{ " title " : " 关闭某个功能 " , " example " : " 菜单 关闭 功能序号 " , " description " : " 按菜单序号关闭某项功能 " } ,
{ " title " : " 查看群管理员 " , " example " : " 菜单 管理员 列表 " , " description " : " 查看当前群管理员清单 " } ,
{ " title " : " 添加群管理员 " , " example " : " 菜单 管理员 添加 @某人 " , " description " : " 把某个群成员加入本群管理员 " } ,
{ " title " : " 删除群管理员 " , " example " : " 菜单 管理员 删除 @某人 " , " description " : " 移除某个群管理员 " } ,
]
if GroupBotManager . is_admin ( requester_id ) :
admin_commands . append (
{ " title " : " 查看托管群列表 " , " example " : " 菜单 群列表 " , " description " : " 查看所有已启用机器人的群 " }
)
return {
" group_id " : str ( group_id or " " ) . strip ( ) ,
" requester_id " : str ( requester_id or " " ) . strip ( ) ,
" is_admin " : is_admin ,
" available_manual " : available_manual ,
" available_auto " : available_auto ,
" unavailable_manual " : unavailable_manual ,
" admin_commands " : admin_commands ,
" generated_at " : time . strftime ( " % Y- % m- %d % H: % M: % S " , time . localtime ( ) ) ,
}
def build_command_catalog_text ( self , group_id : str , requester_id : str ) - > str :
""" 构建适合直接发送给用户的文本版命令清单。 """
catalog = self . _collect_command_catalog ( group_id , requester_id )
lines = [
" 📚 当前群指令清单 " ,
f " 群ID: { catalog [ ' group_id ' ] or ' 私聊 ' } " ,
f " 生成时间: { catalog [ ' generated_at ' ] } " ,
" " ,
" 一、当前可直接使用的命令 " ,
]
if catalog [ " available_manual " ] :
for item in catalog [ " available_manual " ] :
lines . append ( f " 【 { item [ ' name ' ] } 】 { item [ ' description ' ] } " )
if item [ " primary_command " ] :
lines . append ( f " 主指令: { item [ ' primary_command ' ] } " )
if item [ " alias_commands " ] :
lines . append ( f " 别名: { ' / ' . join ( item [ ' alias_commands ' ] ) } " )
lines . append ( " " )
else :
lines . append ( " 当前没有可直接使用的手动命令 " )
lines . append ( " " )
lines . append ( " 二、自动/定时能力 " )
if catalog [ " available_auto " ] :
for item in catalog [ " available_auto " ] :
lines . append ( f " 【 { item [ ' name ' ] } 】 { item [ ' description ' ] } " )
lines . append ( " 触发方式:自动或定时运行 " )
lines . append ( " " )
else :
lines . append ( " 当前没有已启用的自动能力 " )
lines . append ( " " )
if catalog [ " is_admin " ] :
lines . append ( " 三、管理员额外可见 " )
if catalog [ " unavailable_manual " ] :
lines . append ( " 未启用或暂不可用命令: " )
for item in catalog [ " unavailable_manual " ] :
primary = item [ " primary_command " ] or " 无手动指令 "
lines . append ( f " - { item [ ' name ' ] } : { primary } ( { item [ ' availability_reason ' ] } ) " )
lines . append ( " " )
else :
lines . append ( " 当前没有未启用的命令项 " )
lines . append ( " " )
lines . append ( " 管理命令: " )
for item in catalog [ " admin_commands " ] :
lines . append ( f " - { item [ ' example ' ] } : { item [ ' description ' ] } " )
lines . append ( " " )
lines . append ( " 提示:发送“菜单”查看功能开关;发送“菜单 状态”查看本群功能状态。 " )
return " \n " . join ( lines ) . strip ( )
def build_command_catalog_markdown ( self , group_id : str , requester_id : str ) - > str :
""" 构建适合图片渲染的 Markdown 版指令清单。 """
catalog = self . _collect_command_catalog ( group_id , requester_id )
lines = [
" # 机器人指令清单 " ,
" " ,
f " - 目标:` { catalog [ ' group_id ' ] or ' 私聊 ' } ` " ,
f " - 生成时间:` { catalog [ ' generated_at ' ] } ` " ,
" " ,
" ## 当前可直接使用的命令 " ,
]
if catalog [ " available_manual " ] :
for item in catalog [ " available_manual " ] :
lines . append ( f " ### { item [ ' name ' ] } " )
lines . append ( f " - 说明: { item [ ' description ' ] } " )
if item [ " primary_command " ] :
lines . append ( f " - 主指令:` { item [ ' primary_command ' ] } ` " )
if item [ " alias_commands " ] :
alias_text = " / " . join ( f " ` { alias } ` " for alias in item [ " alias_commands " ] )
lines . append ( f " - 别名: { alias_text } " )
lines . append ( " " )
else :
lines . append ( " - 当前没有可直接使用的手动命令 " )
lines . append ( " " )
lines . append ( " ## 自动/定时能力 " )
if catalog [ " available_auto " ] :
for item in catalog [ " available_auto " ] :
lines . append ( f " - ** { item [ ' name ' ] } **: { item [ ' description ' ] } " )
else :
lines . append ( " - 当前没有已启用的自动能力 " )
lines . append ( " " )
if catalog [ " is_admin " ] :
lines . append ( " ## 管理员额外可见 " )
if catalog [ " unavailable_manual " ] :
lines . append ( " ### 未启用或暂不可用命令 " )
for item in catalog [ " unavailable_manual " ] :
primary = item [ " primary_command " ] or " 无手动指令 "
lines . append ( f " - ** { item [ ' name ' ] } **: ` { primary } `( { item [ ' availability_reason ' ] } ) " )
lines . append ( " " )
lines . append ( " ### 管理命令 " )
for item in catalog [ " admin_commands " ] :
lines . append ( f " - ` { item [ ' example ' ] } `: { item [ ' description ' ] } " )
lines . append ( " " )
lines . append ( " > 提示:发送 `菜单` 查看功能开关;发送 `菜单 状态` 查看本群功能状态。 " )
return " \n " . join ( lines )
async def send_menu_content (
self ,
bot : WechatAPIClient ,