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 # 创建蓝图 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/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)}"})