#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio from typing import Dict, List, Any, Optional from aiohttp import web import json logger = logging.getLogger(__name__) class GUIAPI: """GUI API接口 - 为未来的GUI界面提供操作接口""" def __init__(self, service_manager): self.service_manager = service_manager self.app = web.Application() self.setup_routes() logger.debug("GUIAPI初始化开始") def setup_routes(self): """设置API路由""" try: # 框架状态接口 self.app.router.add_get('/api/framework/status', self.handle_framework_status) self.app.router.add_get('/api/framework/health', self.handle_framework_health) # 日志接口 self.app.router.add_get('/api/logs/recent', self.handle_recent_logs) self.app.router.add_get('/api/logs/stream', self.handle_log_stream) # 插件接口 self.app.router.add_get('/api/plugins', self.handle_plugins_list) self.app.router.add_post('/api/plugins/{plugin_name}/reload', self.handle_plugin_reload) self.app.router.add_post('/api/plugins/{plugin_name}/unload', self.handle_plugin_unload) # 命令接口 self.app.router.add_post('/api/command', self.handle_command_execute) self.app.router.add_get('/api/command/history', self.handle_command_history) # 服务接口 self.app.router.add_get('/api/services', self.handle_services_list) self.app.router.add_get('/api/services/{service_name}/status', self.handle_service_status) logger.debug("GUI API路由设置完成") except Exception as e: logger.error(f"设置GUI API路由时出错: {str(e)}", exc_info=True) raise async def handle_framework_status(self, request): """处理框架状态请求""" try: logger.debug("处理框架状态API请求") status_info = { "framework": { "name": "CatFramework", "version": "1.0.0", "status": "running", "uptime": "0s" # 实际应该计算运行时间 }, "services": { "total": 0, "running": 0 }, "plugins": { "total": 0, "loaded": 0 } } # 获取服务状态 try: service_manager = self.service_manager.get_service("service_manager") if service_manager: status_info["services"]["total"] = len(service_manager.services) status_info["services"]["running"] = len(service_manager.services) except: pass # 获取插件状态 try: plugin_service = self.service_manager.get_service("plugin") if plugin_service: status_info["plugins"]["total"] = len(plugin_service.plugin_info) status_info["plugins"]["loaded"] = len(plugin_service.plugins) except: pass return web.json_response(status_info) except Exception as e: logger.error(f"处理框架状态请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_framework_health(self, request): """处理框架健康检查""" try: logger.debug("处理框架健康检查API请求") health_data = { "status": "healthy", "timestamp": asyncio.get_event_loop().time(), "components": { "core_services": "healthy", "plugins": "healthy", "network": "healthy" } } return web.json_response(health_data) except Exception as e: logger.error(f"处理框架健康检查时出错: {str(e)}", exc_info=True) return web.json_response({"status": "unhealthy", "error": str(e)}, status=503) async def handle_recent_logs(self, request): """处理最近日志请求""" try: logger.debug("处理最近日志API请求") count = int(request.query.get('count', 50)) level = request.query.get('level', '') log_service = self.service_manager.get_service("log") if not log_service: return web.json_response({"error": "Log service not available"}, status=503) logs = log_service.get_recent_logs(count) # 按级别过滤 if level: logs = [log for log in logs if log['level'].lower() == level.lower()] return web.json_response({"logs": logs, "count": len(logs)}) except Exception as e: logger.error(f"处理最近日志请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_log_stream(self, request): """处理日志流请求（SSE）""" try: logger.debug("处理日志流API请求") response = web.StreamResponse() response.headers['Content-Type'] = 'text/event-stream' response.headers['Cache-Control'] = 'no-cache' response.headers['Connection'] = 'keep-alive' await response.prepare(request) # 这里应该实现真正的日志流 # 暂时发送测试数据 try: while True: test_log = { "timestamp": asyncio.get_event_loop().time(), "level": "INFO", "message": "Log stream test message", "source": "gui_api" } event_data = f"data: {json.dumps(test_log)}\n\n" await response.write(event_data.encode('utf-8')) await asyncio.sleep(5) except asyncio.CancelledError: logger.debug("日志流连接关闭") finally: await response.write_eof() return response except Exception as e: logger.error(f"处理日志流请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_plugins_list(self, request): """处理插件列表请求""" try: logger.debug("处理插件列表API请求") plugin_service = self.service_manager.get_service("plugin") if not plugin_service: return web.json_response({"error": "Plugin service not available"}, status=503) plugins_info = [] for plugin_name, plugin_info in plugin_service.plugin_info.items(): plugins_info.append({ "name": plugin_info.name, "version": plugin_info.version, "description": plugin_info.description, "author": plugin_info.author, "enabled": plugin_info.enabled, "loaded": plugin_info.loaded, "error_count": plugin_info.error_count, "permissions": plugin_info.permissions }) return web.json_response({"plugins": plugins_info}) except Exception as e: logger.error(f"处理插件列表请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_plugin_reload(self, request): """处理插件重载请求""" try: plugin_name = request.match_info['plugin_name'] logger.debug(f"处理插件重载API请求: {plugin_name}") plugin_service = self.service_manager.get_service("plugin") if not plugin_service: return web.json_response({"error": "Plugin service not available"}, status=503) # 先卸载再加载 unload_success = await plugin_service.unload_plugin(plugin_name) if unload_success: load_success = await plugin_service.load_plugin(plugin_name) result = {"reloaded": load_success} else: result = {"reloaded": False, "error": "Unload failed"} return web.json_response(result) except Exception as e: logger.error(f"处理插件重载请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_plugin_unload(self, request): """处理插件卸载请求""" try: plugin_name = request.match_info['plugin_name'] logger.debug(f"处理插件卸载API请求: {plugin_name}") plugin_service = self.service_manager.get_service("plugin") if not plugin_service: return web.json_response({"error": "Plugin service not available"}, status=503) success = await plugin_service.unload_plugin(plugin_name) return web.json_response({"unloaded": success}) except Exception as e: logger.error(f"处理插件卸载请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_command_execute(self, request): """处理命令执行请求""" try: data = await request.json() command = data.get('command', '') source = data.get('source', 'gui') logger.debug(f"处理命令执行API请求: {command}") command_service = self.service_manager.get_service("command") if not command_service: return web.json_response({"error": "Command service not available"}, status=503) result = await command_service.process_command(command, source) return web.json_response({ "command": command, "result": str(result), "success": True }) except Exception as e: logger.error(f"处理命令执行请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_command_history(self, request): """处理命令历史请求""" try: limit = int(request.query.get('limit', 10)) logger.debug(f"处理命令历史API请求，限制: {limit}") command_service = self.service_manager.get_service("command") if not command_service: return web.json_response({"error": "Command service not available"}, status=503) history = command_service.get_command_history(limit) return web.json_response({"history": history}) except Exception as e: logger.error(f"处理命令历史请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_services_list(self, request): """处理服务列表请求""" try: logger.debug("处理服务列表API请求") service_manager = self.service_manager.get_service("service_manager") if not service_manager: return web.json_response({"error": "Service manager not available"}, status=503) services_info = [] for name, service in service_manager.services.items(): services_info.append({ "name": name, "type": type(service).__name__, "status": "running" # 简化状态 }) return web.json_response({"services": services_info}) except Exception as e: logger.error(f"处理服务列表请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def handle_service_status(self, request): """处理服务状态请求""" try: service_name = request.match_info['service_name'] logger.debug(f"处理服务状态API请求: {service_name}") try: service = self.service_manager.get_service(service_name) status_info = { "name": service_name, "available": True, "status": "running" } # 可以添加特定服务的状态检查 if hasattr(service, 'is_running'): status_info["status"] = "running" if service.is_running else "stopped" return web.json_response(status_info) except ValueError: return web.json_response({"error": f"Service {service_name} not found"}, status=404) except Exception as e: logger.error(f"处理服务状态请求时出错: {str(e)}", exc_info=True) return web.json_response({"error": str(e)}, status=500) async def start(self, host: str = "localhost", port: int = 8080): """启动GUI API服务器""" try: logger.info(f"启动GUI API服务器: {host}:{port}") runner = web.AppRunner(self.app) await runner.setup() site = web.TCPSite(runner, host, port) await site.start() logger.debug("GUI API服务器启动完成") return runner except Exception as e: logger.error(f"启动GUI API服务器时出错: {str(e)}", exc_info=True) raise async def shutdown(self): """关闭GUI API""" try: logger.info("关闭GUI API") # 清理资源 logger.debug("GUI API关闭完成") except Exception as e: logger.error(f"关闭GUI API时出错: {str(e)}", exc_info=True)