abot/admin/dashboard/blueprints/plugin_routes.py

import logging
from flask import Blueprint, request, jsonify, render_template

from admin.dashboard.blueprints.auth import login_required
from plugin_common.plugin_manager import PluginManager
from plugin_common.plugin_registry import PluginRegistry

# 创建蓝图
plugin_routes = Blueprint('plugin_routes', __name__)
LOG = logging.getLogger(__name__)


# 机器人管理页面
@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:
        # 获取插件注册表
        plugin_registry = PluginRegistry()
        plugins = 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:
        plugin_name = request.args.get('plugin_name')
        if not plugin_name:
            return jsonify({"success": False, "message": "缺少插件名称参数"})
            
        # 获取插件管理器
        plugin_manager = PluginManager().get_instance()
        display_name, plugin = 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:
        data = request.get_json()
        plugin_name = data.get('plugin_name')
        if not plugin_name:
            return jsonify({"success": False, "message": "缺少插件名称参数"})
            
        # 获取插件管理器
        plugin_manager = PluginManager().get_instance()
        
        # 启用插件
        if 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:
        data = request.get_json()
        plugin_name = data.get('plugin_name')
        if not plugin_name:
            return jsonify({"success": False, "message": "缺少插件名称参数"})
            
        # 获取插件管理器
        plugin_manager = PluginManager().get_instance()
        
        # 禁用插件
        if 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:
        data = request.get_json()
        plugin_name = data.get('plugin_name')
        if not plugin_name:
            return jsonify({"success": False, "message": "缺少插件名称参数"})
            
        # 获取插件管理器
        plugin_manager = PluginManager().get_instance()
        
        # 重载插件
        reloaded_plugin = 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)}"})