#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio import shlex from typing import Dict, List, Callable, Any from dataclasses import dataclass import os logger = logging.getLogger(__name__) @dataclass class Command: """命令数据类""" name: str handler: Callable description: str permissions: List[str] source: str = "internal" class CommandService: """指令服务""" def __init__(self, service_manager=None): self.commands: Dict[str, Command] = {} self.command_history: List[Dict] = [] self.max_history_size = 100 self.service_manager = service_manager # 添加服务管理器引用 logger.debug("CommandService初始化开始") def register_command(self, name: str, handler: Callable, description: str = "", permissions: List[str] = None, source: str = "plugins"): """注册命令""" # 来源如 internal plugins system 等 try: if name in self.commands: logger.warning(f"命令 {name} 已存在，将被覆盖") self.commands[name] = Command( name=name, handler=handler, description=description or f"命令: {name}", permissions=permissions or [], source=source ) logger.debug(f"注册命令: {name} (来源: {source})") except Exception as e: logger.error(f"注册命令 {name} 时出错: {str(e)}", exc_info=True) raise async def _handle_permission_command(self, command: str, args: List[str], source: str) -> str: """处理权限相关命令""" try: permission_service = self.service_manager.get_service("permission") if not permission_service: return "❌ 权限服务不可用" return await permission_service.process_permission_command(command, args) except Exception as e: logger.error(f"处理权限命令时出错: {str(e)}", exc_info=True) return f"❌ 处理权限命令时出错: {str(e)}" async def process_command(self, command_string: str, source: str = "unknown") -> Any: """处理命令""" try: logger.debug(f"处理命令: '{command_string}' (来源: {source})") # 解析命令 parts = shlex.split(command_string.strip()) if not parts: logger.warning("空命令") return "空命令" command_name = parts[0] args = parts[1:] # 记录命令历史 self._add_to_history(command_string, source) # 检查权限命令 permission_commands = ['pmallow', 'pmdeny', 'pmignore', 'permissions', 'pmpending', 'pmrequests', 'pm_plugin_status', 'pmhelp', 'pmtest'] if command_name in permission_commands: permission_service = self.service_manager.get_service("permission") if not permission_service: return "❌ 权限服务不可用" # 直接调用权限服务处理命令 return await permission_service.process_permission_command(command_name, args) # 查找其他命令 if command_name not in self.commands: logger.warning(f"未知命令: {command_name}") return f"未知命令: {command_name}" command = self.commands[command_name] # 执行命令 try: result = await self._execute_command(command, args, source) logger.debug(f"命令执行成功: {command_name}") return result except Exception as e: logger.error(f"命令执行失败 {command_name}: {str(e)}", exc_info=True) return f"命令执行错误: {str(e)}" except Exception as e: logger.error(f"处理命令时出错: {str(e)}", exc_info=True) return f"命令处理错误: {str(e)}" async def _execute_command(self, command: Command, args: List[str], source: str) -> Any: """执行命令""" try: # 检查处理器类型 if asyncio.iscoroutinefunction(command.handler): result = await command.handler(*args) else: result = command.handler(*args) logger.debug(f"命令 {command.name} 执行完成") return result except TypeError as e: logger.error(f"命令参数错误 {command.name}: {str(e)}", exc_info=True) raise ValueError(f"参数错误: {str(e)}") except Exception as e: logger.error(f"命令执行异常 {command.name}: {str(e)}", exc_info=True) raise def _add_to_history(self, command: str, source: str): """添加到命令历史""" try: history_entry = { "command": command, "source": source, "timestamp": asyncio.get_event_loop().time() } self.command_history.append(history_entry) # 限制历史记录大小 if len(self.command_history) > self.max_history_size: self.command_history.pop(0) logger.debug(f"命令历史记录添加，当前大小: {len(self.command_history)}") except Exception as e: logger.error(f"添加命令历史时出错: {str(e)}", exc_info=True) def get_command_list(self) -> List[Dict]: """获取命令列表""" try: command_list = [] for name, cmd in self.commands.items(): command_list.append({ "name": name, "description": cmd.description, "permissions": cmd.permissions, "source": cmd.source }) logger.debug(f"获取命令列表，共 {len(command_list)} 个命令") return command_list except Exception as e: logger.error(f"获取命令列表时出错: {str(e)}", exc_info=True) return [] def get_command_history(self, limit: int = 10) -> List[Dict]: """获取命令历史""" try: history = self.command_history[-limit:] logger.debug(f"获取命令历史，返回 {len(history)} 条记录") return history except Exception as e: logger.error(f"获取命令历史时出错: {str(e)}", exc_info=True) return [] def register_builtin_commands(self): """注册内置命令""" try: logger.debug("开始注册内置命令") # 帮助命令 self.register_command( name="help", handler=self._cmd_help, description="显示帮助信息", permissions=["framework.command.help.read"], source="internal" ) # 测试日志命令 self.register_command( name="testlog", handler=self._cmd_test_log, description="生成测试日志", permissions=["framework.command.test"], source="internal" ) # 状态命令 self.register_command( name="status", handler=self._cmd_status, description="显示框架状态", permissions=["framework.status.read"], source="internal" ) # 历史命令 self.register_command( name="history", handler=self._cmd_history, description="显示命令历史", permissions=["framework.command.history.read"], source="internal" ) # 网络诊断 self.register_command( name="netdiag", handler=self._cmd_netdiag, description="网络服务诊断", permissions=["framework.network.diagnose"], source="internal" ) # 权限管理命令组 - 保留注册但不使用（在process_command中直接处理） # 这些注册是为了在help命令中显示 permission_commands = [ ("pmallow", "权限管理: 同意权限请求"), ("pmdeny", "权限管理: 拒绝权限请求"), ("pmignore", "权限管理: 暂时忽略权限请求"), ("permissions", "权限管理: 显示权限状态"), ("pmpending", "权限管理: 查看待授权请求列表"), ("pmrequests", "权限管理: 查看待授权请求列表(别名)"), ("pm_plugin_status", "权限管理: 查看插件权限状态"), ("pmtest", "权限管理: 测试权限配置文件"), ("pmhelp", "权限管理: 显示权限命令帮助") ] for cmd_name, description in permission_commands: self.register_command( name=cmd_name, handler=self._cmd_permission, # 使用统一的备用处理器 description=description, permissions=["framework.permission.read"], source="internal" ) # 滚动控制命令组 scroll_commands = [ ("scroll", "滚动控制: 手动滚动到底部"), ("autoscroll", "滚动控制: 切换自动滚动") ] # 脚手架命令 self.register_command( name="create-plugin", handler=self._cmd_create_plugin, description="创建新插件脚手架", permissions=["framework.scaffold.plugin"], source="internal" ) logger.info(f"内置命令注册完成，共注册 {len(self.commands)} 个命令") for cmd_name, description in scroll_commands: self.register_command( name=cmd_name, handler=self._cmd_scroll_control, description=description, permissions=["framework.tui.control"], source="internal" ) logger.info(f"内置命令注册完成，共注册 {len(self.commands)} 个命令") except Exception as e: logger.error(f"注册内置命令时出错: {str(e)}", exc_info=True) raise async def _cmd_netdiag(self, *args) -> str: """网络诊断命令""" try: internet_service = self.service_manager.get_service("internet") result = ["🔧 **网络服务诊断报告**"] result.append("=" * 50) if not internet_service: result.append("❌ 网络服务未注册") result.append("\n💡 **可能的原因:**") result.append(" 1. 网络服务启动失败") result.append(" 2. 依赖包缺失 (aiohttp)") result.append(" 3. 端口被占用") result.append(" 4. 权限不足") result.append("\n🔧 **解决方案:**") result.append(" - 检查上方日志中的错误信息") result.append(" - 运行: pip install aiohttp") result.append(" - 尝试更换端口号") result.append(" - 使用 sudo (如果需要)") return "\n".join(result) # 获取健康信息 health_info = await internet_service.check_service_health() result.append(f"🔄 服务运行: {'✅ 是' if health_info.get('is_running') else '❌ 否'}") result.append(f"🔌 HTTP端口: {health_info.get('http_port', 'N/A')}") result.append(f"📡 WebSocket端口: {health_info.get('websocket_port', 'N/A')}") result.append(f"🌐 HTTP活跃: {'✅ 是' if health_info.get('http_active') else '❌ 否'}") result.append(f"📦 依赖状态: {'✅ 正常' if health_info.get('dependencies_available') else '❌ 缺失'}") if health_info.get('error'): result.append(f"❌ 错误信息: {health_info['error']}") # 端口占用检查 if not health_info.get('http_active') and health_info.get('is_running'): result.append("\n⚠️ **端口问题检测:**") result.append(" HTTP服务已启动但端口未响应") result.append(" 可能被防火墙阻止或配置错误") # 路由信息 routes = internet_service.get_plugin_routes() total_routes = sum(len(plugin_routes) for plugin_routes in routes.values()) result.append(f"\n🛣️ 注册路由: {total_routes} 个") for plugin_name, plugin_routes in routes.items(): result.append(f" 📍 {plugin_name}: {len(plugin_routes)} 个路由") return "\n".join(result) except Exception as e: logger.error(f"网络诊断命令执行失败: {str(e)}") return f"❌ 网络诊断失败: {str(e)}" async def _cmd_scroll_control(self, *args) -> str: """处理滚动控制命令""" try: tui_service = self.service_manager.get_service("tui") if not tui_service: return "❌ TUI服务不可用" if not args: return "🔧 滚动控制命令\n💡 使用: scroll [log|message|all]\n💡 使用: autoscroll [on|off|toggle] [log|message|all]" command = args[0].lower() if command == "scroll": target = args[1] if len(args) > 1 else "all" if target not in ["log", "message", "all"]: return "❌ 无效的目标，请使用: log, message, all" return tui_service.scroll_to_bottom(target) elif command == "autoscroll": if len(args) < 2: return "❌ 请指定操作: on, off, toggle" action = args[1].lower() target = args[2] if len(args) > 2 else "all" if target not in ["log", "message", "all"]: return "❌ 无效的目标，请使用: log, message, all" if action == "on": return tui_service.toggle_auto_scroll(target, True) elif action == "off": return tui_service.toggle_auto_scroll(target, False) elif action == "toggle": return tui_service.toggle_auto_scroll(target, None) else: return "❌ 无效的操作，请使用: on, off, toggle" else: return "❌ 未知滚动命令\n💡 可用命令: scroll, autoscroll" except Exception as e: logger.error(f"处理滚动命令时出错: {str(e)}") return f"❌ 滚动命令错误: {str(e)}" async def _cmd_permission(self, *args) -> str: """处理权限相关命令 - 备用处理器""" try: permission_service = self.service_manager.get_service("permission") if not permission_service: return "❌ 权限服务不可用" # 如果没有参数，显示通用帮助 if not args: return "🔐 权限管理命令\n💡 使用 pmhelp 查看详细帮助" # 否则直接转发到权限服务 command_name = str(args[0]).lower() permission_args = [str(arg) for arg in args[1:]] if len(args) > 1 else [] return await permission_service.process_permission_command(command_name, permission_args) except Exception as e: logger.error(f"处理权限命令时出错: {str(e)}", exc_info=True) return f"❌ 权限命令错误: {str(e)}" async def _cmd_help(self, *args) -> str: """帮助命令处理器""" try: commands = self.get_command_list() if not commands: return "❌ 没有可用的命令" help_text = ["📋 **可用命令:**", ""] # 按来源分组显示命令 commands_by_source = {} for cmd in commands: source = cmd['source'] if source not in commands_by_source: commands_by_source[source] = [] commands_by_source[source].append(cmd) # 显示内置命令 if 'internal' in commands_by_source: help_text.append("🔧 **内置命令:**") for cmd in commands_by_source['internal']: help_text.append(f" 🟢 {cmd['name']:15} - {cmd['description']}") help_text.append("") # 显示插件命令 if 'plugin' in commands_by_source: help_text.append("🔌 **插件命令:**") for cmd in commands_by_source['plugin']: help_text.append(f" 🟡 {cmd['name']:15} - {cmd['description']}") help_text.append("") # 显示系统命令 if 'system' in commands_by_source: help_text.append("⚙️ **系统命令:**") for cmd in commands_by_source['system']: help_text.append(f" 🔵 {cmd['name']:15} - {cmd['description']}") # 添加使用提示 help_text.extend([ "", "💡 **使用提示:**", " - 输入命令名称执行命令", " - 使用 'status' 查看框架状态", " - 使用 'history' 查看命令历史", " - 使用 'permissions' 管理插件权限" ]) return "\n".join(help_text) except Exception as e: logger.error(f"处理help命令时出错: {str(e)}", exc_info=True) return f"❌ 帮助命令错误: {str(e)}" async def _cmd_status(self, *args) -> str: """状态命令处理器""" try: status_info = [ f"命令服务状态:", f" 注册命令数: {len(self.commands)}", f" 历史记录数: {len(self.command_history)}", f" 最大历史大小: {self.max_history_size}" ] return "\n".join(status_info) except Exception as e: logger.error(f"处理status命令时出错: {str(e)}", exc_info=True) return f"状态命令错误: {str(e)}" async def _cmd_history(self, *args) -> str: """历史命令处理器""" try: limit = 10 if args and args[0].isdigit(): limit = min(int(args[0]), 50) # 限制最大50条 history = self.get_command_history(limit) if not history: return "没有命令历史" history_text = [f"最近 {len(history)} 条命令历史:"] for i, entry in enumerate(reversed(history), 1): history_text.append(f" {i}. [{entry['source']}] {entry['command']}") return "\n".join(history_text) except Exception as e: logger.error(f"处理history命令时出错: {str(e)}", exc_info=True) return f"历史命令错误: {str(e)}" async def _cmd_test_log(self, *args) -> str: """测试日志命令""" try: logger.debug("这是一条DEBUG测试日志") logger.info("这是一条INFO测试日志") logger.warning("这是一条WARNING测试日志") logger.error("这是一条ERROR测试日志") return "✅ 测试日志已生成，请检查TUI显示" except Exception as e: return f"❌ 测试日志生成失败: {str(e)}" async def _cmd_create_plugin(self, *args) -> str: """创建新插件脚手架""" try: import re import shutil from pathlib import Path from string import Template # 1. 参数解析 if not args: return "❌ 用法: create-plugin <插件名> [--author <作者>] [--desc <描述>]\n💡 插件名需为小写字母/数字/下划线，如: my_cool_plugin" plugin_name = args[0] author = "Unknown" description = "暂无描述" # 解析可选参数 i = 1 while i < len(args): if args[i] == "--author" and i + 1 < len(args): author = args[i+1] i += 2 elif args[i] == "--desc" and i + 1 < len(args): description = args[i+1] i += 2 else: i += 1 # 2. 命名校验 if not re.match(r'^[a-z][a-z0-9_]*$', plugin_name): return "❌ 插件名格式错误。请使用小写字母开头，仅包含小写字母、数字和下划线（如: data_sync）" plugin_dir = Path("plugins") / plugin_name if plugin_dir.exists(): return f"❌ 插件目录已存在: {plugin_dir}" # 3. 模板路径 template_dir = Path(os.getenv("SENSU_CODE_DIR", ".")) / "templates" / "plugin" if not template_dir.exists(): return "❌ 模板目录不存在: templates/plugin/" # 4. 创建目录与渲染文件 plugin_dir.mkdir(parents=True, exist_ok=True) context = { "plugin_name": plugin_name, "author": author, "description": description } for template_file in template_dir.iterdir(): if template_file.is_file() and template_file.name.endswith(".template"): with open(template_file, 'r', encoding='utf-8') as f: tpl = Template(f.read()) content = tpl.safe_substitute(context) target_name = template_file.stem target_path = plugin_dir / target_name with open(target_path, 'w', encoding='utf-8') as f: f.write(content) logger.debug(f"脚手架文件生成: {target_path}") return ( f"✅ 插件脚手架创建成功!\n" f"📁 路径: {plugin_dir}\n" f"👤 作者: {author}\n" f"📝 描述: {description}\n\n" f"🔧 下一步:\n" f" 1. 编辑 {plugin_dir}/__init__.py 实现业务逻辑\n" f" 2. 运行框架自动加载插件\n" f" 3. 使用 `help` 查看可用命令" ) except Exception as e: logger.error(f"创建插件脚手架失败: {str(e)}", exc_info=True) return f"❌ 创建失败: {str(e)}" def shutdown(self): """关闭指令服务""" try: logger.info("关闭指令服务") self.commands.clear() self.command_history.clear() logger.debug("指令服务关闭完成") except Exception as e: logger.error(f"关闭指令服务时出错: {str(e)}", exc_info=True)