#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import inspect import re from typing import Dict, List, Any, Callable from pathlib import Path import importlib logger = logging.getLogger(__name__) class PluginUtils: """插件工具类""" @staticmethod def validate_plugin_structure(plugin_path: Path) -> bool: """验证插件结构""" try: logger.debug(f"验证插件结构: {plugin_path}") required_files = [ "__init__.py", "config.yaml", "permissions.yaml" ] # 检查必需文件 for file_name in required_files: if not (plugin_path / file_name).exists(): logger.error(f"插件缺少必需文件: {file_name}") return False # 检查主模块是否有Plugin类 try: spec = importlib.util.spec_from_file_location("plugin_module", plugin_path / "__init__.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) if not hasattr(module, 'Plugin'): logger.error("插件主模块缺少Plugin类") return False # 检查Plugin类是否有必要方法 plugin_class = module.Plugin required_methods = ['initialize', 'shutdown'] for method_name in required_methods: if not hasattr(plugin_class, method_name): logger.error(f"Plugin类缺少必要方法: {method_name}") return False logger.debug(f"插件结构验证通过: {plugin_path.name}") return True except Exception as e: logger.error(f"验证插件类时出错: {str(e)}", exc_info=True) return False except Exception as e: logger.error(f"验证插件结构时出错: {str(e)}", exc_info=True) return False @staticmethod def get_plugin_dependencies(plugin_path: Path) -> List[str]: """获取插件依赖""" try: config_file = plugin_path / "config.yaml" if not config_file.exists(): return [] import yaml with open(config_file, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) dependencies = config.get('dependencies', []) if isinstance(dependencies, list): logger.debug(f"获取插件依赖: {plugin_path.name} -> {dependencies}") return dependencies else: logger.warning(f"插件依赖格式错误: {plugin_path.name}") return [] except Exception as e: logger.error(f"获取插件依赖时出错: {str(e)}", exc_info=True) return [] @staticmethod def scan_plugin_methods(plugin_instance) -> Dict[str, List[str]]: """扫描插件方法""" try: logger.debug(f"扫描插件方法: {type(plugin_instance).__name__}") methods_info = { "public_methods": [], "private_methods": [], "async_methods": [], "event_handlers": [] } for name, method in inspect.getmembers(plugin_instance, predicate=inspect.ismethod): # 跳过特殊方法 if name.startswith('_') and not name.startswith('__'): methods_info["private_methods"].append(name) elif not name.startswith('_'): methods_info["public_methods"].append(name) # 检查是否为异步方法 if inspect.iscoroutinefunction(method): methods_info["async_methods"].append(name) # 检查是否为事件处理器 if name.startswith('handle_') or name.startswith('on_'): methods_info["event_handlers"].append(name) logger.debug(f"插件方法扫描完成: 公共{len(methods_info['public_methods'])}个, 私有{len(methods_info['private_methods'])}个") return methods_info except Exception as e: logger.error(f"扫描插件方法时出错: {str(e)}", exc_info=True) return {} @staticmethod def create_plugin_skeleton(plugin_name: str, plugin_path: Path) -> bool: """创建插件骨架""" try: logger.debug(f"创建插件骨架: {plugin_name} -> {plugin_path}") # 创建插件目录 plugin_path.mkdir(parents=True, exist_ok=True) # 创建主模块文件 init_content = '''#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio from typing import Dict, Any logger = logging.getLogger(__name__) class Plugin: """{plugin_name} 插件""" def __init__(self, plugin_name: str, config: Dict, bridge): self.plugin_name = plugin_name self.config = config self.bridge = bridge self.is_running = False logger.debug(f"插件初始化: {{plugin_name}}") async def initialize(self): """初始化插件""" try: logger.info(f"初始化插件: {{self.plugin_name}}") # 在这里注册事件处理器和命令 # 示例: self.bridge.subscribe_plugin(self.plugin_name, "event.name", self.handler) self.is_running = True logger.debug(f"插件初始化完成: {{self.plugin_name}}") except Exception as e: logger.error(f"初始化插件时出错: {{str(e)}}", exc_info=True) raise async def shutdown(self): """关闭插件""" try: logger.info(f"关闭插件: {{self.plugin_name}}") self.is_running = False # 清理资源 self.bridge.cleanup_plugin_subscriptions(self.plugin_name) logger.debug(f"插件关闭完成: {{self.plugin_name}}") except Exception as e: logger.error(f"关闭插件时出错: {{str(e)}}", exc_info=True) # 在这里添加你的插件方法 async def example_method(self, message: str) -> str: """示例方法""" try: logger.debug(f"插件方法调用: {{message}}") return f"插件响应: {{message}}" except Exception as e: logger.error(f"插件方法调用出错: {{str(e)}}", exc_info=True) raise '''.format(plugin_name=plugin_name) with open(plugin_path / "__init__.py", 'w', encoding='utf-8') as f: f.write(init_content) # 创建配置文件 config_content = f'''# {plugin_name} 插件配置 name: "{plugin_name}" version: "1.0.0" description: "{plugin_name} 插件描述" author: "插件作者" # 插件特定配置 settings: enabled: true auto_start: true log_level: "INFO" # 依赖配置 dependencies: [] ''' with open(plugin_path / "config.yaml", 'w', encoding='utf-8') as f: f.write(config_content) # 创建权限文件 permissions_content = f'''# {plugin_name} 插件权限申请 plugin_name: "{plugin_name}" permissions: - "plugin.{plugin_name}.read" - "plugin.{plugin_name}.write" # 权限说明 permission_descriptions: plugin.{plugin_name}.read: "读取{plugin_name}插件数据" plugin.{plugin_name}.write: "写入{plugin_name}插件数据" ''' with open(plugin_path / "permissions.yaml", 'w', encoding='utf-8') as f: f.write(permissions_content) logger.info(f"插件骨架创建完成: {plugin_name}") return True except Exception as e: logger.error(f"创建插件骨架时出错: {str(e)}", exc_info=True) return False @staticmethod def validate_plugin_permissions(plugin_path: Path, requested_permissions: List[str]) -> bool: """验证插件权限申请""" try: logger.debug(f"验证插件权限: {plugin_path.name}") # 检查权限格式 for permission in requested_permissions: if not isinstance(permission, str): logger.error(f"权限格式错误: {permission}") return False # 检查权限命名规范 if not re.match(r'^[a-z][a-z0-9_.]*$', permission): logger.error(f"权限命名不规范: {permission}") return False logger.debug(f"插件权限验证通过: {len(requested_permissions)} 个权限") return True except Exception as e: logger.error(f"验证插件权限时出错: {str(e)}", exc_info=True) return False