import nonebot from nonebot_plugin_uninfo import Uninfo from zhenxun import ui from zhenxun.configs.config import BotConfig, Config from zhenxun.configs.utils import PluginExtraData from zhenxun.models.bot_console import BotConsole from zhenxun.models.group_console import GroupConsole from zhenxun.models.level_user import LevelUser from zhenxun.models.plugin_info import PluginInfo from zhenxun.models.statistics import Statistics from zhenxun.services import avatar_service from zhenxun.services.ai.core.exceptions import LLMException from zhenxun.services.ai.llm.api import chat from zhenxun.services.db_context import with_db_timeout from zhenxun.services.log import logger from zhenxun.services.message_load import is_db_unhealthy from zhenxun.services.renderer.result_cache import RenderResultMemoryCache from zhenxun.ui.models import PluginMenuCategory, PluginMenuData 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 driver = nonebot.get_driver() _DB_BUSY_MESSAGE = "数据库繁忙，请稍后再试" _HELP_DB_TIMEOUT = 3.0 _HELP_MENU_IMAGE_CACHE = RenderResultMemoryCache( ttl_seconds=300, max_items=64, max_total_bytes=64 * 1024 * 1024, ) class _DbBusyError(Exception): pass async def _read_db(factory, operation: str): if is_db_unhealthy(): raise _DbBusyError try: return await with_db_timeout( factory(), timeout=_HELP_DB_TIMEOUT, operation=operation, source="help", ) except TimeoutError as exc: raise _DbBusyError from exc def _create_plugin_menu_item( bot: BotConsole | None, plugin: PluginInfo, group: GroupConsole | None, is_detail: bool, ) -> dict: """为插件菜单构造一个插件菜单项数据字典""" 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: extra_data = PluginExtraData(**nb_plugin.metadata.extra) 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_type = 3 elif group and plugin.block_type == BlockType.GROUP: status_type = 3 elif not group and plugin.block_type == BlockType.PRIVATE: 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: extra_data = PluginExtraData(**nb_plugin.metadata.extra) commands = [cmd.command for cmd in extra_data.commands] return { "id": str(plugin.id), "name": plugin.name, "status": status_type, "has_superuser_help": has_superuser_help, "commands": commands, } async def create_help_img( session: Uninfo, group_id: str | None, is_detail: bool ) -> str | bytes: """使用渲染服务生成帮助图片""" try: classified_data = await _read_db( lambda: classify_plugin( session, group_id, is_detail, _create_plugin_menu_item ), "Help.classify_plugin", ) except _DbBusyError: return _DB_BUSY_MESSAGE sorted_categories = dict( sorted(classified_data.items(), key=lambda x: len(x[1]), reverse=True) ) categories_for_model = [] plugin_count = 0 active_count = 0 if sorted_categories: menu_key = next(iter(sorted_categories.keys())) max_data = sorted_categories.pop(menu_key) 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"] == 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"] == 0) platform = PlatformUtils.get_platform(session) bot_id = BotConfig.get_qbot_uid(session.self_id) or session.self_id bot_avatar_path = await avatar_service.get_avatar_path(platform, bot_id) bot_avatar_url = bot_avatar_path.as_uri() if bot_avatar_path else "" categories_objects = [] for category in categories_for_model: categories_objects.append( PluginMenuCategory(name=category["name"], items=category["items"]) ) menu_data = PluginMenuData( bot_name=BotConfig.self_nickname, bot_avatar_url=bot_avatar_url, is_detail=is_detail, plugin_count=plugin_count, active_count=active_count, categories=categories_objects, ) cache_payload = { "self_id": session.self_id, "group_id": group_id, "is_detail": is_detail, "theme": Config.get_config("UI", "THEME", "default"), "menu_data": menu_data, } cache_key = RenderResultMemoryCache.build_key(cache_payload) if cached_image := await _HELP_MENU_IMAGE_CACHE.get(cache_key): return cached_image image_bytes = await ui.render( menu_data, clip_selector=".wrapper", clip_padding=20, disable_animations=True, ) await _HELP_MENU_IMAGE_CACHE.set(cache_key, image_bytes) return image_bytes async def get_user_allow_help(user_id: str) -> list[str]: """获取用户可访问插件类型列表 参数: user_id: 用户id 返回: list[str]: 插件类型列表 """ type_list = ["NORMAL", "DEPENDANT"] levels = await _read_db( lambda: LevelUser.filter(user_id=user_id).values_list("user_level", flat=True), "Help.user_allow_level", ) for level in levels: if level > 0: # type: ignore type_list.extend(("ADMIN", "ADMIN_SUPER")) break if user_id in driver.config.superusers: type_list.append("SUPERUSER") return type_list async def get_plugin_help( user_id: str, name: str, is_superuser: bool, variant: str | None = None ) -> str | bytes | None: """获取功能的帮助信息 参数: user_id: 用户id name: 插件名称或id is_superuser: 是否为超级用户 variant: 使用的皮肤/变体名称 """ try: type_list = await get_user_allow_help(user_id) if name.isdigit(): plugin = await _read_db( lambda: PluginInfo.get_or_none(id=int(name), plugin_type__in=type_list), "Help.plugin_by_id", ) else: plugin = await _read_db( lambda: PluginInfo.get_or_none( name__iexact=name, load_status=True, plugin_type__in=type_list ), "Help.plugin_by_name", ) except _DbBusyError: return _DB_BUSY_MESSAGE if plugin: _plugin = nonebot.get_plugin_by_module_name(plugin.module_path) if _plugin and _plugin.metadata: extra_data = PluginExtraData(**_plugin.metadata.extra) try: call_count = await _read_db( lambda: Statistics.filter(plugin_name=plugin.module).count(), "Help.plugin_call_count", ) except _DbBusyError: return _DB_BUSY_MESSAGE usage = _plugin.metadata.usage metadata_items = [ {"label": "作者", "value": extra_data.author or "未知"}, {"label": "版本", "value": extra_data.version or "未知"}, {"label": "调用次数", "value": call_count}, ] sections = [] sections.append( { "title": "功能简介", "content": [ format_usage_for_markdown(_plugin.metadata.description.strip()) ], "is_admin": False, } ) 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, "metadata": metadata_items, "sections": sections, } component = ui.template("pages/builtin/help", data=page_data) if variant: component.variant = variant return await ui.render(component, use_cache=True, device_scale_factor=2) return None return None async def get_llm_help(question: str, user_id: str) -> str | bytes: """ 使用LLM来回答用户的自然语言求助。 参数: question: 用户的问题。 user_id: 提问用户的ID。 返回: str | bytes: LLM生成的回答或错误提示。 """ try: try: allowed_types = await get_user_allow_help(user_id) plugins = await _read_db( lambda: PluginInfo.get_plugins( load_status=None, filter_parent=False, is_show=True, plugin_type__in=allowed_types, ), "Help.llm_plugin_list", ) except _DbBusyError: return _DB_BUSY_MESSAGE knowledge_base_parts = [] for p in plugins: meta = nonebot.get_plugin_by_module_name(p.module_path) if not meta or not meta.metadata: continue usage = meta.metadata.usage.strip() or "无" desc = meta.metadata.description.strip() or "无" part = f"功能名称: {p.name}\n功能描述: {desc}\n用法示例:\n{usage}" knowledge_base_parts.append(part) if not knowledge_base_parts: return "抱歉，根据您的权限，当前没有可供查询的功能信息。" knowledge_base = "\n\n---\n\n".join(knowledge_base_parts) user_role = "普通用户" if PluginType.SUPERUSER in allowed_types: user_role = "超级管理员" elif PluginType.ADMIN in allowed_types: user_role = "管理员" base_system_prompt = ( f"你是一个精通机器人功能的AI助手。当前向你提问的用户是一位「{user_role}」。\n" "你的任务是根据下面提供的功能列表和详细说明，来回答用户关于如何使用机器人的问题。\n" "请仔细阅读每个功能的描述和用法，然后用简洁、清晰的语言告诉用户应该使用哪个或哪些命令来解决他们的问题。\n" "如果找不到完全匹配的功能，可以推荐最相关的一个或几个。直接给出操作指令和简要解释即可。" ) if ( Config.get_config("help", "LLM_HELPER_STYLE") and Config.get_config("help", "LLM_HELPER_STYLE").strip() ): style = Config.get_config("help", "LLM_HELPER_STYLE") style_instruction = f"请务必使用「{style}」的风格和口吻来回答。" system_prompt = f"{base_system_prompt}\n{style_instruction}" else: system_prompt = base_system_prompt full_instruction = ( f"{system_prompt}\n\n=== 功能列表和说明 ===\n{knowledge_base}" ) response = await chat( message=question, instruction=full_instruction, model=Config.get_config("help", "DEFAULT_LLM_MODEL"), ) reply_text = response.text if response else "抱歉，我暂时无法回答这个问题。" threshold = Config.get_config("help", "LLM_HELPER_REPLY_AS_IMAGE_THRESHOLD", 50) if len(reply_text) > threshold: notebook = ui.notebook() notebook.text(reply_text) return await ui.render(notebook) return reply_text except LLMException as e: logger.error(f"LLM智能帮助出错: {e}", "帮助", e=e) return "抱歉，智能帮助功能当前不可用，请稍后再试或联系管理员。" except Exception as e: logger.error(f"构建LLM帮助时发生未知错误: {e}", "帮助", e=e) return "抱歉，智能帮助功能遇到了一点小问题，正在紧急处理中！"