SenSu/services/command_service.py

#!/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)