import json from loguru import logger import os import toml from flask import Blueprint, request, jsonify, render_template, current_app from admin.dashboard.blueprints.auth import login_required from utils.robot_cmd.robot_command import GroupBotManager, PermissionStatus # 创建蓝图 plugin_routes = Blueprint('plugin_routes', __name__) LOG = logger # 机器人管理页面 @plugin_routes.route('/plugins_manage') @login_required def robot_management(): return render_template('plugins_manage.html') @plugin_routes.route('/api/plugins', methods=['GET']) @login_required def get_plugins(): """获取所有插件列表""" try: server = current_app.dashboard_server # 获取插件注册表 plugins = server.plugin_registry.get_all_plugins() # 转换为前端需要的格式 plugin_list = [] for name, plugin in plugins.items(): # 获取插件模块名 try: module_name = plugin.__class__.__module__.split('.')[-2] except (IndexError, AttributeError): module_name = "unknown" plugin_info = { "name": plugin.name, "module_name": module_name, "version": getattr(plugin, 'version', 'N/A'), "author": getattr(plugin, 'author', 'N/A'), "description": getattr(plugin, 'description', 'N/A'), "status": plugin.status.name if hasattr(plugin, 'status') else 'UNKNOWN' } plugin_list.append(plugin_info) return jsonify({"success": True, "data": plugin_list}) except Exception as e: LOG.error(f"获取插件列表失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"获取插件列表失败: {str(e)}"}) @plugin_routes.route('/api/plugins/group_status', methods=['GET']) @login_required def get_plugin_group_status(): """获取单个插件在各群的启用状态（已启用/未启用）""" try: server = current_app.dashboard_server plugin_name = request.args.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 通过现有插件管理器查找插件实例，兼容传入模块名或展示名两种形式。 display_name, plugin = server.plugin_manager.find_plugin_by_name(plugin_name) if not plugin: return jsonify({"success": False, "message": f"未找到插件: {plugin_name}"}) # 统一构建群列表：优先使用通讯录中的群（覆盖面更全），并补齐机器人管理列表中的群。 group_contacts = {} try: group_contacts = (server.contact_manager.get_group_contacts() or {}) except Exception: # 这里降级为空字典，后续仍会使用 GroupBotManager 群列表，不中断主流程。 group_contacts = {} # 从群机器人管理缓存中取群，确保“已托管但通讯录暂缺”的群也会展示出来。 managed_groups = set(GroupBotManager.get_group_list() or []) contact_groups = set(group_contacts.keys()) all_group_ids = sorted(contact_groups | managed_groups) # 再做一次兜底：极端情况下如果前两者都为空，尝试读取本地缓存中的群集合。 if not all_group_ids and hasattr(GroupBotManager, "local_cache"): all_group_ids = sorted(GroupBotManager.local_cache.get("group_list", set()) or []) # 插件是否支持“按群开关”依赖于插件是否注册了 feature（多数消息插件会注册）。 plugin_feature = getattr(plugin, "feature", None) supports_group_switch = plugin_feature is not None feature_key = getattr(plugin_feature, "name", "") if plugin_feature else "" feature_description = getattr(plugin_feature, "description", "") if plugin_feature else "" enabled_groups = [] disabled_groups = [] for group_id in all_group_ids: # 群名优先从通讯录获取，拿不到时降级为群ID，保证页面总能显示。 group_name = group_contacts.get(group_id) or server.contact_manager.get_nickname(group_id) or group_id group_item = { "group_id": group_id, "group_name": group_name, } # 支持群开关的插件：按 feature 的真实权限划分已开启/未开启。 if supports_group_switch: permission_status = GroupBotManager.get_group_permission(group_id, plugin_feature) if permission_status == PermissionStatus.ENABLED: enabled_groups.append(group_item) else: disabled_groups.append(group_item) else: # 不支持群级开关的插件默认归入“未开启”列表，并通过 supports_group_switch 让前端给提示。 disabled_groups.append(group_item) return jsonify({ "success": True, "data": { "plugin_name": plugin.name, "module_name": plugin_name, "supports_group_switch": supports_group_switch, "feature_key": feature_key, "feature_description": feature_description, "enabled_count": len(enabled_groups), "disabled_count": len(disabled_groups), "total_group_count": len(all_group_ids), "enabled_groups": enabled_groups, "disabled_groups": disabled_groups, } }) except Exception as e: LOG.error(f"获取插件群状态失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"获取插件群状态失败: {str(e)}"}) @plugin_routes.route('/api/plugins/group_status/toggle', methods=['POST']) @login_required def toggle_plugin_group_status(): """切换插件在指定群的启用状态（开启/关闭）""" try: server = current_app.dashboard_server data = request.get_json() or {} plugin_name = (data.get('plugin_name') or '').strip() group_id = (data.get('group_id') or '').strip() status = (data.get('status') or '').strip().lower() # 参数校验：保证前端传入完整且合法的切换目标。 if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) if not group_id: return jsonify({"success": False, "message": "缺少群ID参数"}) if status not in ('enabled', 'disabled'): return jsonify({"success": False, "message": "状态参数非法，仅支持 enabled / disabled"}) # 查找插件实例，沿用现有查找逻辑，兼容模块名与展示名。 display_name, plugin = server.plugin_manager.find_plugin_by_name(plugin_name) if not plugin: return jsonify({"success": False, "message": f"未找到插件: {plugin_name}"}) # 插件必须注册了 feature 才能进行群级开关；否则无法映射到权限系统。 plugin_feature = getattr(plugin, "feature", None) if plugin_feature is None: return jsonify({"success": False, "message": "该插件未接入群级开关能力，无法直接切换"}) # 将字符串状态映射到统一权限枚举，然后写入群权限缓存与Redis。 target_status = PermissionStatus.ENABLED if status == 'enabled' else PermissionStatus.DISABLED GroupBotManager.set_group_permission(group_id, plugin_feature, target_status) # 返回简洁确认信息，方便前端即时提示与二次渲染。 return jsonify({ "success": True, "message": f"已将群 {group_id} 的插件“{plugin.name}”设置为{'开启' if target_status == PermissionStatus.ENABLED else '关闭'}", "data": { "plugin_name": plugin.name, "group_id": group_id, "status": target_status.value, "feature_key": getattr(plugin_feature, "name", ""), } }) except Exception as e: LOG.error(f"切换插件群状态失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"切换插件群状态失败: {str(e)}"}) @plugin_routes.route('/api/plugins/info', methods=['GET']) @login_required def get_plugin_info(): """获取插件详细信息""" try: server = current_app.dashboard_server plugin_name = request.args.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 获取插件管理器 display_name, plugin = server.plugin_manager.find_plugin_by_name(plugin_name) if not plugin: return jsonify({"success": False, "message": f"未找到插件: {plugin_name}"}) # 获取插件模块名 try: module_name = plugin.__class__.__module__.split('.')[-2] except (IndexError, AttributeError): module_name = "unknown" # 构建详细信息 plugin_info = { "name": plugin.name, "module_name": module_name, "version": getattr(plugin, 'version', 'N/A'), "author": getattr(plugin, 'author', 'N/A'), "description": getattr(plugin, 'description', 'N/A'), "status": plugin.status.name if hasattr(plugin, 'status') else 'UNKNOWN', "command_prefix": getattr(plugin, 'command_prefix', ''), "commands": getattr(plugin, 'commands', []), "config": getattr(plugin, '_config', {}) } return jsonify({"success": True, "data": plugin_info}) except Exception as e: LOG.error(f"获取插件详情失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"获取插件详情失败: {str(e)}"}) @plugin_routes.route('/api/plugins/enable', methods=['POST']) @login_required def enable_plugin(): """启用插件""" try: server = current_app.dashboard_server data = request.get_json() plugin_name = data.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 获取插件管理器 # 启用插件 if server.plugin_manager.start_plugin(plugin_name): return jsonify({"success": True, "message": f"插件 {plugin_name} 启用成功"}) else: return jsonify({"success": False, "message": f"插件 {plugin_name} 启用失败"}) except Exception as e: LOG.error(f"启用插件失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"启用插件失败: {str(e)}"}) @plugin_routes.route('/api/plugins/disable', methods=['POST']) @login_required def disable_plugin(): """禁用插件""" try: server = current_app.dashboard_server data = request.get_json() plugin_name = data.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 禁用插件 if server.plugin_manager.stop_plugin(plugin_name): return jsonify({"success": True, "message": f"插件 {plugin_name} 禁用成功"}) else: return jsonify({"success": False, "message": f"插件 {plugin_name} 禁用失败"}) except Exception as e: LOG.error(f"禁用插件失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"禁用插件失败: {str(e)}"}) @plugin_routes.route('/api/plugins/reload', methods=['POST']) @login_required def reload_plugin(): """重载插件""" try: server = current_app.dashboard_server data = request.get_json() plugin_name = data.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 重载插件 reloaded_plugin = server.plugin_manager.reload_plugin(plugin_name) if reloaded_plugin: return jsonify({"success": True, "message": f"插件 {plugin_name} 重载成功"}) else: return jsonify({"success": False, "message": f"插件 {plugin_name} 重载失败"}) except Exception as e: LOG.error(f"重载插件失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"重载插件失败: {str(e)}"}) @plugin_routes.route('/api/plugins/config/raw', methods=['GET']) @login_required def get_raw_plugin_config(): """获取插件原始配置文件内容""" try: server = current_app.dashboard_server plugin_name = request.args.get('plugin_name') if not plugin_name: return jsonify({"success": False, "message": "缺少插件名称参数"}) # 获取插件管理器 # 查找插件 display_name, plugin = server.plugin_manager.find_plugin_by_name(plugin_name) if not plugin: return jsonify({"success": False, "message": f"未找到插件: {plugin_name}"}) # 获取配置文件路径 config_path = plugin.get_config_path() if not os.path.exists(config_path): return jsonify({"success": False, "message": f"配置文件不存在: {config_path}"}) # 读取配置文件内容 with open(config_path, 'r', encoding='utf-8') as f: config_text = f.read() # 确定配置文件格式 format_type = 'toml' if config_path.endswith('.json'): format_type = 'json' elif config_path.endswith('.yaml') or config_path.endswith('.yml'): format_type = 'yaml' return jsonify({ "success": True, "data": config_text, "format": format_type }) except Exception as e: LOG.error(f"获取插件配置文件失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"获取配置文件失败: {str(e)}"}) @plugin_routes.route('/api/plugins/config/update', methods=['POST']) @login_required def update_plugin_config(): """更新插件配置""" try: server = current_app.dashboard_server data = request.json plugin_name = data.get('plugin_name') config_text = data.get('config_text') format_type = data.get('format', 'toml') if not plugin_name or config_text is None: return jsonify({"success": False, "message": "缺少必要参数"}) # 查找插件 # 获取插件管理器 display_name, plugin = server.plugin_manager.find_plugin_by_name(plugin_name) if not plugin: return jsonify({"success": False, "message": f"未找到插件: {plugin_name}"}) # 获取配置文件路径 config_path = plugin.get_config_path() # 确保配置目录存在 os.makedirs(os.path.dirname(config_path), exist_ok=True) # 写入配置文件 with open(config_path, 'w', encoding='utf-8') as f: f.write(config_text) # 解析配置并更新插件内部配置 try: if format_type == 'toml': config_obj = toml.loads(config_text) elif format_type == 'json': config_obj = json.loads(config_text) else: return jsonify({"success": False, "message": f"不支持的配置格式: {format_type}"}) # 更新插件内部配置 plugin._config = config_obj return jsonify({"success": True, "message": "配置已保存"}) except Exception as e: LOG.error(f"解析配置失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"配置已保存，但解析失败: {str(e)}"}) except Exception as e: LOG.error(f"更新插件配置失败: {str(e)}", exc_info=True) return jsonify({"success": False, "message": f"更新配置失败: {str(e)}"})