SenSu/services/permission_service.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import logging
import asyncio
import uuid
import json
from typing import Dict, List, Set, Optional
from pathlib import Path
import yaml

logger = logging.getLogger(__name__)

class PermissionService:
    """权限服务"""
    
    def __init__(self, config: Dict, tui_service, core_bridge):
        self.config = config
        self.tui_service = tui_service
        self.core_bridge = core_bridge
        self.permission_rules: Dict = {}
        self.granted_permissions: Dict[str, Set[str]] = {}
        self.pending_requests: Dict[str, Dict] = {}
        self.plugin_status: Dict[str, str] = {}  # 插件状态跟踪
        self.is_running = False
        
        # 配置文件路径
        self.config_dir = Path("config") / "permissions"
        self.granted_file = self.config_dir / "granted_permissions.json"
        self.pending_file = self.config_dir / "pending_requests.json"
        self.plugin_status_file = self.config_dir / "plugin_status.json"
        
        # 确保配置目录存在
        self.config_dir.mkdir(parents=True, exist_ok=True)
        
        logger.debug("PermissionService初始化开始")
    
    async def start(self):
        """启动权限服务"""
        try:
            logger.info("启动权限服务")
            
            # 检查核心桥接服务是否可用
            if not self.core_bridge:
                logger.error("核心桥接服务不可用")
                return False
            
            # 加载权限规则
            await self._load_permission_rules()
            
            # 加载持久化数据
            await self._load_persisted_data()
            
            # 订阅权限相关事件
            self.core_bridge.subscribe("permission.request", self._handle_permission_request)
            self.core_bridge.subscribe("permission.grant", self._handle_permission_grant)
            self.core_bridge.subscribe("permission.deny", self._handle_permission_deny)
            self.core_bridge.subscribe("permission.ignore", self._handle_permission_ignore)
            
            # 检查TUI服务连接状态
            if self.tui_service:
                logger.debug("TUI服务已连接")
            else:
                logger.warning("TUI服务未连接，权限请求将显示在控制台")
            
            self.is_running = True
            logger.info("权限服务启动完成")
            return True
            
        except Exception as e:
            logger.error(f"启动权限服务时出错: {str(e)}", exc_info=True)
            return False
            
    async def _load_permission_rules(self):
        """加载权限规则"""
        try:
            rules_path = Path("config") / "permissions" / "permission_rules.yaml"
            if rules_path.exists():
                with open(rules_path, 'r', encoding='utf-8') as f:
                    self.permission_rules = yaml.safe_load(f)
                logger.debug(f"加载权限规则: {len(self.permission_rules.get('rules', {}))} 条规则")
            else:
                logger.warning("权限规则文件不存在，使用默认规则")
                self.permission_rules = {
                    "rules": {},
                    "default_policy": "ask"
                }
        except Exception as e:
            logger.error(f"加载权限规则时出错: {str(e)}", exc_info=True)
            self.permission_rules = {
                "rules": {},
                "default_policy": "ask"
            }
        
    async def _load_persisted_data(self):
        """加载持久化的权限数据"""
        try:
            # 如果配置文件不存在，创建空的配置文件
            if not self.granted_file.exists():
                await self._save_granted_permissions()
                logger.info("创建空的已授予权限文件")
            
            if not self.pending_file.exists():
                await self._save_pending_requests()
                logger.info("创建空的待处理请求文件")
            
            if not self.plugin_status_file.exists():
                await self._save_plugin_status()
                logger.info("创建空的插件状态文件")
            
            # 加载已授予权限
            if self.granted_file.exists():
                with open(self.granted_file, 'r', encoding='utf-8') as f:
                    granted_data = json.load(f)
                    # 将列表转换回集合
                    for plugin, permissions in granted_data.items():
                        self.granted_permissions[plugin] = set(permissions)
                logger.debug(f"加载已授予权限: {len(self.granted_permissions)} 个插件")
            
            # 加载待处理请求
            if self.pending_file.exists():
                with open(self.pending_file, 'r', encoding='utf-8') as f:
                    self.pending_requests = json.load(f)
                logger.debug(f"加载待处理请求: {len(self.pending_requests)} 个")
            
            # 加载插件状态
            if self.plugin_status_file.exists():
                with open(self.plugin_status_file, 'r', encoding='utf-8') as f:
                    self.plugin_status = json.load(f)
                logger.debug(f"加载插件状态: {len(self.plugin_status)} 个插件")
            
            logger.info("权限持久化数据加载完成")
            
        except Exception as e:
            logger.error(f"加载持久化权限数据时出错: {str(e)}", exc_info=True)
   
    async def _save_granted_permissions(self):
        """保存已授予权限到文件"""
        try:
            # 将集合转换为列表以便JSON序列化
            granted_data = {}
            for plugin, permissions in self.granted_permissions.items():
                granted_data[plugin] = list(permissions)
            
            with open(self.granted_file, 'w', encoding='utf-8') as f:
                json.dump(granted_data, f, ensure_ascii=False, indent=2)
            
            logger.debug(f"已授予权限已保存: {len(granted_data)} 个插件")
            
        except Exception as e:
            logger.error(f"保存已授予权限时出错: {str(e)}", exc_info=True)
    
    async def _save_pending_requests(self):
        """保存待处理请求到文件"""
        try:
            with open(self.pending_file, 'w', encoding='utf-8') as f:
                json.dump(self.pending_requests, f, ensure_ascii=False, indent=2)
            
            logger.debug(f"待处理请求已保存: {len(self.pending_requests)} 个")
            
        except Exception as e:
            logger.error(f"保存待处理请求时出错: {str(e)}", exc_info=True)
    
    async def _save_plugin_status(self):
        """保存插件状态到文件"""
        try:
            with open(self.plugin_status_file, 'w', encoding='utf-8') as f:
                json.dump(self.plugin_status, f, ensure_ascii=False, indent=2)
            
            logger.debug(f"插件状态已保存: {len(self.plugin_status)} 个插件")
            
        except Exception as e:
            logger.error(f"保存插件状态时出错: {str(e)}", exc_info=True)
    
    async def _save_all_data(self):
        """保存所有权限数据"""
        try:
            await asyncio.gather(
                self._save_granted_permissions(),
                self._save_pending_requests(),
                self._save_plugin_status()
            )
            logger.debug("所有权限数据已保存")
        except Exception as e:
            logger.error(f"保存权限数据时出错: {str(e)}", exc_info=True)
        
    def _handle_permission_request(self, message: Dict):
        """处理权限请求"""
        try:
            # 从 message 的 data 字段中获取插件名称 
            data = message.get('data', {}) 
            plugin_name = data.get('plugin_name') 
            logger.debug(f"传递的消息原文：{message}")
            logger.debug(f"传递的插件名：{plugin_name}")
            requested_permissions = data.get('permissions', [])
            request_id = str(uuid.uuid4())[:8]  # 简短的请求ID
            
            # 加强插件名称验证
            if not plugin_name or plugin_name == 'None' or plugin_name.strip() == '':
                logger.error(f"无效的插件名称: {repr(plugin_name)}")
                logger.debug(f"完整权限请求消息: {message}")
                return
            
            # 验证插件名称格式
            if not self._is_valid_plugin_name(plugin_name):
                logger.error(f"插件名称格式无效: {plugin_name}")
                return
            
            # 验证权限列表
            if not requested_permissions or not isinstance(requested_permissions, list):
                logger.warning(f"插件 {plugin_name} 请求的权限列表为空或格式错误")
                requested_permissions = []  # 确保是列表
            
            # 存储待处理请求
            self.pending_requests[request_id] = {
                'plugin_name': plugin_name,
                'permissions': requested_permissions,
                'timestamp': asyncio.get_event_loop().time()
            }
            
            # 更新插件状态
            self.plugin_status[plugin_name] = "pending"
            
            logger.debug(f"处理权限请求: {plugin_name} -> {len(requested_permissions)} 个权限")
            
            # 保存数据
            asyncio.create_task(self._save_pending_requests())
            asyncio.create_task(self._save_plugin_status())
            
            # 显示用户友好的权限请求界面
            asyncio.create_task(self._delayed_permission_ui(request_id, plugin_name, requested_permissions))
        
        except Exception as e:
            logger.error(f"处理权限请求时出错: {str(e)}", exc_info=True)

    def _is_valid_plugin_name(self, plugin_name: str) -> bool:
        """验证插件名称是否有效"""
        try:
            if not plugin_name or not isinstance(plugin_name, str):
                return False
            
            # 基本格式检查
            if plugin_name.strip() == '':
                return False
            
            # 检查常见无效值
            invalid_values = ['None', 'null', 'undefined', '']
            if plugin_name in invalid_values:
                return False
            
            # 检查长度限制
            if len(plugin_name) > 100:
                return False
            
            # 检查字符有效性（允许字母、数字、下划线、点、连字符）
            import re
            if not re.match(r'^[a-zA-Z0-9_\.\-]+$', plugin_name):
                return False
            
            return True
            
        except Exception as e:
            logger.error(f"验证插件名称时出错: {str(e)}")
            return False

    async def _delayed_permission_ui(self, request_id: str, plugin_name: str, permissions: List[str]):
        """延迟显示权限请求UI，等待TUI就绪"""
        try:
            # 等待TUI服务就绪
            tui_ready = await self.wait_for_tui_ready()
            
            if tui_ready:
                logger.debug(f"TUI已就绪，显示权限请求: {plugin_name}")
                await self._show_permission_request_ui(request_id, plugin_name, permissions)
            else:
                # TUI未就绪，使用回退显示
                logger.warning(f"TUI未就绪，使用控制台显示权限请求: {plugin_name}")
                message = f"🔐 插件 {plugin_name} 请求 {len(permissions)} 个权限 (请求ID: {request_id})"
                self._fallback_permission_display(request_id, plugin_name, message)
                
        except Exception as e:
            logger.error(f"延迟显示权限UI时出错: {str(e)}", exc_info=True)
        
    async def _show_permission_request_ui(self, request_id: str, plugin_name: str, permissions: List[str]):
        """显示权限请求用户界面 - 优化显示"""
        try:
            # 创建更友好的权限描述
            permission_descriptions = {
                "plugin.example.read": "📖 读取示例插件数据",
                "plugin.example.write": "✏️ 写入示例插件数据", 
                "plugin.example.execute": "⚡ 执行示例插件操作",
                "framework.event.subscribe": "📡 订阅框架事件",
                "framework.command.execute": "⌨️ 执行框架命令"
            }
            
            # 构建权限列表显示
            permission_list = []
            for perm in permissions:
                desc = permission_descriptions.get(perm, f"🔧 {perm}")
                permission_list.append(f"  ✅ {desc}")
            
            permission_display = "\n".join(permission_list) if permission_list else "  无具体权限请求"
            
            # 显示权限请求界面 - 使用更简洁的格式
            messages = [
                f"🔐 **插件权限请求**",
                f"",
                f"**插件**: {plugin_name}",
                f"**请求权限**:",
                permission_display,
                f"",
                f"**操作选项**:",
                f"  🟢  pmallow {request_id}  - 同意所有权限",
                f"  🟡  pmallow {request_id} read,write  - 仅同意部分权限",
                f"  🔴  pmdeny {request_id}  - 拒绝所有权限",
                f"  ⏸️  ignore {request_id}  - 暂时忽略",
                f"",
                f"**快捷命令**:",
                f"   pmallow all  - 同意所有待处理请求",
                f"   pmdeny all  - 拒绝所有待处理请求"
                f"   pmignore all  - 忽略所有待处理请求"
        ]
            
            # 逐行发送消息，确保每行都能正确显示
            if self.tui_service and hasattr(self.tui_service, 'show_message'):
                for line in messages:
                    if line.strip():  # 忽略空行
                        self.tui_service.show_message(line, "warning", persistent=True)
                        await asyncio.sleep(0.1)  # 小延迟确保消息顺序
            else:
                # TUI服务不可用，使用控制台输出
                self._fallback_permission_display(request_id, plugin_name, message)
                
        except Exception as e:
            logger.error(f"显示权限请求界面时出错: {str(e)}", exc_info=True)
    
    def _fallback_permission_display(self, request_id: str, plugin_name: str, message: str):
        """回退到控制台显示权限请求"""
        try:
            print("\n" + "="*60)
            print(message)
            print("="*60)
            print("🐱 请输入命令处理权限请求:")
            logger.info(f"权限请求已显示在控制台: {plugin_name} -> {request_id}")
        except Exception as e:
            logger.error(f"回退显示权限请求时出错: {str(e)}")
    
    def _handle_permission_grant(self, message: Dict):
        """处理权限授予"""
        try:
            request_id = message.get('request_id')
            granted_permissions = message.get('permissions', [])
            
            if request_id in self.pending_requests:
                request = self.pending_requests[request_id]
                plugin_name = request['plugin_name']
                
                # 如果未指定具体权限，授予所有请求的权限
                if not granted_permissions:
                    granted_permissions = request['permissions']
                
                # 授予权限
                asyncio.create_task(self.grant_permissions(plugin_name, granted_permissions))
                
                # 更新插件状态
                self.plugin_status[plugin_name] = "granted"
                
                # 移除待处理请求
                del self.pending_requests[request_id]
                
                # 显示成功消息
                success_msg = f"✅ 已为插件 '{plugin_name}' 授予 {len(granted_permissions)} 个权限"
                if self.tui_service and hasattr(self.tui_service, 'show_message'):
                    self.tui_service.show_message(success_msg, "success")
                else:
                    print(f"🐱 {success_msg}")
                
                logger.info(f"权限授予完成: {plugin_name} -> {granted_permissions}")
                
        except Exception as e:
            logger.error(f"处理权限授予时出错: {str(e)}", exc_info=True)
    
    def _handle_permission_deny(self, message: Dict):
        """处理权限拒绝"""
        try:
            request_id = message.get('request_id')
            
            if request_id in self.pending_requests:
                request = self.pending_requests[request_id]
                plugin_name = request['plugin_name']
                
                # 更新插件状态
                self.plugin_status[plugin_name] = "denied"
                
                # 移除待处理请求
                del self.pending_requests[request_id]
                
                # 保存数据
                asyncio.create_task(self._save_pending_requests())
                asyncio.create_task(self._save_plugin_status())
                
                # 显示拒绝消息
                deny_msg = f"❌ 已拒绝插件 '{plugin_name}' 的权限请求"
                if self.tui_service and hasattr(self.tui_service, 'show_message'):
                    self.tui_service.show_message(deny_msg, "error")
                else:
                    print(f"🐱 {deny_msg}")
                
                logger.info(f"权限拒绝完成: {plugin_name}")
                
        except Exception as e:
            logger.error(f"处理权限拒绝时出错: {str(e)}", exc_info=True)

    def _handle_permission_ignore(self, message: Dict):
        """处理权限忽略"""
        try:
            request_id = message.get('request_id')
            
            if request_id in self.pending_requests:
                request = self.pending_requests[request_id]
                plugin_name = request['plugin_name']
                
                # 更新插件状态
                self.plugin_status[plugin_name] = "ignored"
                
                # 移除待处理请求
                del self.pending_requests[request_id]
                
                # 保存数据
                asyncio.create_task(self._save_pending_requests())
                asyncio.create_task(self._save_plugin_status())
                
                # 显示忽略消息
                ignore_msg = f"⏸️ 已暂时忽略插件 '{plugin_name}' 的权限请求"
                if self.tui_service and hasattr(self.tui_service, 'show_message'):
                    self.tui_service.show_message(ignore_msg, "info")
                else:
                    print(f"🐱 {ignore_msg}")
                
                logger.info(f"权限请求被忽略: {plugin_name}")
                
        except Exception as e:
            logger.error(f"处理权限忽略时出错: {str(e)}", exc_info=True)

    async def grant_permissions(self, plugin_name: str, permissions: List[str]):
        """授予权限"""
        try:
            if plugin_name not in self.granted_permissions:
                self.granted_permissions[plugin_name] = set()
            
            for permission in permissions:
                self.granted_permissions[plugin_name].add(permission)
            
            # 更新插件状态
            self.plugin_status[plugin_name] = "granted"
            
            logger.debug(f"授予权限: {plugin_name} -> {permissions}")
            
            # 保存数据
            await asyncio.gather(
                self._save_granted_permissions(),
                self._save_plugin_status()
            )
            
            # 通知插件权限已授予
            await self.core_bridge.publish("permission.granted", {
                'plugin_name': plugin_name,
                'permissions': permissions
            })
            
        except Exception as e:
            logger.error(f"授予权限时出错: {str(e)}", exc_info=True)
            raise

    def has_permission(self, plugin_name: str, permission: str) -> bool:
        """检查是否具有权限"""
        try:
            # 检查显式授予的权限
            if plugin_name in self.granted_permissions:
                if permission in self.granted_permissions[plugin_name]:
                    return True
            
            # 检查权限规则
            rule_key = f"{plugin_name}.{permission}"
            if rule_key in self.permission_rules.get('rules', {}):
                return self.permission_rules['rules'][rule_key] == 'allow'
            
            # 默认策略
            default_policy = self.permission_rules.get('default_policy', 'ask')
            return default_policy == 'allow'
            
        except Exception as e:
            logger.error(f"检查权限时出错: {str(e)}", exc_info=True)
            return False
    
    async def request_permissions(self, plugin_name: str, permissions: List[str]) -> bool:
        """请求权限 - 非阻塞版本"""
        try:
            logger.debug(f"权限请求: {plugin_name} -> {permissions}")
            
            # 首先检查是否已经有所有权限
            if all(self.has_permission(plugin_name, perm) for perm in permissions):
                logger.debug(f"插件 {plugin_name} 已有所有请求的权限")
                return True
                
            # 标记插件为等待权限状态
            self.plugin_status[plugin_name] = "pending"
                        
            # 发布权限请求事件（非阻塞）
            await self.core_bridge.publish("permission.request", {
                'plugin_name': plugin_name,
                'permissions': permissions
            })
            
            # 立即返回，不等待用户响应
            # 插件将在权限被授予后通过事件机制得到通知
            logger.debug(f"权限请求已发送，等待用户响应: {plugin_name}")
            return True  # 立即返回True，让插件继续加载
            
        except Exception as e:
            logger.error(f"请求权限时出错: {str(e)}", exc_info=True)
            return True  # 出错时也返回True，避免阻塞插件加载
    
    def get_pending_requests(self) -> Dict[str, Dict]:
        """获取待处理请求"""
        return self.pending_requests.copy()
    
    async def process_permission_command(self, command: str, args: List[str]) -> str:
        """处理权限相关命令"""
        try:
            if command == "pmallow":
                if not args:
                    return "❌ 请指定请求ID，如: pmallow abc123"
                
                request_id = args[0]
                
                if request_id == "all":
                    # 同意所有待处理请求
                    count = len(self.pending_requests)
                    for rid in list(self.pending_requests.keys()):
                        self._handle_permission_grant({'request_id': rid})
                    return f"✅ 已同意所有 {count} 个待处理权限请求"
                
                # 检查特定权限
                specific_permissions = []
                if len(args) > 1:
                    specific_permissions = args[1].split(',')
                
                self._handle_permission_grant({
                    'request_id': request_id,
                    'permissions': specific_permissions
                })
                return f"✅ 已处理权限请求 {request_id}"
                
            elif command == "pmdeny":
                if not args:
                    return "❌ 请指定请求ID，如: pmdeny abc123"
                
                request_id = args[0]
                
                if request_id == "all":
                    # 拒绝所有待处理请求
                    count = len(self.pending_requests)
                    for rid in list(self.pending_requests.keys()):
                        await self._handle_permission_deny({'request_id': rid})
                    return f"❌ 已拒绝所有 {count} 个待处理权限请求"
                
                await self._handle_permission_deny({'request_id': request_id})
                return f"❌ 已拒绝权限请求 {request_id}"

            elif command == "pmignore":
                if not args:
                    return "❌ 请指定请求ID，如: pmignore abc123"
                
                request_id = args[0]
                
                if request_id == "all":
                    # 忽略所有待处理请求
                    count = len(self.pending_requests)
                    for rid in list(self.pending_requests.keys()):
                        await self._handle_permission_ignore({'request_id': rid})
                    return f"⏸️ 已忽略所有 {count} 个待处理权限请求"
                
                await self._handle_permission_ignore({'request_id': request_id})
                return f"⏸️ 已忽略权限请求 {request_id}"
                            
            elif command == "permissions":
                # 显示当前权限状态
                return await self._show_permission_status()
                
            elif command == "pm_plugin_status":
                # 显示插件状态
                return await self._show_plugin_status()
                
            elif command == "pmpending" or command == "pmrequests":
                # 查询待授权权限请求列表
                return await self._show_pending_requests()
                
            elif command == "pmhelp":
                # 显示权限命令帮助
                return self._show_permission_help()
                
            elif command == "pmfix":
                # 修复权限状态
                return await self._fix_permission_status()
                
            elif command == "pmclean":
                # 清理权限数据
                return await self._clean_permission_data(args)
                
            elif command == "pmbackup":
                # 备份权限数据
                return await self._backup_permission_data()
                
            elif command == "pmtest":
                # 测试权限配置文件
                return await self._test_permission_config()
                            
            else:
                return f"❌ 未知权限命令: {command}\n💡 输入 'pmhelp' 查看可用命令"               
                
        except Exception as e:
            logger.error(f"处理权限命令时出错: {str(e)}", exc_info=True)
            return f"❌ 处理命令时出错: {str(e)}"

    async def _test_permission_config(self) -> str:
        """测试权限配置文件"""
        try:
            logger.debug("开始权限配置文件测试")
            result = ["🔧 **权限配置文件测试**"]
            result.append("=" * 50)
            
            # 测试配置文件路径
            result.append("📁 **配置文件路径**:")
            result.append(f"  配置目录: {self.config_dir}")
            result.append(f"  已授予权限: {self.granted_file}")
            result.append(f"  待处理请求: {self.pending_file}")
            result.append(f"  插件状态: {self.plugin_status_file}")
            
            # 测试文件存在性
            result.append("\n✅ **文件存在性检查**:")
            config_files = [
                ("配置目录", self.config_dir, self.config_dir.exists()),
                ("已授予权限", self.granted_file, self.granted_file.exists()),
                ("待处理请求", self.pending_file, self.pending_file.exists()),
                ("插件状态", self.plugin_status_file, self.plugin_status_file.exists())
            ]
            
            for name, path, exists in config_files:
                status = "✅ 存在" if exists else "❌ 不存在"
                result.append(f"  {name}: {status}")
            
            # 测试写入权限
            result.append("\n✏️ **写入权限测试**:")
            try:
                test_data = {"test": "test_data", "timestamp": asyncio.get_event_loop().time()}
                with open(self.config_dir / "test_write.json", 'w', encoding='utf-8') as f:
                    json.dump(test_data, f, ensure_ascii=False, indent=2)
                
                # 读取测试
                with open(self.config_dir / "test_write.json", 'r', encoding='utf-8') as f:
                    read_data = json.load(f)
                
                # 清理测试文件
                (self.config_dir / "test_write.json").unlink(missing_ok=True)
                
                result.append("  ✅ 读写测试: 成功")
            except Exception as e:
                result.append(f"  ❌ 读写测试: 失败 - {str(e)}")
            
            # 显示当前数据状态
            result.append("\n📊 **当前数据状态**:")
            result.append(f"  已授予权限: {len(self.granted_permissions)} 个插件")
            result.append(f"  待处理请求: {len(self.pending_requests)} 个")
            result.append(f"  插件状态: {len(self.plugin_status)} 个")
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"测试权限配置时出错: {str(e)}", exc_info=True)
            return f"❌ 测试权限配置时出错: {str(e)}"
    
    async def _clean_permission_data(self, args: List[str]) -> str:
        """清理权限数据"""
        try:
            if not args:
                return "❌ 请指定清理类型\n💡 可用选项: expired, all, plugin <插件名>"
            
            clean_type = args[0].lower()
            result = []
            
            if clean_type == "expired":
                # 清理过期请求（超过24小时）
                current_time = asyncio.get_event_loop().time()
                expired_count = 0
                
                for request_id, request in list(self.pending_requests.items()):
                    if current_time - request.get('timestamp', 0) > 86400:  # 24小时
                        plugin_name = request['plugin_name']
                        del self.pending_requests[request_id]
                        expired_count += 1
                        result.append(f"🗑️ 清理过期请求: {request_id} ({plugin_name})")
                
                if expired_count > 0:
                    await self._save_pending_requests()
                    result.insert(0, f"✅ 已清理 {expired_count} 个过期权限请求")
                else:
                    result.append("✅ 没有发现过期权限请求")
                    
            elif clean_type == "all":
                # 清理所有数据
                pending_count = len(self.pending_requests)
                granted_count = len(self.granted_permissions)
                status_count = len(self.plugin_status)
                
                self.pending_requests.clear()
                self.granted_permissions.clear()
                self.plugin_status.clear()
                
                await self._save_all_data()
                
                result = [
                    f"✅ 已清理所有权限数据:",
                    f"  🗑️ 待处理请求: {pending_count} 个",
                    f"  🗑️ 已授予权限: {granted_count} 个插件", 
                    f"  🗑️ 插件状态: {status_count} 个"
                ]
                
            elif clean_type == "plugin" and len(args) > 1:
                # 清理特定插件的数据
                plugin_name = args[1]
                cleaned_items = []
                
                # 清理待处理请求
                for request_id, request in list(self.pending_requests.items()):
                    if request['plugin_name'] == plugin_name:
                        del self.pending_requests[request_id]
                        cleaned_items.append(f"待处理请求: {request_id}")
                
                # 清理已授予权限
                if plugin_name in self.granted_permissions:
                    del self.granted_permissions[plugin_name]
                    cleaned_items.append("已授予权限")
                
                # 清理插件状态
                if plugin_name in self.plugin_status:
                    del self.plugin_status[plugin_name]
                    cleaned_items.append("插件状态")
                
                if cleaned_items:
                    await self._save_all_data()
                    result = [f"✅ 已清理插件 '{plugin_name}' 的权限数据:"] + cleaned_items
                else:
                    result = [f"ℹ️ 未找到插件 '{plugin_name}' 的权限数据"]
                    
            else:
                return "❌ 无效的清理类型\n💡 可用选项: expired, all, plugin <插件名>"
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"清理权限数据时出错: {str(e)}", exc_info=True)
            return f"❌ 清理权限数据时出错: {str(e)}"

    async def _show_permission_status(self) -> str:
        """显示当前权限状态"""
        try:
            if not self.pending_requests and not self.granted_permissions:
                return "📋 暂无权限请求和授予记录"
            
            result = ["📋 **权限状态**"]
            
            if self.pending_requests:
                result.append("\n🟡 **待处理请求**:")
                for rid, req in self.pending_requests.items():
                    result.append(f"  {rid}: {req['plugin_name']} -> {len(req['permissions'])} 个权限")
            
            if self.granted_permissions:
                result.append("\n🟢 **已授予权限**:")
                for plugin, perms in self.granted_permissions.items():
                    result.append(f"  {plugin}: {len(perms)} 个权限")
            
            if self.plugin_status:
                result.append("\n🔵 **插件状态**:")
                for plugin, status in self.plugin_status.items():
                    status_icon = {
                        "granted": "✅",
                        "pending": "🟡", 
                        "denied": "❌",
                        "ignored": "⏸️",
                        "error": "⚠️"
                    }.get(status, "🔵")
                    result.append(f"  {status_icon} {plugin}: {status}")
            
            # 添加配置文件状态
            result.append("\n📁 **配置文件状态**:")
            config_files = [
                ("已授予权限", self.granted_file),
                ("待处理请求", self.pending_file),
                ("插件状态", self.plugin_status_file)
            ]
            
            for name, file_path in config_files:
                if file_path.exists():
                    result.append(f"  ✅ {name}: 存在")
                else:
                    result.append(f"  ❌ {name}: 不存在")
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"显示权限状态时出错: {str(e)}", exc_info=True)
            return f"❌ 显示权限状态时出错: {str(e)}"

    async def _show_plugin_status(self) -> str:
        """显示插件状态"""
        try:
            if not self.plugin_status:
                return "📊 暂无插件状态信息"
            
            result = ["📊 **插件状态**"]
            for plugin, status in self.plugin_status.items():
                if status == "granted":
                    result.append(f"  ✅ {plugin}: 权限已授予")
                elif status == "pending":
                    result.append(f"  🟡 {plugin}: 等待权限授予")
                elif status == "denied":
                    result.append(f"  ❌ {plugin}: 权限被拒绝")
                elif status == "ignored":
                    result.append(f"  ⏸️ {plugin}: 权限请求被忽略")
                elif status == "error":
                    result.append(f"  ⚠️ {plugin}: 权限错误")
                else:
                    result.append(f"  🔵 {plugin}: {status}")
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"显示插件状态时出错: {str(e)}", exc_info=True)
            return f"❌ 显示插件状态时出错: {str(e)}"

    async def _fix_permission_status(self) -> str:
        """修复权限状态不一致问题"""
        try:
            fixes_applied = []
            
            # 检查插件状态与待处理请求的一致性
            for plugin, status in list(self.plugin_status.items()):
                # 如果插件状态是pending但没有对应的待处理请求
                if status == "pending":
                    has_pending_request = False
                    for request in self.pending_requests.values():
                        if request['plugin_name'] == plugin:
                            has_pending_request = True
                            break
                    
                    if not has_pending_request:
                        # 修复：将状态改为error
                        self.plugin_status[plugin] = "error"
                        fixes_applied.append(f"🟡 {plugin}: pending → error (无权限请求)")
            
            # 清理过期的待处理请求
            current_time = asyncio.get_event_loop().time()
            expired_requests = []
            for request_id, request in list(self.pending_requests.items()):
                # 假设请求超过1小时为过期
                if current_time - request.get('timestamp', 0) > 3600:
                    expired_requests.append(request_id)
            
            for request_id in expired_requests:
                plugin_name = self.pending_requests[request_id]['plugin_name']
                del self.pending_requests[request_id]
                fixes_applied.append(f"🗑️ 清理过期请求: {request_id} ({plugin_name})")
            
            # 保存修复后的数据
            if fixes_applied:
                await self._save_all_data()
                result = ["🔧 **权限状态修复完成**"]
                result.extend(fixes_applied)
                result.append(f"\n✅ 共应用 {len(fixes_applied)} 个修复")
            else:
                result = ["✅ **权限状态正常**", "未发现需要修复的问题"]
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"修复权限状态时出错: {str(e)}", exc_info=True)
            return f"❌ 修复权限状态时出错: {str(e)}"

    async def _backup_permission_data(self) -> str:
        """备份权限数据"""
        try:
            backup_dir = self.config_dir / "backups"
            backup_dir.mkdir(exist_ok=True)
            
            import datetime
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            
            # 备份文件路径
            granted_backup = backup_dir / f"granted_permissions_{timestamp}.json"
            pending_backup = backup_dir / f"pending_requests_{timestamp}.json"
            status_backup = backup_dir / f"plugin_status_{timestamp}.json"
            
            # 复制文件
            import shutil
            if self.granted_file.exists():
                shutil.copy2(self.granted_file, granted_backup)
            if self.pending_file.exists():
                shutil.copy2(self.pending_file, pending_backup)
            if self.plugin_status_file.exists():
                shutil.copy2(self.plugin_status_file, status_backup)
            
            return f"✅ 权限数据备份完成\n📁 备份位置: {backup_dir}\n⏰ 时间戳: {timestamp}"
            
        except Exception as e:
            logger.error(f"备份权限数据时出错: {str(e)}", exc_info=True)
            return f"❌ 备份权限数据时出错: {str(e)}"

    async def _show_pending_requests(self) -> str:
        """显示待授权权限请求列表 - 更新为pm前缀"""
        try:
            if not self.pending_requests:
                return "📭 暂无待处理的权限请求"
            
            result = ["🟡 **待授权权限请求列表**"]
            result.append("=" * 50)
            
            for request_id, request in self.pending_requests.items():
                plugin_name = request['plugin_name']
                permissions = request['permissions']
                
                # 创建友好的权限描述
                permission_descriptions = {
                    "plugin.example.read": "📖 读取示例插件数据",
                    "plugin.example.write": "✏️ 写入示例插件数据", 
                    "plugin.example.execute": "⚡ 执行示例插件操作",
                    "framework.event.subscribe": "📡 订阅框架事件",
                    "framework.command.execute": "⌨️ 执行框架命令"
                }
                
                # 构建权限列表
                permission_list = []
                for perm in permissions:
                    desc = permission_descriptions.get(perm, f"🔧 {perm}")
                    permission_list.append(f"    • {desc}")
                
                permission_display = "\n".join(permission_list)
                
                # 添加请求信息
                result.append(f"\n📦 **插件**: {plugin_name}")
                result.append(f"🆔 **请求ID**: {request_id}")
                result.append(f"🔐 **请求权限** ({len(permissions)} 个):")
                result.append(permission_display)
                
                # 添加交互指令 - 更新为pm前缀
                result.append(f"\n💡 **交互指令**:")
                result.append(f"  🟢 同意所有权限:  pmallow {request_id}")
                result.append(f"  🟡 同意部分权限:  pmallow {request_id} read,write")
                result.append(f"  🔴 拒绝所有权限:  pmdeny {request_id}")
                result.append(f"  ⏸️  暂时忽略:  pmignore {request_id}")
                
                result.append("-" * 50)
            
            # 添加快捷指令 - 更新为pm前缀
            result.append("\n🚀 **快捷指令**:")
            result.append("  🟢 同意所有请求:  pmallow pmall")
            result.append("  🔴 拒绝所有请求:  pmdeny all")
            result.append("  ⏸️  忽略所有请求:  pmignore all")
            result.append("  📋 查看权限状态:  permissions")
            result.append("  📊 查看插件状态:  pm_plugin_status")
            result.append("  ❓ 查看帮助:  pmhelp")
            
            return "\n".join(result)
            
        except Exception as e:
            logger.error(f"显示待处理请求时出错: {str(e)}", exc_info=True)
            return f"❌ 显示待处理请求时出错: {str(e)}"
    
    def _show_permission_help(self) -> str:
        """显示权限命令帮助 - 更新为pm前缀"""
        help_text = """
    🔐 **权限管理命令帮助 (pm前缀)**
    
    📋 **查询命令**:
       permissions  - 查看权限状态
       pmpending  或  pmrequests  - 查看待授权请求列表
       pm_plugin_status  - 查看插件权限状态
    
    🛠️ **操作命令**:
       pmallow <请求ID>  - 同意指定请求的所有权限
       pmallow <请求ID> <权限列表>  - 同意指定请求的部分权限
       pmdeny <请求ID>  - 拒绝指定请求的所有权限
      。pmignore <请求ID>  - 暂时忽略指定请求
    
    🚀 **快捷命令**:
       pmallow pmall  - 同意所有待处理请求
       pmdeny all  - 拒绝所有待处理请求
       pmignore all  - 忽略所有待处理请求
      
    🔧 **维护命令**:
       pmfix  - 修复权限状态不一致问题
       pmtest  - 测试权限配置文件
    
    📖 **示例**:
       pmallow abc123  - 同意请求ID为abc123的所有权限
       pmallow abc123 read,write  - 只同意abc123的读取和写入权限
       pmdeny abc123  - 拒绝abc123的所有权限
       pmignore abc123  - 暂时忽略abc123的请求
       pmpending  - 查看所有待处理的权限请求
    
    💡 **提示**:
      • 权限请求ID是自动生成的8位字符串
      • 使用  pmpending  命令查看所有待处理请求及其ID
      • 插件在获得权限前可能以受限模式运行
      • 权限管理命令都以 `pm` 为前缀，避免与其他命令冲突
    """
        return help_text.strip()
                    
    def is_tui_ready(self) -> bool:
        """检查TUI服务是否就绪"""
        try:
            return (self.tui_service is not None and 
                    hasattr(self.tui_service, 'show_message') and
                    hasattr(self.tui_service, 'tui_app') and
                    self.tui_service.tui_app is not None)
        except Exception as e:
            logger.debug(f"检查TUI状态时出错: {e}")
            return False
    
    async def wait_for_tui_ready(self, timeout: float = 10.0) -> bool:
        """等待TUI服务就绪"""
        try:
            start_time = asyncio.get_event_loop().time()
            while asyncio.get_event_loop().time() - start_time < timeout:
                if self.is_tui_ready():
                    logger.debug("TUI服务已就绪")
                    return True
                await asyncio.sleep(0.5)
            
            logger.warning(f"等待TUI服务就绪超时 ({timeout}秒)")
            return False
        except Exception as e:
            logger.error(f"等待TUI就绪时出错: {e}")
            return False
    
    def shutdown(self):
        """关闭权限服务"""
        try:
            logger.info("关闭权限服务")
            self.is_running = False
            self.pending_requests.clear()
            logger.debug("权限服务关闭完成")
        except Exception as e:
            logger.error(f"关闭权限服务时出错: {str(e)}", exc_info=True)