# SenSu 插件开发超详细指南 # 目录 - [SenSu 插件开发超详细指南](#sensu-插件开发超详细指南) - [一、插件系统架构深度解析](#一插件系统架构深度解析) - [二、插件开发完整方案](#二插件开发完整方案) - [三、插件生命周期管理](#三插件生命周期管理) - [四、插件开发最佳实践](#四插件开发最佳实践) - [4.1 错误处理最佳实践](#41-错误处理最佳实践) - [4.2 性能优化最佳实践](#42-性能优化最佳实践) - [4.3 安全最佳实践](#43-安全最佳实践) - [4.4 测试最佳实践](#44-测试最佳实践) - [五、插件发布与部署](#五插件发布与部署) - [5.1 插件打包](#51-插件打包) - [5.2 插件发布清单](#52-插件发布清单) - [5.3 持续集成配置](#53-持续集成配置) - [六、插件调试与故障排除](#六插件调试与故障排除) - [6.1 调试工具](#61-调试工具) - [6.2 故障排除指南](#62-故障排除指南) - [在插件配置中](#在插件配置中) - [查找插件相关日志](#查找插件相关日志) - [实时查看日志](#实时查看日志) - [七、插件开发检查清单](#七插件开发检查清单) - [7.1 开发前检查清单](#71-开发前检查清单) - [7.2 开发中检查清单](#72-开发中检查清单) - [7.3 测试检查清单](#73-测试检查清单) - [7.4 发布检查清单](#74-发布检查清单) - [八、文件管理器集成 (v0.6.0)](#八文件管理器集成-v060-新增) - [九、总结](#九总结) - [8.1 成功插件的特点](#81-成功插件的特点) - [8.2 持续改进](#82-持续改进) - [8.3 资源推荐](#83-资源推荐) --- ## 一、插件系统架构深度解析 ### 1.1 插件生命周期 ``` ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ 扫描插件 │────▶│ 权限申请 │────▶│ 实例化插件 │ └───────────────┘ └───────────────┘ └───────────────┘ │ │ │ ▼ ▼ ▼ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ 加载配置文件 │ │ 权限验证/用户 │ │ 注册命令/路由 │ └───────────────┘ └───────────────┘ └───────────────┘ │ │ │ └──────────────────────┴──────────────────────┘ │ ▼ ┌───────────────────┐ │ 插件就绪运行 │ └───────────────────┘ ``` ### 1.2 插件通信架构 ```mermaid graph TB subgraph "插件内部" A[插件主类] --> B[命令处理器] A --> C[网络处理器] A --> D[事件处理器] end subgraph "框架服务" E[PluginBridge] --> F[CoreBridge] F --> G[网络服务] F --> H[命令服务] F --> I[权限服务] end B --> H C --> G D --> E subgraph "外部接口" J[HTTP客户端] --> G K[WebSocket客户端] --> G L[TUI用户] --> H end ``` ## 二、插件开发完整方案 ### 2.1 环境准备 ```bash # 1. 克隆或下载框架 git clone cd SenSu-Alpha0.2 # 2. 安装依赖(建议使用虚拟环境) python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install -r requirements.txt # 3. 运行框架测试 python main.py ``` ### 2.2 创建新插件 #### 2.2.1 插件目录结构 ``` plugins/ └── my_awesome_plugin/ # 插件目录(建议使用小写和下划线) ├── __init__.py # 插件主模块(必需) ├── config.yaml # 插件配置文件(必需) ├── permissions.yaml # 权限申请文件(必需) ├── requirements.txt # 插件特定依赖(可选) ├── README.md # 插件说明文档(推荐) ├── utils/ # 插件内部工具(可选) │ ├── __init__.py │ └── helper.py ├── models/ # 数据模型(可选) │ └── data_model.py ├── services/ # 插件服务模块(可选) │ └── background_service.py └── static/ # 静态资源(可选) ├── css/ ├── js/ └── images/ ``` #### 2.2.2 插件命名规范 1. **目录名**:小写字母、数字、下划线,如 `my_plugin` 2. **插件类名**:`Plugin`(必须使用这个类名) 3. **命令名**:小写字母、数字、下划线,如 `my_command` 4. **权限名**:`plugin.<插件名>.<操作>`,如 `plugin.my_plugin.read` ### 2.3 配置文件详解 #### 2.3.1 config.yaml 完整示例 ```yaml # my_awesome_plugin/config.yaml # ========== 基础信息(必需)========== name: "PluginName" # 插件显示名称 version: "1.0.0" # 版本号(遵循语义化版本) description: "这是一个功能强大的示例插件,用于演示插件开发" author: "开发者名字 " license: "MIT" # 开源许可证 # ========== 插件配置 ========== settings: enabled: true # 是否启用 auto_start: true # 是否自动启动 log_level: "INFO" # 日志级别:DEBUG, INFO, WARNING, ERROR max_retry_count: 3 # 失败重试次数 health_check_interval: 60 # 健康检查间隔(秒) background_task_interval: 300 # 后台任务间隔(秒) # ========== 功能配置 ========== features: # 网络功能配置 network: enable_http: true # 启用HTTP接口 enable_websocket: true # 启用WebSocket enable_cors: true # 启用跨域支持 cors_origins: ["*"] # 允许的跨域来源 # 数据库配置(如果有) database: type: "sqlite" # sqlite, mysql, postgresql path: "data/my_plugin.db" # SQLite数据库路径 host: "localhost" # 数据库主机 port: 3306 # 数据库端口 name: "my_plugin_db" # 数据库名 user: "username" # 用户名 password: "password" # 密码(建议使用环境变量) # 缓存配置 cache: type: "memory" # memory, redis ttl: 3600 # 缓存时间(秒) max_size: 1000 # 最大缓存项数 # 安全配置 security: require_auth: true # 是否需要认证 token_expiry: 86400 # Token过期时间(秒) rate_limit: 100 # 每秒请求限制 blacklist_enabled: true # 启用黑名单 # ========== 业务配置 ========== business: # API配置 api: default_page_size: 20 # 默认分页大小 max_page_size: 100 # 最大分页大小 date_format: "%Y-%m-%d %H:%M:%S" # 日期格式 # 文件存储 storage: type: "local" # local, s3, minio path: "data/files" # 本地存储路径 max_file_size: 10485760 # 最大文件大小(10MB) allowed_extensions: # 允许的文件扩展名 - .txt - .json - .yaml - .csv # 通知配置 notification: email_enabled: false webhook_enabled: true webhook_url: "" # ========== 定时任务配置 ========== schedules: - name: "daily_cleanup" cron: "0 2 * * *" # 每天凌晨2点 task: "cleanup_old_data" enabled: true - name: "hourly_sync" cron: "0 * * * *" # 每小时 task: "sync_external_data" enabled: true # ========== 依赖配置 ========== dependencies: required: # 必需依赖 - requests>=2.25.0 - pydantic>=1.8.0 optional: # 可选依赖 - redis>=3.5.0 # 如果使用Redis缓存 - aiomysql>=0.1.0 # 如果使用MySQL system: # 系统依赖 - ffmpeg # 如果处理音视频 - imagemagick # 如果处理图片 # ========== 国际化配置 ========== i18n: default_language: "zh_CN" supported_languages: - zh_CN - en_US translation_files: "translations/" # ========== 调试配置 ========== debug: enable_debug_endpoints: false # 是否启用调试端点 log_requests: true # 是否记录请求日志 log_responses: false # 是否记录响应日志 profile_performance: false # 是否启用性能分析 ``` #### 2.3.2 配置加载和验证 ```python # 在插件中加载和验证配置 from pydantic import BaseModel, validator from typing import Optional, List import os class PluginConfig(BaseModel): """插件配置模型""" name: str version: str description: str author: str settings: dict features: dict @validator('name') def validate_name(cls, v): if len(v) < 2 or len(v) > 50: raise ValueError('插件名称长度必须在2-50字符之间') return v @validator('version') def validate_version(cls, v): import re if not re.match(r'^\d+\.\d+\.\d+$', v): raise ValueError('版本号格式必须为 X.Y.Z') return v # 使用示例 config_data = { ... } # 从config.yaml加载 validated_config = PluginConfig(**config_data) ``` ### 2.4 权限文件详解 #### 2.4.1 permissions.yaml 完整示例 ```yaml # my_awesome_plugin/permissions.yaml # ========== 基础信息 ========== plugin_name: "my_awesome_plugin" # 必须与目录名一致 plugin_version: "1.0.0" # ========== 权限申请列表 ========== permissions: # 框架基础权限 - "framework.status.read" # 读取框架状态 - "framework.event.subscribe" # 订阅框架事件 - "framework.command.execute" # 执行框架命令 # 插件自身权限 - "plugin.my_awesome_plugin.read" # 读取插件数据 - "plugin.my_awesome_plugin.write" # 写入插件数据 - "plugin.my_awesome_plugin.execute" # 执行插件操作 - "plugin.my_awesome_plugin.delete" # 删除插件数据 # 网络权限 - "plugin.my_awesome_plugin.network.access" # 访问网络 - "plugin.my_awesome_plugin.network.http" # HTTP服务 - "plugin.my_awesome_plugin.network.websocket" # WebSocket服务 # 文件系统权限 - "plugin.my_awesome_plugin.filesystem.read" # 读取文件 - "plugin.my_awesome_plugin.filesystem.write" # 写入文件 # 外部服务权限 - "plugin.my_awesome_plugin.external_api.access" # 访问外部API # 系统权限(谨慎申请) - "plugin.my_awesome_plugin.system.execute" # 执行系统命令 # 管理权限 - "plugin.my_awesome_plugin.admin" # 插件管理员权限 # ========== 权限分组说明 ========== permission_groups: basic: # 基础组 - "plugin.my_awesome_plugin.read" - "plugin.my_awesome_plugin.write" network: # 网络组 - "plugin.my_awesome_plugin.network.access" - "plugin.my_awesome_plugin.network.http" - "plugin.my_awesome_plugin.network.websocket" advanced: # 高级组(需要特别说明) - "plugin.my_awesome_plugin.system.execute" - "plugin.my_awesome_plugin.admin" # ========== 权限详细说明 ========== permission_descriptions: # 基础权限说明 framework.status.read: "读取框架运行状态和基本信息" framework.event.subscribe: "订阅框架事件通知" framework.command.execute: "在框架中执行命令" # 插件权限说明 plugin.my_awesome_plugin.read: "读取插件的配置和数据" plugin.my_awesome_plugin.write: "修改插件的配置和数据" plugin.my_awesome_plugin.execute: "执行插件提供的操作" plugin.my_awesome_plugin.delete: "删除插件创建的数据" # 网络权限说明 plugin.my_awesome_plugin.network.access: "允许插件访问网络服务" plugin.my_awesome_plugin.network.http: "提供HTTP API接口" plugin.my_awesome_plugin.network.websocket: "提供WebSocket实时通信" # 文件系统权限说明 plugin.my_awesome_plugin.filesystem.read: "读取插件目录下的文件" plugin.my_awesome_plugin.filesystem.write: "在插件目录下创建和修改文件" # 外部服务权限说明 plugin.my_awesome_plugin.external_api.access: "访问第三方API服务(如天气、翻译等)" # 系统权限说明(危险权限) plugin.my_awesome_plugin.system.execute: "⚠️ 执行系统级命令(可能影响系统安全)" plugin.my_awesome_plugin.admin: "⚡ 插件管理员权限,可执行所有插件操作" # ========== 权限风险评估 ========== permission_risk_levels: low_risk: # 低风险权限 - "framework.status.read" - "plugin.my_awesome_plugin.read" medium_risk: # 中风险权限 - "plugin.my_awesome_plugin.write" - "plugin.my_awesome_plugin.network.access" high_risk: # 高风险权限 - "plugin.my_awesome_plugin.system.execute" - "plugin.my_awesome_plugin.admin" # ========== 依赖权限说明 ========== permission_dependencies: # 某些权限需要其他权限的支持 plugin.my_awesome_plugin.network.http: requires: "plugin.my_awesome_plugin.network.access" plugin.my_awesome_plugin.network.websocket: requires: "plugin.my_awesome_plugin.network.access" plugin.my_awesome_plugin.admin: requires_all: # 需要所有以下权限 - "plugin.my_awesome_plugin.read" - "plugin.my_awesome_plugin.write" - "plugin.my_awesome_plugin.execute" - "plugin.my_awesome_plugin.delete" # ========== 权限使用场景示例 ========== usage_scenarios: - scenario: "数据查看" required_permissions: - "plugin.my_awesome_plugin.read" description: "用户只能查看数据,不能修改" - scenario: "数据管理" required_permissions: - "plugin.my_awesome_plugin.read" - "plugin.my_awesome_plugin.write" - "plugin.my_awesome_plugin.delete" description: "用户可以完全管理数据" - scenario: "API服务" required_permissions: - "plugin.my_awesome_plugin.network.access" - "plugin.my_awesome_plugin.network.http" description: "插件可以提供HTTP API服务" - scenario: "实时通信" required_permissions: - "plugin.my_awesome_plugin.network.access" - "plugin.my_awesome_plugin.network.websocket" description: "插件可以提供WebSocket实时通信" # ========== 插件启动模式 ========== startup_modes: # 权限不足时的启动模式 fallback_mode: enabled: true permissions_required: # 必需的最小权限集 - "framework.status.read" - "plugin.my_awesome_plugin.read" degraded_features: # 降级运行的功能 - "network_services" - "background_tasks" message: "插件将在受限模式下运行,部分功能不可用" # ========== 权限版本控制 ========== versioning: current_version: "1.0" deprecated_permissions: # 已废弃的权限 - "plugin.my_awesome_plugin.old_read" new_permissions: # 新增权限 - "plugin.my_awesome_plugin.enhanced_write" migration_guide: "从v0.9升级到v1.0,请重新申请权限" ``` #### 2.4.2 权限验证代码示例 ```python class PermissionValidator: """权限验证辅助类""" @staticmethod def validate_permission_structure(permissions: list) -> tuple[bool, str]: """验证权限列表结构""" if not permissions: return False, "权限列表不能为空" for perm in permissions: if not isinstance(perm, str): return False, f"权限必须是字符串: {perm}" # 检查格式:plugin.plugin_name.action if not perm.startswith("plugin.") and not perm.startswith("framework."): return False, f"权限格式错误: {perm}" # 检查长度 if len(perm) > 100: return False, f"权限名称过长: {perm}" return True, "验证通过" @staticmethod def group_permissions_by_risk(permissions: list) -> dict: """按风险等级分组权限""" risk_groups = { "low": [], "medium": [], "high": [] } risk_mapping = { "read": "low", "write": "medium", "delete": "medium", "execute": "high", "admin": "high", "system": "high" } for perm in permissions: risk = "medium" # 默认中风险 for keyword, level in risk_mapping.items(): if keyword in perm.lower(): risk = level break risk_groups[risk].append(perm) return risk_groups ``` ### 2.5 插件主类完整实现 #### 2.5.1 __init__.py 完整模板 ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 我的插件 - 插件主模块 版本: 1.0.0 作者: 开发者名字 描述: 这是一个功能完整的插件示例 """ import logging import asyncio import sys import os from pathlib import Path from typing import Dict, Any, List, Optional, Union from dataclasses import dataclass from datetime import datetime, timedelta import json import traceback # 导入框架装饰器 try: from core.plugin_command_decorator import plugin_command, command except ImportError: # 回退方案 - 本地定义装饰器 def plugin_command(name=None, description=None, permissions=None): def decorator(func): func._is_plugin_command = True func._command_name = name or func.__name__ func._command_description = description or func.__doc__ or f"命令: {func.__name__}" func._command_permissions = permissions or [] return func return decorator command = plugin_command # 导入网络桥接 try: from bridges.plugin_network_bridge import PluginNetworkBridge except ImportError: # 虚拟网络桥接类 class PluginNetworkBridge: def __init__(self, plugin_name, internet_service, plugin_bridge): self.plugin_name = plugin_name self._logger = logging.getLogger(f"{__name__}.NetworkBridge") self._logger.warning(f"网络桥接不可用,插件将以无网络模式运行") async def register_http_route(self, *args, **kwargs): self._logger.warning("网络功能不可用,跳过HTTP路由注册") async def register_websocket(self, *args, **kwargs): self._logger.warning("网络功能不可用,跳过WebSocket注册") async def broadcast_websocket(self, *args, **kwargs): self._logger.warning("网络功能不可用,无法广播消息") def get_network_info(self): return { 'plugin_name': self.plugin_name, 'registered_routes': [], 'websocket_handlers': [], 'base_url': '网络服务不可用' } async def setup_data_transfer(self, *args, **kwargs): self._logger.warning("网络功能不可用,跳过数据传输设置") # 插件内部模块 try: from .utils.helper import HelperClass from .models.data_model import DataModel except ImportError: # 如果内部模块不可用,创建虚拟类 HelperClass = type('HelperClass', (), {}) DataModel = type('DataModel', (), {}) # 日志记录器 logger = logging.getLogger(__name__) # 数据类定义 @dataclass class PluginStatus: """插件状态数据类""" is_running: bool = False start_time: Optional[datetime] = None uptime: Optional[timedelta] = None request_count: int = 0 error_count: int = 0 last_error: Optional[str] = None memory_usage: Optional[int] = None @dataclass class PluginMetrics: """插件指标数据类""" requests_per_second: float = 0.0 average_response_time: float = 0.0 active_connections: int = 0 cache_hit_rate: float = 0.0 queue_size: int = 0 class Plugin: """ 我的插件主类 功能特性: 1. 完整的HTTP API接口 2. WebSocket实时通信 3. 后台定时任务 4. 数据缓存机制 5. 健康检查系统 6. 完整的错误处理 7. 性能监控指标 使用方法: 1. 确保框架已安装并运行 2. 将此插件放入plugins目录 3. 重启框架或使用插件管理命令加载 """ # 类常量 PLUGIN_NAME = "my_awesome_plugin" PLUGIN_VERSION = "1.0.0" DEFAULT_CONFIG = { "enabled": True, "log_level": "INFO" } def __init__(self, plugin_name: str, config: Dict, bridge): """ 初始化插件 Args: plugin_name: 插件名称(框架传入) config: 插件配置(从config.yaml加载) bridge: PluginBridge实例,用于插件间通信 """ self.plugin_name = plugin_name self.original_config = config self.bridge = bridge self.service_manager = None # 配置处理 self.config = self._merge_configs(self.DEFAULT_CONFIG, config) # 网络桥接 self.network_bridge = None # 状态管理 self.status = PluginStatus() self.metrics = PluginMetrics() # 缓存系统 self.cache = {} self.cache_ttl = {} # 后台任务 self.background_tasks = [] self.task_handles = {} # 资源锁 self._lock = asyncio.Lock() self._resource_locks = {} # 内部服务 self.helper = HelperClass() self.data_model = DataModel() # 事件处理器映射 self.event_handlers = {} # WebSocket连接管理 self.websocket_connections = {} # API速率限制 self.rate_limiter = {} logger.info(f"插件初始化: {self.plugin_name} v{self.PLUGIN_VERSION}") def _merge_configs(self, default: Dict, override: Dict) -> Dict: """深度合并配置""" result = default.copy() for key, value in override.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = self._merge_configs(result[key], value) else: result[key] = value return result async def initialize(self): """ 初始化插件 - 核心入口点 执行顺序: 1. 基础初始化 2. 获取服务管理器 3. 设置网络功能 4. 注册事件处理器 5. 启动后台任务 6. 健康检查 """ try: logger.info(f"开始初始化插件: {self.plugin_name}") # 1. 记录启动时间 self.status.start_time = datetime.now() # 2. 获取服务管理器(如果可用) await self._get_service_manager() # 3. 初始化网络功能 await self._initialize_network() # 4. 注册事件处理器 await self._register_event_handlers() # 5. 启动后台任务 await self._start_background_tasks() # 6. 初始化缓存系统 await self._initialize_cache() # 7. 设置健康检查 await self._setup_health_check() # 8. 更新状态 self.status.is_running = True self.status.uptime = datetime.now() - self.status.start_time logger.info(f"✅ 插件初始化完成: {self.plugin_name}") logger.info(f" 版本: {self.PLUGIN_VERSION}") logger.info(f" 配置: {len(self.config)} 项") logger.info(f" 网络: {'可用' if self.network_bridge else '不可用'}") # 发送初始化完成事件 await self._send_initialization_event() return True except Exception as e: logger.error(f"❌ 插件初始化失败: {str(e)}") logger.error(traceback.format_exc()) # 尝试清理已初始化的资源 await self._emergency_cleanup() return False async def _get_service_manager(self): """安全获取服务管理器""" try: if hasattr(self.bridge, 'service_manager'): self.service_manager = self.bridge.service_manager logger.debug("服务管理器获取成功") else: logger.warning("服务管理器不可用,部分功能可能受限") except Exception as e: logger.warning(f"获取服务管理器时出错: {str(e)}") async def _initialize_network(self): """初始化网络功能""" try: # 获取网络服务 internet_service = None if self.service_manager: try: internet_service = self.service_manager.get_service("internet") except ValueError: logger.warning("网络服务未注册") # 创建网络桥接 if internet_service: self.network_bridge = PluginNetworkBridge( self.plugin_name, internet_service, self.bridge ) # 注册网络路由 await self._register_network_routes() logger.info(f"网络功能初始化完成,基础URL: {self.network_bridge.get_network_info()['base_url']}") else: logger.info("网络服务不可用,插件将以无网络模式运行") # 创建虚拟网络桥接 self.network_bridge = PluginNetworkBridge(self.plugin_name, None, self.bridge) except Exception as e: logger.error(f"初始化网络功能时出错: {str(e)}") raise async def _register_network_routes(self): """注册所有网络路由""" try: if not self.network_bridge: logger.warning("网络桥接不可用,跳过路由注册") return logger.info("开始注册网络路由...") # 1. 信息接口(公开) await self.network_bridge.register_http_route( "/api/info", self._handle_api_info, methods=["GET"], require_auth=False ) # 2. 健康检查接口(公开) await self.network_bridge.register_http_route( "/api/health", self._handle_api_health, methods=["GET"], require_auth=False ) # 3. 数据查询接口(需要认证) await self.network_bridge.register_http_route( "/api/data", self._handle_api_data, methods=["GET", "POST"], require_auth=True ) # 4. 文件上传接口(需要认证) await self.network_bridge.register_http_route( "/api/upload", self._handle_api_upload, methods=["POST"], require_auth=True ) # 5. 管理接口(需要管理员权限) await self.network_bridge.register_http_route( "/api/admin/status", self._handle_admin_status, methods=["GET"], require_auth=True ) # 6. WebSocket聊天接口 await self.network_bridge.register_websocket( "/ws/chat", self._handle_websocket_chat, require_auth=True ) # 7. WebSocket实时数据接口 await self.network_bridge.register_websocket( "/ws/data", self._handle_websocket_data, require_auth=True ) # 8. 设置跨端数据传输 await self.network_bridge.setup_data_transfer( self._handle_cross_platform_data ) logger.info(f"网络路由注册完成,共注册 {len(self._get_registered_routes())} 个路由") except Exception as e: logger.error(f"注册网络路由时出错: {str(e)}") raise def _get_registered_routes(self): """获取已注册的路由信息""" if not self.network_bridge: return [] info = self.network_bridge.get_network_info() return info.get('registered_routes', []) async def _register_event_handlers(self): """注册事件处理器""" try: # 定义事件处理器映射 self.event_handlers = { "framework.start": self._handle_framework_start, "framework.shutdown": self._handle_framework_shutdown, "plugin.load": self._handle_plugin_load, "plugin.unload": self._handle_plugin_unload, "permission.granted": self._handle_permission_granted, "permission.denied": self._handle_permission_denied, "network.data.receive": self._handle_network_data_receive, "user.login": self._handle_user_login, "user.logout": self._handle_user_logout, } # 注册事件处理器 for event_type, handler in self.event_handlers.items(): self.bridge.subscribe_plugin( self.plugin_name, f"event.{event_type}", handler ) logger.info(f"事件处理器注册完成,共 {len(self.event_handlers)} 个") except Exception as e: logger.error(f"注册事件处理器时出错: {str(e)}") async def _start_background_tasks(self): """启动后台任务""" try: config = self.config.get('schedules', []) for schedule in config: if schedule.get('enabled', True): task_name = schedule['name'] cron_expr = schedule['cron'] task_func = getattr(self, f"_task_{schedule['task']}", None) if task_func: # 创建后台任务 task = asyncio.create_task( self._schedule_task(task_name, cron_expr, task_func) ) self.background_tasks.append(task) self.task_handles[task_name] = task logger.info(f"后台任务启动: {task_name} ({cron_expr})") logger.info(f"后台任务启动完成,共 {len(self.background_tasks)} 个任务") except Exception as e: logger.error(f"启动后台任务时出错: {str(e)}") async def _schedule_task(self, name: str, cron_expr: str, task_func): """按Cron表达式调度任务""" from croniter import croniter import time base_time = time.time() cron = croniter(cron_expr, base_time) while self.status.is_running: try: # 计算下一次执行时间 next_time = cron.get_next(float) sleep_time = next_time - time.time() if sleep_time > 0: await asyncio.sleep(sleep_time) # 执行任务 logger.debug(f"执行定时任务: {name}") await task_func() except asyncio.CancelledError: logger.info(f"任务被取消: {name}") break except Exception as e: logger.error(f"任务执行出错 {name}: {str(e)}") await asyncio.sleep(60) # 出错后等待1分钟 async def _initialize_cache(self): """初始化缓存系统""" try: cache_config = self.config.get('cache', {}) if cache_config.get('type') == 'redis': # 初始化Redis缓存 import redis self.redis_client = redis.Redis( host=cache_config.get('host', 'localhost'), port=cache_config.get('port', 6379), db=cache_config.get('db', 0) ) logger.info("Redis缓存初始化完成") else: # 使用内存缓存 logger.info("内存缓存初始化完成") except Exception as e: logger.warning(f"缓存初始化失败,使用无缓存模式: {str(e)}") async def _setup_health_check(self): """设置健康检查""" try: # 创建健康检查任务 health_task = asyncio.create_task(self._health_check_loop()) self.background_tasks.append(health_task) logger.info("健康检查系统已启动") except Exception as e: logger.warning(f"健康检查设置失败: {str(e)}") async def _health_check_loop(self): """健康检查循环""" while self.status.is_running: try: await asyncio.sleep(60) # 每分钟检查一次 # 检查网络连接 network_healthy = await self._check_network_health() # 检查缓存 cache_healthy = await self._check_cache_health() # 检查后台任务 tasks_healthy = await self._check_tasks_health() # 记录健康状态 self.metrics.requests_per_second = self._calculate_rps() if not all([network_healthy, cache_healthy, tasks_healthy]): logger.warning("健康检查发现问题") except Exception as e: logger.error(f"健康检查出错: {str(e)}") async def _send_initialization_event(self): """发送初始化完成事件""" try: await self.bridge.publish_to_plugin( "framework", "event.plugin.initialized", { "plugin_name": self.plugin_name, "version": self.PLUGIN_VERSION, "timestamp": datetime.now().isoformat() } ) except Exception as e: logger.debug(f"发送初始化事件失败: {str(e)}") async def _emergency_cleanup(self): """紧急清理资源""" try: # 取消所有后台任务 for task in self.background_tasks: if not task.done(): task.cancel() # 清理缓存 self.cache.clear() logger.info("紧急清理完成") except Exception as e: logger.error(f"紧急清理时出错: {str(e)}") # ========== 网络处理器方法 ========== async def _handle_api_info(self, request): """处理API信息请求""" from aiohttp import web try: self.status.request_count += 1 info = { "plugin": { "name": self.plugin_name, "version": self.PLUGIN_VERSION, "description": self.config.get('description', ''), "author": self.config.get('author', ''), "status": "running" if self.status.is_running else "stopped" }, "system": { "start_time": self.status.start_time.isoformat() if self.status.start_time else None, "uptime": str(self.status.uptime) if self.status.uptime else None, "request_count": self.status.request_count, "error_count": self.status.error_count }, "network": self.network_bridge.get_network_info() if self.network_bridge else None, "timestamp": datetime.now().isoformat() } return web.json_response(info) except Exception as e: logger.error(f"处理API信息请求时出错: {str(e)}") return web.json_response( {"error": "服务器内部错误", "details": str(e)}, status=500 ) async def _handle_api_health(self, request): """处理健康检查请求""" from aiohttp import web try: # 检查各项健康指标 checks = { "plugin_running": self.status.is_running, "network_available": self.network_bridge is not None, "background_tasks": len([t for t in self.background_tasks if not t.done()]), "cache_available": len(self.cache) > 0 or hasattr(self, 'redis_client'), "last_error": self.status.last_error } # 计算总体状态 all_healthy = all([ checks["plugin_running"], checks["network_available"], checks["background_tasks"] > 0 ]) response = { "status": "healthy" if all_healthy else "unhealthy", "timestamp": datetime.now().isoformat(), "checks": checks, "metrics": { "requests_per_second": self.metrics.requests_per_second, "active_connections": len(self.websocket_connections), "cache_size": len(self.cache) } } status_code = 200 if all_healthy else 503 return web.json_response(response, status=status_code) except Exception as e: logger.error(f"处理健康检查请求时出错: {str(e)}") return web.json_response( {"status": "error", "error": str(e)}, status=500 ) async def _handle_api_data(self, request): """处理数据API请求""" from aiohttp import web try: # 检查速率限制 client_ip = request.remote if not await self._check_rate_limit(client_ip): return web.json_response( {"error": "请求过于频繁,请稍后再试"}, status=429 ) if request.method == "GET": # 查询数据 query_params = dict(request.query) data = await self._query_data(query_params) return web.json_response({ "success": True, "data": data, "count": len(data), "timestamp": datetime.now().isoformat() }) elif request.method == "POST": # 创建数据 data = await request.json() result = await self._create_data(data) return web.json_response({ "success": True, "id": result.get("id"), "message": "数据创建成功", "timestamp": datetime.now().isoformat() }, status=201) except json.JSONDecodeError: return web.json_response( {"error": "无效的JSON数据"}, status=400 ) except Exception as e: logger.error(f"处理数据API请求时出错: {str(e)}") return web.json_response( {"error": "服务器内部错误", "details": str(e)}, status=500 ) async def _handle_api_upload(self, request): """处理文件上传请求""" from aiohttp import web import aiofiles try: # 检查内容类型 if not request.content_type.startswith('multipart/form-data'): return web.json_response( {"error": "必须使用multipart/form-data格式"}, status=400 ) reader = await request.multipart() files = [] async for field in reader: if field.filename: # 保存文件 filename = field.filename filepath = Path("data/uploads") / self.plugin_name / filename filepath.parent.mkdir(parents=True, exist_ok=True) size = 0 async with aiofiles.open(filepath, 'wb') as f: while True: chunk = await field.read_chunk() if not chunk: break size += len(chunk) await f.write(chunk) files.append({ "filename": filename, "size": size, "path": str(filepath) }) return web.json_response({ "success": True, "files": files, "count": len(files), "timestamp": datetime.now().isoformat() }) except Exception as e: logger.error(f"处理文件上传时出错: {str(e)}") return web.json_response( {"error": "文件上传失败", "details": str(e)}, status=500 ) async def _handle_admin_status(self, request): """处理管理状态请求""" from aiohttp import web try: # 检查管理员权限 if not await self._check_admin_permission(request): return web.json_response( {"error": "需要管理员权限"}, status=403 ) status_info = { "plugin": { "name": self.plugin_name, "config": self.config, "status": self.status, "metrics": self.metrics }, "system": { "background_tasks": [ { "name": name, "running": not task.done(), "cancelled": task.cancelled() } for name, task in self.task_handles.items() ], "cache_info": { "size": len(self.cache), "keys": list(self.cache.keys())[:10] }, "websocket_connections": len(self.websocket_connections) }, "timestamp": datetime.now().isoformat() } return web.json_response(status_info) except Exception as e: logger.error(f"处理管理状态请求时出错: {str(e)}") return web.json_response( {"error": "服务器内部错误", "details": str(e)}, status=500 ) async def _handle_websocket_chat(self, ws, request): """处理WebSocket聊天""" from aiohttp import web try: # 获取用户信息 user = await self._get_user_from_request(request) if not user: await ws.close(code=1008, message="未认证") return # 记录连接 connection_id = f"{user['id']}_{id(ws)}" self.websocket_connections[connection_id] = { "ws": ws, "user": user, "connected_at": datetime.now() } logger.info(f"WebSocket聊天连接建立: {connection_id}") # 发送欢迎消息 await ws.send_str(json.dumps({ "type": "system", "message": f"欢迎 {user['username']} 进入聊天室", "timestamp": datetime.now().isoformat() })) # 广播用户上线消息 await self._broadcast_chat_message({ "type": "user_join", "user": user, "timestamp": datetime.now().isoformat() }) # 处理消息 async for msg in ws: if msg.type == web.WSMsgType.TEXT: try: data = json.loads(msg.data) # 处理不同类型的消息 if data.get('type') == 'message': # 广播聊天消息 message = { "type": "message", "from": user, "content": data.get('content', ''), "timestamp": datetime.now().isoformat() } await self._broadcast_chat_message(message) elif data.get('type') == 'typing': # 广播输入状态 await self._broadcast_chat_message({ "type": "typing", "user": user, "is_typing": data.get('is_typing', False), "timestamp": datetime.now().isoformat() }) except json.JSONDecodeError: logger.warning(f"收到无效的JSON消息: {msg.data}") elif msg.type == web.WSMsgType.ERROR: logger.error(f"WebSocket错误: {ws.exception()}") elif msg.type == web.WSMsgType.CLOSE: logger.info(f"WebSocket连接关闭: {connection_id}") except Exception as e: logger.error(f"WebSocket聊天处理出错: {str(e)}") finally: # 清理连接 if connection_id in self.websocket_connections: del self.websocket_connections[connection_id] # 广播用户离线消息 if 'user' in locals(): await self._broadcast_chat_message({ "type": "user_leave", "user": user, "timestamp": datetime.now().isoformat() }) async def _handle_websocket_data(self, ws, request): """处理WebSocket实时数据""" from aiohttp import web try: # 获取用户信息 user = await self._get_user_from_request(request) if not user: await ws.close(code=1008, message="未认证") return connection_id = f"data_{user['id']}_{id(ws)}" logger.info(f"WebSocket数据连接建立: {connection_id}") # 发送初始数据 await ws.send_str(json.dumps({ "type": "init", "data": await self._get_initial_data(), "timestamp": datetime.now().isoformat() })) # 定期发送更新 while not ws.closed: try: await asyncio.sleep(5) # 每5秒发送一次更新 if not ws.closed: await ws.send_str(json.dumps({ "type": "update", "data": await self._get_updated_data(), "timestamp": datetime.now().isoformat() })) except asyncio.CancelledError: break except Exception as e: logger.error(f"发送WebSocket数据更新时出错: {str(e)}") break except Exception as e: logger.error(f"WebSocket数据处理出错: {str(e)}") finally: logger.info(f"WebSocket数据连接关闭: {connection_id}") async def _handle_cross_platform_data(self, event_data): """处理跨端数据""" try: logger.info(f"收到跨端数据: {event_data.get('type')}") # 根据数据类型处理 data_type = event_data.get('type') if data_type == "sync_request": # 处理同步请求 await self._handle_sync_request(event_data) elif data_type == "notification": # 处理通知 await self._handle_notification(event_data) elif data_type == "command": # 处理远程命令 await self._handle_remote_command(event_data) # 广播到WebSocket if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "cross_platform", "source": event_data.get('source', 'unknown'), "data": event_data.get('data', {}), "timestamp": datetime.now().isoformat() }) except Exception as e: logger.error(f"处理跨端数据时出错: {str(e)}") # ========== 事件处理器方法 ========== async def _handle_framework_start(self, event_data): """处理框架启动事件""" try: logger.info(f"框架启动事件: {event_data}") # 发送欢迎消息 if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "system", "message": f"插件 {self.plugin_name} 已就绪,框架已启动", "timestamp": datetime.now().isoformat() }) except Exception as e: logger.error(f"处理框架启动事件时出错: {str(e)}") async def _handle_framework_shutdown(self, event_data): """处理框架关闭事件""" try: logger.info("收到框架关闭事件,开始清理...") # 通知所有连接 if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "system", "message": "框架正在关闭,请保存您的工作", "timestamp": datetime.now().isoformat() }) # 执行插件关闭 await self.shutdown() except Exception as e: logger.error(f"处理框架关闭事件时出错: {str(e)}") async def _handle_plugin_load(self, event_data): """处理插件加载事件""" try: loaded_plugin = event_data.get('plugin_name') logger.info(f"插件加载事件: {loaded_plugin}") # 如果是其他插件加载,可以建立连接或同步数据 if loaded_plugin != self.plugin_name: await self._sync_with_plugin(loaded_plugin) except Exception as e: logger.error(f"处理插件加载事件时出错: {str(e)}") async def _handle_plugin_unload(self, event_data): """处理插件卸载事件""" try: unloaded_plugin = event_data.get('plugin_name') logger.info(f"插件卸载事件: {unloaded_plugin}") # 清理与该插件相关的资源 await self._cleanup_plugin_resources(unloaded_plugin) except Exception as e: logger.error(f"处理插件卸载事件时出错: {str(e)}") async def _handle_permission_granted(self, event_data): """处理权限授予事件""" try: plugin_name = event_data.get('plugin_name') permissions = event_data.get('permissions', []) if plugin_name == self.plugin_name: logger.info(f"权限已授予: {permissions}") # 重新初始化需要权限的功能 await self._reinitialize_with_permissions(permissions) except Exception as e: logger.error(f"处理权限授予事件时出错: {str(e)}") async def _handle_permission_denied(self, event_data): """处理权限拒绝事件""" try: plugin_name = event_data.get('plugin_name') if plugin_name == self.plugin_name: logger.warning("权限被拒绝,部分功能将不可用") # 降级运行 await self._degrade_features() except Exception as e: logger.error(f"处理权限拒绝事件时出错: {str(e)}") async def _handle_network_data_receive(self, event_data): """处理网络数据接收事件""" try: data = event_data.get('data', {}) source = event_data.get('source', 'unknown') logger.debug(f"收到网络数据: {data.get('type')} from {source}") # 根据数据类型处理 await self._process_network_data(data, source) except Exception as e: logger.error(f"处理网络数据时出错: {str(e)}") async def _handle_user_login(self, event_data): """处理用户登录事件""" try: user = event_data.get('user', {}) logger.info(f"用户登录: {user.get('username')}") # 发送欢迎消息 if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "user", "action": "login", "user": user, "timestamp": datetime.now().isoformat() }) except Exception as e: logger.error(f"处理用户登录事件时出错: {str(e)}") async def _handle_user_logout(self, event_data): """处理用户登出事件""" try: user = event_data.get('user', {}) logger.info(f"用户登出: {user.get('username')}") # 清理用户相关资源 await self._cleanup_user_resources(user) except Exception as e: logger.error(f"处理用户登出事件时出错: {str(e)}") # ========== 辅助方法 ========== async def _check_rate_limit(self, client_ip: str, limit: int = 100) -> bool: """检查速率限制""" now = datetime.now() if client_ip not in self.rate_limiter: self.rate_limiter[client_ip] = { "count": 1, "window_start": now } return True # 检查时间窗口 window_start = self.rate_limiter[client_ip]["window_start"] window_age = (now - window_start).total_seconds() if window_age > 60: # 1分钟窗口 # 重置计数器 self.rate_limiter[client_ip] = { "count": 1, "window_start": now } return True # 增加计数 self.rate_limiter[client_ip]["count"] += 1 # 检查是否超限 if self.rate_limiter[client_ip]["count"] > limit: return False return True async def _get_user_from_request(self, request): """从请求中获取用户信息""" # 这里实现用户认证逻辑 # 可以从请求头中获取token,然后验证 token = request.headers.get('Authorization', '').replace('Bearer ', '') if token: # 验证token并返回用户信息 # 这里需要连接到认证服务 return { "id": "user_id", "username": "username", "permissions": [] } return None async def _check_admin_permission(self, request): """检查管理员权限""" user = await self._get_user_from_request(request) if user and "admin" in user.get("permissions", []): return True return False async def _broadcast_chat_message(self, message): """广播聊天消息""" if not self.network_bridge: return for connection_id, connection in self.websocket_connections.items(): try: if not connection['ws'].closed: await connection['ws'].send_str(json.dumps(message)) except Exception as e: logger.error(f"广播消息失败 {connection_id}: {str(e)}") async def _query_data(self, query_params): """查询数据""" # 这里实现数据查询逻辑 # 可以从数据库、文件或内存中查询 return [] async def _create_data(self, data): """创建数据""" # 这里实现数据创建逻辑 return {"id": "new_id"} async def _get_initial_data(self): """获取初始数据""" return {"message": "初始数据"} async def _get_updated_data(self): """获取更新数据""" return {"message": "更新数据", "timestamp": datetime.now().isoformat()} async def _check_network_health(self): """检查网络健康状态""" return True async def _check_cache_health(self): """检查缓存健康状态""" return True async def _check_tasks_health(self): """检查任务健康状态""" return True def _calculate_rps(self): """计算每秒请求数""" # 这里实现RPS计算逻辑 return 0.0 async def _sync_with_plugin(self, plugin_name): """与插件同步""" logger.debug(f"与插件同步: {plugin_name}") async def _cleanup_plugin_resources(self, plugin_name): """清理插件资源""" logger.debug(f"清理插件资源: {plugin_name}") async def _reinitialize_with_permissions(self, permissions): """重新初始化权限相关功能""" logger.debug(f"重新初始化权限: {permissions}") async def _degrade_features(self): """降级功能""" logger.debug("功能降级") async def _process_network_data(self, data, source): """处理网络数据""" logger.debug(f"处理网络数据: {data} from {source}") async def _cleanup_user_resources(self, user): """清理用户资源""" logger.debug(f"清理用户资源: {user.get('username')}") # ========== 后台任务方法 ========== async def _task_cleanup_old_data(self): """清理旧数据任务""" try: logger.info("开始清理旧数据...") # 实现清理逻辑 await asyncio.sleep(1) # 模拟清理过程 logger.info("旧数据清理完成") except Exception as e: logger.error(f"清理旧数据时出错: {str(e)}") async def _task_sync_external_data(self): """同步外部数据任务""" try: logger.info("开始同步外部数据...") # 实现同步逻辑 await asyncio.sleep(1) # 模拟同步过程 logger.info("外部数据同步完成") except Exception as e: logger.error(f"同步外部数据时出错: {str(e)}") # ========== 插件命令方法 ========== @plugin_command( name="status", description="查看插件状态", permissions=["plugin.my_awesome_plugin.read"] ) async def cmd_status(self, *args): """查看插件状态命令""" try: result = [] result.append(f"🔍 **{self.plugin_name} 插件状态**") result.append("=" * 50) result.append(f"📊 版本: {self.PLUGIN_VERSION}") result.append(f"🔄 状态: {'✅ 运行中' if self.status.is_running else '❌ 已停止'}") if self.status.start_time: result.append(f"⏰ 启动时间: {self.status.start_time.strftime('%Y-%m-%d %H:%M:%S')}") if self.status.uptime: result.append(f"⏱️ 运行时长: {self.status.uptime}") result.append(f"📈 请求总数: {self.status.request_count}") result.append(f"❌ 错误总数: {self.status.error_count}") # 网络状态 network_info = self.network_bridge.get_network_info() if self.network_bridge else {} result.append(f"🌐 网络状态: {'✅ 可用' if network_info else '❌ 不可用'}") if network_info: result.append(f" 基础URL: {network_info.get('base_url', 'N/A')}") result.append(f" HTTP路由: {len(network_info.get('registered_routes', []))} 个") result.append(f" WebSocket: {len(network_info.get('websocket_handlers', []))} 个") # 后台任务 result.append(f"🔧 后台任务: {len(self.background_tasks)} 个运行中") # 缓存状态 result.append(f"💾 缓存大小: {len(self.cache)} 项") # WebSocket连接 result.append(f"🔗 WebSocket连接: {len(self.websocket_connections)} 个") return "\n".join(result) except Exception as e: logger.error(f"状态命令执行失败: {str(e)}") return f"❌ 获取状态失败: {str(e)}" @plugin_command( name="config", description="查看或修改插件配置", permissions=["plugin.my_awesome_plugin.read", "plugin.my_awesome_plugin.write"] ) async def cmd_config(self, *args): """配置管理命令""" try: if not args: # 显示配置 result = [f"⚙️ **{self.plugin_name} 配置信息**"] result.append("=" * 50) for section, values in self.config.items(): if isinstance(values, dict): result.append(f"\n📁 {section.upper()}:") for key, value in list(values.items())[:5]: # 只显示前5项 result.append(f" {key}: {value}") if len(values) > 5: result.append(f" ... 还有 {len(values) - 5} 项配置") else: result.append(f"{section}: {values}") result.append("\n💡 使用: config get 查看具体配置") result.append("💡 使用: config set 修改配置") return "\n".join(result) command = args[0].lower() if command == "get": if len(args) < 2: return "❌ 请指定配置键,如: config get settings.log_level" key = args[1] value = self._get_nested_config(key) if value is not None: return f"✅ {key} = {value}" else: return f"❌ 配置键不存在: {key}" elif command == "set": if len(args) < 3: return "❌ 请指定配置键和值,如: config set settings.log_level DEBUG" key = args[1] value = args[2] # 尝试转换为适当类型 try: if value.lower() == "true": value = True elif value.lower() == "false": value = False elif value.isdigit(): value = int(value) elif value.replace('.', '', 1).isdigit(): value = float(value) except: pass success = self._set_nested_config(key, value) if success: # 保存配置到文件 await self._save_config() return f"✅ 配置已更新: {key} = {value}" else: return f"❌ 配置更新失败: {key}" else: return f"❌ 未知命令: {command}" except Exception as e: logger.error(f"配置命令执行失败: {str(e)}") return f"❌ 配置命令错误: {str(e)}" def _get_nested_config(self, key_path: str): """获取嵌套配置值""" keys = key_path.split('.') current = self.config for key in keys: if isinstance(current, dict) and key in current: current = current[key] else: return None return current def _set_nested_config(self, key_path: str, value): """设置嵌套配置值""" try: keys = key_path.split('.') current = self.config # 遍历到最后一个键的父级 for key in keys[:-1]: if key not in current or not isinstance(current[key], dict): current[key] = {} current = current[key] # 设置值 current[keys[-1]] = value return True except Exception: return False async def _save_config(self): """保存配置到文件""" try: config_path = Path(f"plugins/{self.plugin_name}/config.yaml") import yaml with open(config_path, 'w', encoding='utf-8') as f: yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True) logger.info(f"配置已保存: {config_path}") except Exception as e: logger.error(f"保存配置失败: {str(e)}") @plugin_command( name="network", description="网络功能管理", permissions=["plugin.my_awesome_plugin.network.access"] ) async def cmd_network(self, *args): """网络功能管理命令""" try: if not args: # 显示网络状态 if not self.network_bridge: return "❌ 网络功能不可用" info = self.network_bridge.get_network_info() result = [f"🌐 **{self.plugin_name} 网络状态**"] result.append("=" * 50) result.append(f"🔌 插件名称: {info['plugin_name']}") result.append(f"🌍 基础URL: {info['base_url']}") result.append(f"🛣️ HTTP路由数: {len(info['registered_routes'])}") result.append(f"🔗 WebSocket处理器数: {len(info['websocket_handlers'])}") if info['registered_routes']: result.append("\n📡 **注册的HTTP路由:**") for route in info['registered_routes']: auth_required = "🔐" if route['require_auth'] else "🔓" result.append(f" {auth_required} {route['path']} [{','.join(route['methods'])}]") if info['websocket_handlers']: result.append("\n⚡ **注册的WebSocket:**") for ws in info['websocket_handlers']: auth_required = "🔐" if ws['require_auth'] else "🔓" result.append(f" {auth_required} {ws['path']}") result.append("\n💡 使用: network test 测试网络连接") result.append("💡 使用: network restart 重启网络功能") return "\n".join(result) command = args[0].lower() if command == "test": # 测试网络连接 if not self.network_bridge: return "❌ 网络功能不可用" info = self.network_bridge.get_network_info() base_url = info['base_url'] if base_url == '网络服务不可用': return "❌ 网络服务不可用,无法测试" try: import aiohttp async with aiohttp.ClientSession() as session: async with session.get(f"{base_url}/api/health") as response: if response.status == 200: return "✅ 网络连接正常" else: return f"❌ 网络连接异常,状态码: {response.status}" except Exception as e: return f"❌ 网络测试失败: {str(e)}" elif command == "restart": # 重启网络功能 if not self.service_manager: return "❌ 服务管理器不可用" # 这里可以实现网络功能重启逻辑 return "🔄 网络功能重启中..." else: return f"❌ 未知命令: {command}" except Exception as e: logger.error(f"网络命令执行失败: {str(e)}") return f"❌ 网络命令错误: {str(e)}" @plugin_command( name="cache", description="缓存管理", permissions=["plugin.my_awesome_plugin.read"] ) async def cmd_cache(self, *args): """缓存管理命令""" try: if not args: # 显示缓存状态 result = [f"💾 **{self.plugin_name} 缓存状态**"] result.append("=" * 50) result.append(f"📊 缓存项数: {len(self.cache)}") result.append(f"⏱️ TTL项数: {len(self.cache_ttl)}") result.append(f"📈 命中率: {self.metrics.cache_hit_rate:.2%}") if self.cache: result.append("\n🔑 **缓存键列表 (前10个):**") for i, key in enumerate(list(self.cache.keys())[:10]): value = self.cache[key] value_preview = str(value)[:50] + "..." if len(str(value)) > 50 else str(value) result.append(f" {i+1}. {key}: {value_preview}") if len(self.cache) > 10: result.append(f" ... 还有 {len(self.cache) - 10} 个键") result.append("\n💡 使用: cache clear 清理所有缓存") result.append("💡 使用: cache get 获取缓存值") result.append("💡 使用: cache set [ttl] 设置缓存") return "\n".join(result) command = args[0].lower() if command == "clear": # 清理缓存 old_size = len(self.cache) self.cache.clear() self.cache_ttl.clear() return f"✅ 缓存已清理,共清理 {old_size} 项" elif command == "get": if len(args) < 2: return "❌ 请指定缓存键,如: cache get my_key" key = args[1] if key in self.cache: value = self.cache[key] # 检查是否过期 if key in self.cache_ttl: expiry = self.cache_ttl[key] if datetime.now() > expiry: del self.cache[key] del self.cache_ttl[key] return f"❌ 缓存已过期: {key}" return f"✅ {key} = {value}" else: return f"❌ 缓存键不存在: {key}" elif command == "set": if len(args) < 3: return "❌ 请指定缓存键和值,如: cache set my_key my_value" key = args[1] value = args[2] # 解析TTL ttl = None if len(args) > 3: try: ttl = int(args[3]) except ValueError: return "❌ TTL必须是整数(秒)" # 设置缓存 self.cache[key] = value if ttl: self.cache_ttl[key] = datetime.now() + timedelta(seconds=ttl) return f"✅ 缓存已设置: {key} = {value}" + (f" (TTL: {ttl}秒)" if ttl else "") elif command == "stats": # 显示详细统计 total_hits = 0 # 这里需要实现命中计数 total_misses = 0 if total_hits + total_misses > 0: hit_rate = total_hits / (total_hits + total_misses) else: hit_rate = 0 return ( f"📊 **缓存统计**\n" f"命中次数: {total_hits}\n" f"未命中次数: {total_misses}\n" f"命中率: {hit_rate:.2%}\n" f"内存使用: 约 {sum(len(str(v)) for v in self.cache.values())} 字节" ) else: return f"❌ 未知命令: {command}" except Exception as e: logger.error(f"缓存命令执行失败: {str(e)}") return f"❌ 缓存命令错误: {str(e)}" @plugin_command( name="tasks", description="后台任务管理", permissions=["plugin.my_awesome_plugin.read"] ) async def cmd_tasks(self, *args): """后台任务管理命令""" try: if not args: # 显示任务状态 result = [f"🔧 **{self.plugin_name} 后台任务**"] result.append("=" * 50) result.append(f"📊 总任务数: {len(self.background_tasks)}") running_tasks = [t for t in self.background_tasks if not t.done()] result.append(f"🔄 运行中: {len(running_tasks)}") result.append(f"✅ 已完成: {len(self.background_tasks) - len(running_tasks)}") if self.task_handles: result.append("\n📋 **任务列表:**") for name, task in self.task_handles.items(): status = "🟢 运行中" if not task.done() else "🔴 已停止" cancelled = " (已取消)" if task.cancelled() else "" result.append(f" {status}{cancelled} {name}") result.append("\n💡 使用: tasks start 启动任务") result.append("💡 使用: tasks stop 停止任务") result.append("💡 使用: tasks list 列出所有任务") return "\n".join(result) command = args[0].lower() if command == "list": # 列出所有任务 if not self.task_handles: return "📭 没有后台任务" result = ["📋 **后台任务列表:**"] for name, task in self.task_handles.items(): if task.done(): if task.cancelled(): status = "🔴 已取消" else: status = "✅ 已完成" else: status = "🟢 运行中" result.append(f" {status} {name}") return "\n".join(result) elif command == "start": if len(args) < 2: return "❌ 请指定任务名称,如: tasks start daily_cleanup" task_name = args[1] # 查找任务配置 task_config = None for schedule in self.config.get('schedules', []): if schedule.get('name') == task_name: task_config = schedule break if not task_config: return f"❌ 找不到任务: {task_name}" # 检查任务是否已在运行 if task_name in self.task_handles: task = self.task_handles[task_name] if not task.done(): return f"ℹ️ 任务已在运行: {task_name}" # 启动任务 task_func = getattr(self, f"_task_{task_config['task']}", None) if not task_func: return f"❌ 找不到任务处理函数: {task_config['task']}" task = asyncio.create_task( self._schedule_task(task_name, task_config['cron'], task_func) ) self.background_tasks.append(task) self.task_handles[task_name] = task return f"✅ 任务已启动: {task_name}" elif command == "stop": if len(args) < 2: return "❌ 请指定任务名称,如: tasks stop daily_cleanup" task_name = args[1] if task_name not in self.task_handles: return f"❌ 找不到任务: {task_name}" task = self.task_handles[task_name] if not task.done(): task.cancel() return f"🛑 任务已取消: {task_name}" else: return f"ℹ️ 任务已停止: {task_name}" elif command == "run": if len(args) < 2: return "❌ 请指定任务名称,如: tasks run daily_cleanup" task_name = args[1] # 查找任务函数 task_func = None for schedule in self.config.get('schedules', []): if schedule.get('name') == task_name: task_func_name = schedule.get('task') task_func = getattr(self, f"_task_{task_func_name}", None) break if not task_func: return f"❌ 找不到任务: {task_name}" # 立即执行任务 try: await task_func() return f"✅ 任务执行完成: {task_name}" except Exception as e: return f"❌ 任务执行失败: {str(e)}" else: return f"❌ 未知命令: {command}" except Exception as e: logger.error(f"任务命令执行失败: {str(e)}") return f"❌ 任务命令错误: {str(e)}" @plugin_command( name="admin", description="管理员命令", permissions=["plugin.my_awesome_plugin.admin"] ) async def cmd_admin(self, *args): """管理员命令""" try: if not args: return ( "⚡ **管理员命令**\n" "💡 使用: admin reload 重新加载插件\n" "💡 使用: admin debug 开启调试模式\n" "💡 使用: admin users 查看在线用户\n" "💡 使用: admin logs [count] 查看日志\n" ) command = args[0].lower() if command == "reload": # 重新加载插件 return "🔄 插件重新加载中..." elif command == "debug": # 切换调试模式 debug_enabled = self.config.get('debug', {}).get('enable_debug_endpoints', False) self.config.setdefault('debug', {})['enable_debug_endpoints'] = not debug_enabled status = "启用" if not debug_enabled else "禁用" return f"🔧 调试模式已{status}" elif command == "users": # 查看在线用户 if not self.websocket_connections: return "📭 没有在线用户" result = ["👥 **在线用户列表:**"] for conn_id, conn_info in self.websocket_connections.items(): user = conn_info.get('user', {}) connected_at = conn_info.get('connected_at') username = user.get('username', '未知用户') user_id = user.get('id', '未知ID') if connected_at: duration = datetime.now() - connected_at duration_str = str(duration).split('.')[0] else: duration_str = "未知" result.append(f" 👤 {username} (ID: {user_id}) - 连接时长: {duration_str}") return "\n".join(result) elif command == "logs": # 查看日志 count = 10 if len(args) > 1: try: count = min(int(args[1]), 50) except ValueError: return "❌ 日志数量必须是数字" # 这里需要实现日志查询逻辑 # 可以从日志文件或内存中读取 return f"📋 显示最近 {count} 条日志 (功能待实现)" else: return f"❌ 未知管理员命令: {command}" except Exception as e: logger.error(f"管理员命令执行失败: {str(e)}") return f"❌ 管理员命令错误: {str(e)}" @plugin_command( name="help", description="显示插件帮助信息" ) async def cmd_help(self, *args): """帮助命令""" try: result = [f"📚 **{self.plugin_name} 插件帮助**"] result.append("=" * 50) result.append(f"版本: {self.PLUGIN_VERSION}") result.append(f"描述: {self.config.get('description', '')}") result.append(f"作者: {self.config.get('author', '')}") result.append("\n🔧 **可用命令:**") # 扫描所有命令方法 command_methods = [] for attr_name in dir(self): if attr_name.startswith('cmd_'): method = getattr(self, attr_name) if hasattr(method, '_is_plugin_command'): command_name = getattr(method, '_command_name', attr_name[4:]) description = getattr(method, '_command_description', '') permissions = getattr(method, '_command_permissions', []) # 检查权限 has_permission = True if permissions: # 这里需要实现权限检查逻辑 pass if has_permission: command_methods.append((command_name, description)) # 按字母顺序排序 command_methods.sort(key=lambda x: x[0]) for cmd_name, cmd_desc in command_methods: result.append(f" 🟢 {cmd_name:15} - {cmd_desc}") result.append("\n🌐 **API接口:**") if self.network_bridge: info = self.network_bridge.get_network_info() result.append(f" 基础URL: {info.get('base_url', 'N/A')}") for route in info.get('registered_routes', []): result.append(f" 🔗 {route['path']} [{','.join(route['methods'])}]") result.append("\n💡 **使用提示:**") result.append(" 1. 使用 help 命令查看帮助") result.append(" 2. 使用 status 命令查看插件状态") result.append(" 3. 使用 config 命令管理配置") result.append(" 4. 使用 network 命令管理网络功能") result.append("\n⚠️ **注意事项:**") result.append(" 1. 部分命令需要特定权限") result.append(" 2. 修改配置后可能需要重启插件") result.append(" 3. 网络功能依赖于框架网络服务") return "\n".join(result) except Exception as e: logger.error(f"帮助命令执行失败: {str(e)}") return f"❌ 帮助命令错误: {str(e)}" # ========== 插件生命周期方法 ========== # ========== 插件生命周期方法 ========== async def shutdown(self): """ 关闭插件 执行顺序: 1. 停止所有后台任务 2. 关闭网络连接 3. 清理缓存和资源 4. 保存状态和配置 5. 清理事件处理器 6. 发送关闭通知 """ try: logger.info(f"开始关闭插件: {self.plugin_name}") # 1. 更新状态 self.status.is_running = False # 2. 发送关闭通知 await self._send_shutdown_notification() # 3. 取消所有后台任务 logger.info("正在停止后台任务...") task_cancellations = [] for task in self.background_tasks: if not task.done(): task.cancel() task_cancellations.append(task) # 等待所有任务取消完成 if task_cancellations: try: await asyncio.wait(task_cancellations, timeout=10.0) logger.info(f"后台任务已停止: {len(task_cancellations)} 个") except asyncio.TimeoutError: logger.warning("部分后台任务停止超时") # 4. 关闭WebSocket连接 logger.info("正在关闭WebSocket连接...") close_tasks = [] for conn_id, conn_info in list(self.websocket_connections.items()): try: if not conn_info['ws'].closed: close_task = asyncio.create_task( conn_info['ws'].close(code=1000, message='插件关闭') ) close_tasks.append(close_task) except Exception as e: logger.error(f"关闭WebSocket连接失败 {conn_id}: {str(e)}") if close_tasks: await asyncio.gather(*close_tasks, return_exceptions=True) self.websocket_connections.clear() # 5. 清理事件处理器 logger.info("正在清理事件处理器...") if hasattr(self.bridge, 'cleanup_plugin_subscriptions'): self.bridge.cleanup_plugin_subscriptions(self.plugin_name) elif hasattr(self.bridge, 'unsubscribe_all'): await self.bridge.unsubscribe_all(self.plugin_name) else: logger.warning("无法找到事件处理器清理方法,手动清理") for event_type in list(self.event_handlers.keys()): try: await self.bridge.unsubscribe_plugin( self.plugin_name, f"event.{event_type}" ) except Exception as e: logger.debug(f"清理事件处理器失败 {event_type}: {str(e)}") # 6. 清理缓存 logger.info("正在清理缓存...") self.cache.clear() self.cache_ttl.clear() # 清理Redis连接(如果存在) if hasattr(self, 'redis_client'): try: self.redis_client.close() logger.debug("Redis连接已关闭") except Exception as e: logger.warning(f"关闭Redis连接失败: {str(e)}") # 7. 保存配置和状态 logger.info("正在保存配置和状态...") await self._save_plugin_state() # 8. 清理锁和资源 logger.info("正在清理资源锁...") self._resource_locks.clear() # 清理任务句柄 self.task_handles.clear() # 9. 计算运行时长 if self.status.start_time: self.status.uptime = datetime.now() - self.status.start_time logger.info(f"插件运行时长: {self.status.uptime}") # 10. 发送插件停止事件 await self._send_plugin_stopped_event() logger.info(f"✅ 插件关闭完成: {self.plugin_name}") except Exception as e: logger.error(f"关闭插件时出错: {str(e)}") logger.error(traceback.format_exc()) # 紧急清理 await self._emergency_shutdown() async def _send_shutdown_notification(self): """发送关闭通知""" try: if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "system", "message": f"插件 {self.plugin_name} 正在关闭...", "timestamp": datetime.now().isoformat() }) # 发送框架事件 await self.bridge.publish_to_plugin( "framework", "event.plugin.shutting_down", { "plugin_name": self.plugin_name, "timestamp": datetime.now().isoformat() } ) except Exception as e: logger.debug(f"发送关闭通知失败: {str(e)}") async def _save_plugin_state(self): """保存插件状态""" try: state_data = { "plugin_name": self.plugin_name, "version": self.PLUGIN_VERSION, "status": { "last_run": datetime.now().isoformat(), "request_count": self.status.request_count, "error_count": self.status.error_count, "uptime": str(self.status.uptime) if self.status.uptime else None }, "config": self.config, "cache_stats": { "size": len(self.cache), "keys": list(self.cache.keys())[:20] # 只保存前20个键 }, "websocket_stats": { "max_connections": len(self.websocket_connections) } } state_path = Path(f"data/plugins/{self.plugin_name}/state.json") state_path.parent.mkdir(parents=True, exist_ok=True) with open(state_path, 'w', encoding='utf-8') as f: json.dump(state_data, f, ensure_ascii=False, indent=2) logger.debug(f"插件状态已保存: {state_path}") except Exception as e: logger.warning(f"保存插件状态失败: {str(e)}") async def _send_plugin_stopped_event(self): """发送插件停止事件""" try: await self.bridge.publish_to_plugin( "framework", "event.plugin.stopped", { "plugin_name": self.plugin_name, "version": self.PLUGIN_VERSION, "timestamp": datetime.now().isoformat(), "uptime": str(self.status.uptime) if self.status.uptime else None } ) except Exception as e: logger.debug(f"发送插件停止事件失败: {str(e)}") async def _emergency_shutdown(self): """紧急关闭""" try: logger.critical("执行紧急关闭...") # 强制取消所有任务 for task in self.background_tasks: if not task.done(): task.cancel() # 强制关闭WebSocket连接 for conn_info in self.websocket_connections.values(): try: if not conn_info['ws'].closed: conn_info['ws'].close() except: pass # 清理内存 self.cache.clear() self.websocket_connections.clear() self.task_handles.clear() logger.critical("紧急关闭完成") except Exception as e: logger.critical(f"紧急关闭时出错: {str(e)}") ``` #### 2.5.2 插件类核心方法详解 ##### 2.5.2.1 生命周期管理方法 ```python class Plugin: """ 插件生命周期管理方法详解 """ async def initialize(self) -> bool: """ 插件初始化 - 框架调用的主要入口点 返回: bool: 初始化是否成功 执行流程: 1. 基础设置和环境检查 2. 配置验证和加载 3. 服务管理器获取 4. 网络功能初始化 5. 事件处理器注册 6. 后台任务启动 7. 状态标记为运行中 """ try: # 1. 环境检查 if not await self._check_environment(): logger.error("环境检查失败") return False # 2. 配置验证 if not await self._validate_config(): logger.error("配置验证失败") return False # 3. 服务管理器获取 if not await self._setup_service_manager(): logger.warning("服务管理器获取失败,部分功能受限") # 4. 网络功能初始化 network_success = await self._initialize_network() if not network_success: logger.warning("网络功能初始化失败,将以受限模式运行") # 5. 事件处理器注册 await self._register_event_handlers() # 6. 后台任务启动 await self._start_background_tasks() # 7. 状态标记 self.status.is_running = True self.status.start_time = datetime.now() logger.info(f"✅ 插件初始化成功: {self.plugin_name}") return True except Exception as e: logger.error(f"❌ 插件初始化失败: {str(e)}") logger.error(traceback.format_exc()) await self._emergency_cleanup() return False async def _check_environment(self) -> bool: """检查运行环境""" try: # 检查Python版本 import sys if sys.version_info < (3, 8): logger.error("需要Python 3.8或更高版本") return False # 检查必要目录 required_dirs = [ f"plugins/{self.plugin_name}", f"data/plugins/{self.plugin_name}", f"logs/plugins/{self.plugin_name}" ] for dir_path in required_dirs: path = Path(dir_path) if not path.exists(): try: path.mkdir(parents=True, exist_ok=True) logger.debug(f"创建目录: {dir_path}") except Exception as e: logger.error(f"无法创建目录 {dir_path}: {str(e)}") return False # 检查依赖包 deps_ok = await self._check_dependencies() if not deps_ok: logger.error("依赖包检查失败") return False return True except Exception as e: logger.error(f"环境检查失败: {str(e)}") return False async def _check_dependencies(self) -> bool: """检查插件依赖""" try: dependencies = self.config.get('dependencies', {}) required = dependencies.get('required', []) optional = dependencies.get('optional', []) missing_required = [] for dep in required: # 解析依赖字符串,如 "requests>=2.25.0" package_name = dep.split('>=')[0].split('==')[0].split('<=')[0].strip() try: import importlib importlib.import_module(package_name) logger.debug(f"依赖检查通过: {package_name}") except ImportError: missing_required.append(package_name) logger.warning(f"缺少依赖包: {package_name}") if missing_required: logger.error(f"缺少必需依赖: {', '.join(missing_required)}") return False # 检查可选依赖 for dep in optional: package_name = dep.split('>=')[0].split('==')[0].split('<=')[0].strip() try: import importlib importlib.import_module(package_name) logger.debug(f"可选依赖可用: {package_name}") except ImportError: logger.info(f"可选依赖未安装: {package_name}") return True except Exception as e: logger.error(f"依赖检查失败: {str(e)}") return False ``` ##### 2.5.2.2 配置管理方法 ```python async def _validate_config(self) -> bool: """验证配置有效性""" try: # 基础配置验证 required_fields = ['name', 'version', 'description', 'author'] for field in required_fields: if field not in self.config: logger.error(f"缺少必需配置字段: {field}") return False # 版本号格式验证 version = self.config.get('version', '') import re if not re.match(r'^\d+\.\d+\.\d+(?:[-.]\w+)?$', version): logger.error(f"版本号格式错误: {version}") return False # 设置项验证 settings = self.config.get('settings', {}) if 'enabled' not in settings: logger.warning("settings.enabled 未设置,使用默认值 True") settings['enabled'] = True # 日志级别验证 log_level = settings.get('log_level', 'INFO') valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] if log_level not in valid_levels: logger.warning(f"无效的日志级别: {log_level},使用默认值 INFO") settings['log_level'] = 'INFO' # 更新配置 self.config['settings'] = settings # 功能配置验证 features = self.config.get('features', {}) if 'network' in features: network_config = features['network'] if network_config.get('enable_http', False) or network_config.get('enable_websocket', False): if not network_config.get('enable_cors', True): logger.warning("启用网络功能但禁用CORS可能导致跨域问题") logger.info("配置验证通过") return True except Exception as e: logger.error(f"配置验证失败: {str(e)}") return False ``` #### 2.5.3 事件处理与命令注册 ##### 2.5.3.1 事件处理系统 ```python async def _setup_event_system(self): """设置事件处理系统""" try: # 创建事件队列 self.event_queue = asyncio.Queue(maxsize=1000) # 启动事件处理器 self.event_handler_task = asyncio.create_task( self._event_handler_loop() ) self.background_tasks.append(self.event_handler_task) # 注册核心事件处理器 await self._register_core_event_handlers() logger.info("事件处理系统已启动") except Exception as e: logger.error(f"设置事件处理系统失败: {str(e)}") async def _event_handler_loop(self): """事件处理循环""" while self.status.is_running: try: # 从队列获取事件 event = await self.event_queue.get() # 处理事件 await self._process_event(event) # 标记任务完成 self.event_queue.task_done() except asyncio.CancelledError: logger.info("事件处理循环被取消") break except Exception as e: logger.error(f"事件处理出错: {str(e)}") await asyncio.sleep(1) # 出错后等待1秒 async def _process_event(self, event: dict): """处理单个事件""" try: event_type = event.get('type') event_data = event.get('data', {}) # 查找事件处理器 handler = self.event_handlers.get(event_type) if handler: # 执行处理器 await handler(event_data) else: # 默认处理器 await self._handle_unknown_event(event) except Exception as e: logger.error(f"处理事件失败 {event.get('type', 'unknown')}: {str(e)}") async def _register_core_event_handlers(self): """注册核心事件处理器""" core_handlers = { # 插件相关事件 'plugin.enable': self._handle_plugin_enable, 'plugin.disable': self._handle_plugin_disable, 'plugin.reload': self._handle_plugin_reload, # 用户相关事件 'user.created': self._handle_user_created, 'user.deleted': self._handle_user_deleted, 'user.updated': self._handle_user_updated, # 系统事件 'system.start': self._handle_system_start, 'system.stop': self._handle_system_stop, 'system.error': self._handle_system_error, # 自定义事件 'custom.notification': self._handle_custom_notification, 'custom.alert': self._handle_custom_alert, } # 注册到事件处理器映射 self.event_handlers.update(core_handlers) # 订阅框架事件 for event_type in core_handlers.keys(): try: await self.bridge.subscribe_event( self.plugin_name, event_type, core_handlers[event_type] ) except Exception as e: logger.warning(f"订阅事件失败 {event_type}: {str(e)}") ``` ##### 2.5.3.2 命令注册与执行 ```python async def _register_commands(self): """注册插件命令""" try: logger.info("开始注册插件命令...") # 扫描命令方法 command_methods = [] for attr_name in dir(self): if attr_name.startswith('cmd_'): method = getattr(self, attr_name) if hasattr(method, '_is_plugin_command'): command_methods.append(method) # 注册到框架 for method in command_methods: command_name = getattr(method, '_command_name', method.__name__[4:]) description = getattr(method, '_command_description', method.__doc__ or '') permissions = getattr(method, '_command_permissions', []) # 构建完整命令名 full_command_name = f"{self.plugin_name}_{command_name}" # 注册命令 await self.bridge.register_command( self.plugin_name, full_command_name, method, description, permissions ) logger.debug(f"命令注册: {full_command_name}") logger.info(f"命令注册完成,共 {len(command_methods)} 个命令") except Exception as e: logger.error(f"命令注册失败: {str(e)}") async def _execute_command(self, command: str, args: list) -> str: """执行命令的统一接口""" try: # 查找命令方法 method_name = f"cmd_{command}" if not hasattr(self, method_name): return f"❌ 未知命令: {command}" method = getattr(self, method_name) # 检查是否是插件命令 if not hasattr(method, '_is_plugin_command'): return f"❌ 不是有效的插件命令: {command}" # 执行命令 result = await method(*args) return result except Exception as e: logger.error(f"执行命令失败 {command}: {str(e)}") return f"❌ 命令执行错误: {str(e)}" ``` #### 2.5.4 异常处理与资源管理 ##### 2.5.4.1 异常处理框架 ```python class PluginExceptionHandler: """插件异常处理器""" def __init__(self, plugin_instance): self.plugin = plugin_instance self.error_history = [] self.max_error_history = 100 async def handle_exception(self, exception: Exception, context: str = "") -> dict: """处理异常并返回用户友好的错误信息""" try: # 记录异常 error_record = { 'timestamp': datetime.now().isoformat(), 'exception_type': type(exception).__name__, 'exception_message': str(exception), 'context': context, 'traceback': traceback.format_exc() } # 添加到历史 self.error_history.append(error_record) if len(self.error_history) > self.max_error_history: self.error_history.pop(0) # 更新插件状态 self.plugin.status.error_count += 1 self.plugin.status.last_error = str(exception) # 根据异常类型处理 if isinstance(exception, (PermissionError, PluginPermissionError)): return self._handle_permission_error(exception, context) elif isinstance(exception, (ConnectionError, TimeoutError)): return self._handle_network_error(exception, context) elif isinstance(exception, ValueError): return self._handle_validation_error(exception, context) elif isinstance(exception, FileNotFoundError): return self._handle_file_error(exception, context) else: return self._handle_generic_error(exception, context) except Exception as e: # 如果异常处理器本身出错 logger.critical(f"异常处理器出错: {str(e)}") return { 'success': False, 'error': '内部服务器错误', 'message': '系统遇到意外错误' } def _handle_permission_error(self, exception: Exception, context: str) -> dict: """处理权限错误""" logger.warning(f"权限错误 [{context}]: {str(exception)}") return { 'success': False, 'error': '权限不足', 'message': f'执行 {context} 需要特定权限', 'details': str(exception) } def _handle_network_error(self, exception: Exception, context: str) -> dict: """处理网络错误""" logger.error(f"网络错误 [{context}]: {str(exception)}") return { 'success': False, 'error': '网络连接失败', 'message': f'{context} 网络连接失败,请检查网络设置', 'details': str(exception) } def _handle_validation_error(self, exception: Exception, context: str) -> dict: """处理验证错误""" logger.warning(f"验证错误 [{context}]: {str(exception)}") return { 'success': False, 'error': '输入验证失败', 'message': f'{context} 输入数据无效', 'details': str(exception) } def _handle_file_error(self, exception: Exception, context: str) -> dict: """处理文件错误""" logger.error(f"文件错误 [{context}]: {str(exception)}") return { 'success': False, 'error': '文件操作失败', 'message': f'{context} 文件操作失败', 'details': str(exception) } def _handle_generic_error(self, exception: Exception, context: str) -> dict: """处理通用错误""" logger.error(f"通用错误 [{context}]: {str(exception)}") return { 'success': False, 'error': '操作失败', 'message': f'{context} 执行过程中发生错误', 'details': str(exception) if self.plugin.config.get('debug', {}).get('show_detailed_errors', False) else '请联系系统管理员' } ``` ##### 2.5.4.2 资源管理与清理 ```python class PluginResourceManager: """插件资源管理器""" def __init__(self, plugin_instance): self.plugin = plugin_instance self.resources = { 'files': [], # 打开的文件 'connections': [], # 网络连接 'locks': [], # 锁资源 'tasks': [], # 后台任务 'cache': [] # 缓存资源 } def register_resource(self, resource_type: str, resource, metadata: dict = None): """注册资源""" if resource_type not in self.resources: self.resources[resource_type] = [] resource_record = { 'resource': resource, 'type': type(resource).__name__, 'registered_at': datetime.now(), 'metadata': metadata or {} } self.resources[resource_type].append(resource_record) # 自动注册清理函数 if hasattr(resource, 'close'): self.plugin._cleanup_functions.append(resource.close) elif hasattr(resource, 'cleanup'): self.plugin._cleanup_functions.append(resource.cleanup) async def cleanup_all(self, force: bool = False): """清理所有资源""" cleanup_results = [] # 按逆序清理(后创建的先清理) for resource_type in reversed(list(self.resources.keys())): resources = self.resources[resource_type].copy() for resource_record in reversed(resources): try: result = await self._cleanup_resource(resource_record, force) cleanup_results.append((resource_type, result)) except Exception as e: logger.error(f"清理资源失败 {resource_type}: {str(e)}") cleanup_results.append((resource_type, False)) # 执行注册的清理函数 for cleanup_func in self.plugin._cleanup_functions: try: if asyncio.iscoroutinefunction(cleanup_func): await cleanup_func() else: cleanup_func() except Exception as e: logger.error(f"清理函数执行失败: {str(e)}") return cleanup_results async def _cleanup_resource(self, resource_record: dict, force: bool) -> bool: """清理单个资源""" resource = resource_record['resource'] resource_type = resource_record['type'] try: # 根据资源类型选择清理方式 if resource_type == 'File': if hasattr(resource, 'closed') and not resource.closed: resource.close() return True elif resource_type in ['Socket', 'Connection']: if hasattr(resource, 'close'): resource.close() return True elif resource_type == 'Lock': # 锁通常在上下文管理器中自动释放 pass elif resource_type == 'Task': if hasattr(resource, 'cancel') and not resource.done(): if force: resource.cancel() return True elif resource_type == 'Cache': if hasattr(resource, 'clear'): resource.clear() return True # 通用清理 if hasattr(resource, 'close'): resource.close() elif hasattr(resource, 'disconnect'): resource.disconnect() elif hasattr(resource, 'shutdown'): resource.shutdown() return True except Exception as e: logger.warning(f"清理资源失败 {resource_type}: {str(e)}") return False def get_resource_stats(self) -> dict: """获取资源统计信息""" stats = { 'total_resources': 0, 'by_type': {}, 'memory_usage': self._estimate_memory_usage() } for resource_type, resources in self.resources.items(): stats['by_type'][resource_type] = len(resources) stats['total_resources'] += len(resources) return stats def _estimate_memory_usage(self) -> int: """估计内存使用量(粗略)""" total_size = 0 # 遍历所有资源 for resource_type, resources in self.resources.items(): for resource_record in resources: resource = resource_record['resource'] # 尝试获取大小 try: if hasattr(resource, '__sizeof__'): total_size += resource.__sizeof__() elif isinstance(resource, (str, bytes, bytearray)): total_size += len(resource) except: pass return total_size ``` ### 2.5.5 插件配置持久化与状态恢复 ```python async def save_state(self) -> bool: """ 保存插件状态 保存内容包括: 1. 当前配置 2. 运行状态 3. 缓存数据 4. 用户会话 5. 任务状态 """ try: state_data = { 'plugin_info': { 'name': self.plugin_name, 'version': self.PLUGIN_VERSION, 'last_saved': datetime.now().isoformat() }, 'config': self.config, 'status': { 'is_running': self.status.is_running, 'start_time': self.status.start_time.isoformat() if self.status.start_time else None, 'request_count': self.status.request_count, 'error_count': self.status.error_count, 'last_error': self.status.last_error }, 'metrics': { 'requests_per_second': self.metrics.requests_per_second, 'average_response_time': self.metrics.average_response_time, 'active_connections': self.metrics.active_connections, 'cache_hit_rate': self.metrics.cache_hit_rate }, 'cache_summary': { 'total_items': len(self.cache), 'keys': list(self.cache.keys())[:50] # 只保存前50个键 }, 'background_tasks': [ { 'name': name, 'status': 'running' if not task.done() else 'completed', 'cancelled': task.cancelled() } for name, task in self.task_handles.items() ] } # 创建状态目录 state_dir = Path(f"data/plugins/{self.plugin_name}/state") state_dir.mkdir(parents=True, exist_ok=True) # 保存状态文件 state_file = state_dir / f"state_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" backup_file = state_dir / "state_backup.json" # 先备份当前状态 if backup_file.exists(): backup_file.unlink() # 写入新状态 with open(state_file, 'w', encoding='utf-8') as f: json.dump(state_data, f, ensure_ascii=False, indent=2) # 创建软链接到最新状态 latest_link = state_dir / "state_latest.json" if latest_link.exists(): latest_link.unlink() latest_link.symlink_to(state_file.name) # 保留最近10个状态文件 self._cleanup_old_state_files(state_dir) logger.info(f"插件状态已保存: {state_file}") return True except Exception as e: logger.error(f"保存插件状态失败: {str(e)}") return False def _cleanup_old_state_files(self, state_dir: Path, keep_count: int = 10): """清理旧的状态文件""" try: # 获取所有状态文件 state_files = list(state_dir.glob("state_*.json")) # 按修改时间排序 state_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) # 删除超出保留数量的文件 for state_file in state_files[keep_count:]: try: state_file.unlink() logger.debug(f"清理旧状态文件: {state_file}") except Exception as e: logger.warning(f"无法清理状态文件 {state_file}: {str(e)}") except Exception as e: logger.error(f"清理状态文件失败: {str(e)}") async def restore_state(self) -> bool: """ 恢复插件状态 从保存的状态文件恢复: 1. 恢复配置 2. 恢复缓存 3. 恢复任务状态 4. 恢复会话数据 """ try: state_file = Path(f"data/plugins/{self.plugin_name}/state/state_latest.json") if not state_file.exists(): logger.info("没有找到状态文件,使用默认状态") return False # 读取状态文件 with open(state_file, 'r', encoding='utf-8') as f: state_data = json.load(f) # 验证状态文件 if not self._validate_state_data(state_data): logger.warning("状态文件验证失败,使用默认状态") return False # 恢复配置 if 'config' in state_data: self.config.update(state_data['config']) logger.info("配置已从状态文件恢复") # 恢复状态信息 if 'status' in state_data: status_data = state_data['status'] self.status.request_count = status_data.get('request_count', 0) self.status.error_count = status_data.get('error_count', 0) logger.info("运行状态已恢复") # 恢复缓存 if 'cache_summary' in state_data: # 这里可以根据需要实现缓存的持久化和恢复 logger.info("缓存摘要已加载") logger.info(f"插件状态已从 {state_file} 恢复") return True except Exception as e: logger.error(f"恢复插件状态失败: {str(e)}") return False def _validate_state_data(self, state_data: dict) -> bool: """验证状态数据有效性""" try: # 检查必需字段 required_fields = ['plugin_info', 'config', 'status'] for field in required_fields: if field not in state_data: logger.error(f"状态文件缺少必需字段: {field}") return False # 验证插件信息 plugin_info = state_data['plugin_info'] if plugin_info.get('name') != self.plugin_name: logger.error(f"状态文件插件名称不匹配: {plugin_info.get('name')}") return False # 验证版本兼容性 saved_version = plugin_info.get('version', '') current_version = self.PLUGIN_VERSION # 简单的版本兼容性检查 if saved_version.split('.')[0] != current_version.split('.')[0]: logger.warning(f"主版本不匹配: 保存版本 {saved_version}, 当前版本 {current_version}") # 主版本不同可能不兼容 return True except Exception as e: logger.error(f"状态数据验证失败: {str(e)}") return False ``` ## 三、插件生命周期管理 ### 3.1 插件完整生命周期 ``` ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 加载阶段 │──▶│ 初始化阶段 │──▶│ 运行阶段 │──▶│ 关闭阶段 │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ▼ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 配置文件解析 │ │ 权限申请验证 │ │ 命令处理 │ │ 资源清理 │ ├─────────────┤ ├─────────────┤ ├─────────────┤ ├─────────────┤ │ 依赖检查 │ │ 网络路由注册 │ │ 事件处理 │ │ 连接关闭 │ ├─────────────┤ ├─────────────┤ ├─────────────┤ ├─────────────┤ │ 模块导入 │ │ 后台任务启动 │ │ API服务 │ │ 状态保存 │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ ``` ### 3.2 继续完成 shutdown 方法 ```python async def shutdown(self): """ 关闭插件 执行顺序: 1. 停止所有后台任务 2. 关闭网络连接 3. 清理缓存和资源 4. 保存状态和配置 5. 清理事件处理器 6. 发送关闭通知 """ try: logger.info(f"开始关闭插件: {self.plugin_name}") # 1. 更新状态 self.status.is_running = False # 2. 发送关闭通知 await self._send_shutdown_notification() # 3. 取消所有后台任务 logger.info("正在停止后台任务...") task_cancellations = [] for task in self.background_tasks: if not task.done(): task.cancel() task_cancellations.append(task) # 等待所有任务取消完成 if task_cancellations: try: await asyncio.wait(task_cancellations, timeout=10.0) logger.info(f"后台任务已停止: {len(task_cancellations)} 个") except asyncio.TimeoutError: logger.warning("部分后台任务停止超时") # 4. 关闭WebSocket连接 logger.info("正在关闭WebSocket连接...") close_tasks = [] for conn_id, conn_info in list(self.websocket_connections.items()): try: if not conn_info['ws'].closed: close_task = asyncio.create_task( conn_info['ws'].close(code=1000, message='插件关闭') ) close_tasks.append(close_task) except Exception as e: logger.error(f"关闭WebSocket连接失败 {conn_id}: {str(e)}") if close_tasks: await asyncio.gather(*close_tasks, return_exceptions=True) self.websocket_connections.clear() # 5. 清理事件处理器 logger.info("正在清理事件处理器...") if hasattr(self.bridge, 'cleanup_plugin_subscriptions'): self.bridge.cleanup_plugin_subscriptions(self.plugin_name) # 6. 清理缓存 logger.info("正在清理缓存...") self.cache.clear() self.cache_ttl.clear() # 7. 保存配置和状态 logger.info("正在保存配置和状态...") await self._save_plugin_state() # 8. 清理锁和资源 logger.info("正在清理资源锁...") self._resource_locks.clear() # 9. 计算运行时长 if self.status.start_time: self.status.uptime = datetime.now() - self.status.start_time logger.info(f"插件运行时长: {self.status.uptime}") logger.info(f"✅ 插件关闭完成: {self.plugin_name}") # 10. 发送关闭完成事件 await self._send_shutdown_complete_event() except Exception as e: logger.error(f"关闭插件时出错: {str(e)}") logger.error(traceback.format_exc()) # 紧急清理 await self._emergency_shutdown() async def _send_shutdown_notification(self): """发送关闭通知""" try: if self.network_bridge: await self.network_bridge.broadcast_websocket({ "type": "system", "message": f"插件 {self.plugin_name} 正在关闭...", "timestamp": datetime.now().isoformat() }) except Exception as e: logger.debug(f"发送关闭通知失败: {str(e)}") async def _save_plugin_state(self): """保存插件状态""" try: state_data = { "plugin_name": self.plugin_name, "version": self.PLUGIN_VERSION, "status": { "last_run": datetime.now().isoformat(), "request_count": self.status.request_count, "error_count": self.status.error_count, "uptime": str(self.status.uptime) if self.status.uptime else None }, "config": self.config, "cache_stats": { "size": len(self.cache), "keys": list(self.cache.keys())[:20] # 只保存前20个键 } } state_path = Path(f"data/plugins/{self.plugin_name}/state.json") state_path.parent.mkdir(parents=True, exist_ok=True) with open(state_path, 'w', encoding='utf-8') as f: json.dump(state_data, f, ensure_ascii=False, indent=2) logger.debug(f"插件状态已保存: {state_path}") except Exception as e: logger.warning(f"保存插件状态失败: {str(e)}") async def _send_shutdown_complete_event(self): """发送关闭完成事件""" try: await self.bridge.publish_to_plugin( "framework", "event.plugin.shutdown", { "plugin_name": self.plugin_name, "version": self.PLUGIN_VERSION, "timestamp": datetime.now().isoformat(), "uptime": str(self.status.uptime) if self.status.uptime else None } ) except Exception as e: logger.debug(f"发送关闭完成事件失败: {str(e)}") async def _emergency_shutdown(self): """紧急关闭""" try: logger.critical("执行紧急关闭...") # 强制取消所有任务 for task in self.background_tasks: if not task.done(): task.cancel() # 强制关闭WebSocket连接 for conn_info in self.websocket_connections.values(): try: if not conn_info['ws'].closed: conn_info['ws'].close() except: pass # 清理内存 self.cache.clear() self.websocket_connections.clear() logger.critical("紧急关闭完成") except Exception as e: logger.critical(f"紧急关闭时出错: {str(e)}") ``` ## 四、插件开发最佳实践 ### 4.1 错误处理最佳实践 ```python class PluginError(Exception): """插件基础异常类""" pass class PluginInitializationError(PluginError): """插件初始化异常""" pass class PluginPermissionError(PluginError): """插件权限异常""" pass class PluginNetworkError(PluginError): """插件网络异常""" pass def error_handler(func): """错误处理装饰器""" @wraps(func) async def wrapper(*args, **kwargs): try: return await func(*args, **kwargs) except PluginPermissionError as e: logger.error(f"权限错误: {str(e)}") return {"error": "权限不足", "details": str(e)} except PluginNetworkError as e: logger.error(f"网络错误: {str(e)}") return {"error": "网络错误", "details": str(e)} except asyncio.TimeoutError as e: logger.error(f"操作超时: {str(e)}") return {"error": "操作超时", "details": str(e)} except Exception as e: logger.error(f"未预期的错误: {str(e)}") logger.error(traceback.format_exc()) return {"error": "内部服务器错误", "details": str(e)} return wrapper class SafePlugin: """安全插件基类""" def __init__(self): self._error_context = [] def _record_error_context(self, context: str): """记录错误上下文""" self._error_context.append({ "timestamp": datetime.now().isoformat(), "context": context }) # 只保留最近的100条错误上下文 if len(self._error_context) > 100: self._error_context.pop(0) async def _safe_execute(self, func, *args, **kwargs): """安全执行函数""" try: return await func(*args, **kwargs) except Exception as e: # 记录错误上下文 error_info = { "function": func.__name__, "args": str(args), "kwargs": str(kwargs), "error": str(e), "traceback": traceback.format_exc(), "context": self._error_context.copy() } # 保存错误日志 await self._log_error(error_info) # 根据错误类型处理 if isinstance(e, (PermissionError, PluginPermissionError)): raise PluginPermissionError(f"权限错误: {str(e)}") elif isinstance(e, (ConnectionError, TimeoutError)): raise PluginNetworkError(f"网络错误: {str(e)}") else: raise PluginError(f"插件错误: {str(e)}") async def _log_error(self, error_info: dict): """记录错误日志""" error_log = { "plugin": self.plugin_name, "timestamp": datetime.now().isoformat(), "error": error_info } # 保存到文件 log_path = Path(f"logs/plugins/{self.plugin_name}/errors") log_path.mkdir(parents=True, exist_ok=True) log_file = log_path / f"error_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" try: with open(log_file, 'w', encoding='utf-8') as f: json.dump(error_log, f, ensure_ascii=False, indent=2) except Exception: pass ``` ### 4.2 性能优化最佳实践 ```python class OptimizedPlugin: """性能优化插件基类""" def __init__(self): # 缓存配置 self._cache_config = { "max_size": 1000, "ttl": 3600, "cleanup_interval": 300 } # 性能监控 self._performance_stats = { "request_times": [], "cache_hits": 0, "cache_misses": 0, "db_queries": 0 } # 连接池 self._connection_pools = {} # 异步锁 self._async_locks = {} def _get_cache_key(self, func_name: str, *args, **kwargs) -> str: """生成缓存键""" arg_str = str(args) kwarg_str = str(sorted(kwargs.items())) return f"{func_name}:{hashlib.md5((arg_str + kwarg_str).encode()).hexdigest()}" async def _cached_execute(self, func, ttl: int = None, *args, **kwargs): """带缓存执行""" cache_key = self._get_cache_key(func.__name__, *args, **kwargs) # 检查缓存 if cache_key in self.cache: # 检查TTL if cache_key in self.cache_ttl: if datetime.now() > self.cache_ttl[cache_key]: del self.cache[cache_key] del self.cache_ttl[cache_key] else: self._performance_stats["cache_hits"] += 1 return self.cache[cache_key] # 缓存未命中,执行函数 self._performance_stats["cache_misses"] += 1 result = await func(*args, **kwargs) # 存入缓存 self.cache[cache_key] = result if ttl: self.cache_ttl[cache_key] = datetime.now() + timedelta(seconds=ttl) # 清理过期的缓存 await self._cleanup_expired_cache() return result async def _cleanup_expired_cache(self): """清理过期缓存""" now = datetime.now() expired_keys = [] for key, expiry in self.cache_ttl.items(): if now > expiry: expired_keys.append(key) for key in expired_keys: if key in self.cache: del self.cache[key] if key in self.cache_ttl: del self.cache_ttl[key] # 如果缓存太大,清理最旧的项 if len(self.cache) > self._cache_config["max_size"]: # 简单的LRU策略:删除最早的缓存项 keys_to_remove = list(self.cache.keys())[:100] # 删除前100个 for key in keys_to_remove: if key in self.cache: del self.cache[key] if key in self.cache_ttl: del self.cache_ttl[key] def _get_async_lock(self, lock_name: str) -> asyncio.Lock: """获取异步锁""" if lock_name not in self._async_locks: self._async_locks[lock_name] = asyncio.Lock() return self._async_locks[lock_name] async def _rate_limited_execute(self, func, rate_limit: int = 10, *args, **kwargs): """限速执行""" lock = self._get_async_lock(f"rate_limit_{func.__name__}") async with lock: # 检查速率限制 current_time = time.time() key = f"rate_{func.__name__}" if key not in self.rate_limiter: self.rate_limiter[key] = [] # 清理旧的记录 self.rate_limiter[key] = [ t for t in self.rate_limiter[key] if current_time - t < 60 # 1分钟窗口 ] # 检查是否超限 if len(self.rate_limiter[key]) >= rate_limit: await asyncio.sleep(1) # 等待1秒 # 重新检查 self.rate_limiter[key] = [ t for t in self.rate_limiter[key] if current_time - t < 60 ] # 执行函数 self.rate_limiter[key].append(current_time) return await func(*args, **kwargs) def _measure_performance(self, func): """性能测量装饰器""" @wraps(func) async def wrapper(*args, **kwargs): start_time = time.time() try: result = await func(*args, **kwargs) return result finally: end_time = time.time() execution_time = end_time - start_time # 记录执行时间 self._performance_stats["request_times"].append(execution_time) # 只保留最近的1000个记录 if len(self._performance_stats["request_times"]) > 1000: self._performance_stats["request_times"].pop(0) # 记录慢查询 if execution_time > 1.0: # 超过1秒 logger.warning( f"慢查询: {func.__name__} 耗时 {execution_time:.2f}秒" ) return wrapper ``` ### 4.3 安全最佳实践 ```python class SecurePlugin: """安全插件基类""" def __init__(self): # 输入验证器 self._validators = { "email": self._validate_email, "url": self._validate_url, "ip_address": self._validate_ip_address, "filename": self._validate_filename, "sql_injection": self._check_sql_injection, "xss": self._check_xss } # 安全配置 self._security_config = { "max_file_size": 10 * 1024 * 1024, # 10MB "allowed_file_types": ['.txt', '.json', '.yaml', '.csv', '.log'], "max_request_size": 1024 * 1024, # 1MB "rate_limit_per_ip": 100, "session_timeout": 3600 } def _validate_input(self, input_data, validators=None): """验证输入数据""" if validators is None: validators = ["sql_injection", "xss"] errors = [] # 递归验证嵌套结构 def _validate_recursive(data, path=""): if isinstance(data, dict): for key, value in data.items(): current_path = f"{path}.{key}" if path else key _validate_recursive(value, current_path) elif isinstance(data, list): for i, item in enumerate(data): current_path = f"{path}[{i}]" _validate_recursive(item, current_path) elif isinstance(data, str): for validator_name in validators: if validator_name in self._validators: is_valid, error_msg = self._validators[validator_name](data) if not is_valid: errors.append(f"{path}: {error_msg}") _validate_recursive(input_data) if errors: raise PluginError(f"输入验证失败: {', '.join(errors)}") return True def _validate_email(self, email: str) -> tuple[bool, str]: """验证邮箱""" import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' if re.match(pattern, email): return True, "" return False, "无效的邮箱格式" def _validate_url(self, url: str) -> tuple[bool, str]: """验证URL""" import re pattern = r'^https?://[^\s/$.?#].[^\s]*$' if re.match(pattern, url): return True, "" return False, "无效的URL格式" def _validate_ip_address(self, ip: str) -> tuple[bool, str]: """验证IP地址""" import ipaddress try: ipaddress.ip_address(ip) return True, "" except ValueError: return False, "无效的IP地址" def _validate_filename(self, filename: str) -> tuple[bool, str]: """验证文件名""" import re # 防止路径遍历攻击 if '..' in filename or '/' in filename or '\\' in filename: return False, "文件名包含非法字符" # 检查文件扩展名 if '.' in filename: ext = filename[filename.rfind('.'):].lower() if ext not in self._security_config["allowed_file_types"]: return False, f"不允许的文件类型: {ext}" # 检查文件名长度 if len(filename) > 255: return False, "文件名过长" return True, "" def _check_sql_injection(self, text: str) -> tuple[bool, str]: """检查SQL注入""" sql_keywords = [ 'SELECT', 'INSERT', 'UPDATE', 'DELETE', 'DROP', 'UNION', 'OR', 'AND', 'WHERE', 'FROM', 'TABLE', 'DATABASE' ] text_upper = text.upper() for keyword in sql_keywords: # 简单的关键词检查 if f' {keyword} ' in f' {text_upper} ': return False, f"检测到SQL关键词: {keyword}" # 检查常见的注入模式 injection_patterns = [ r"'.*--", r"'.*;", r"OR\s+'.*'='.*'", r"AND\s+'.*'='.*'" ] import re for pattern in injection_patterns: if re.search(pattern, text_upper, re.IGNORECASE): return False, "检测到SQL注入模式" return True, "" def _check_xss(self, text: str) -> tuple[bool, str]: """检查XSS攻击""" xss_patterns = [ r".*?", r"javascript:", r"on\w+\s*=", r"<\s*iframe", r"<\s*img.*src\s*=", r"<\s*a.*href\s*=" ] import re for pattern in xss_patterns: if re.search(pattern, text, re.IGNORECASE): return False, "检测到XSS攻击模式" return True, "" async def _sanitize_output(self, data): """净化输出数据""" if isinstance(data, dict): return {k: await self._sanitize_output(v) for k, v in data.items()} elif isinstance(data, list): return [await self._sanitize_output(item) for item in data] elif isinstance(data, str): # 转义HTML特殊字符 import html return html.escape(data) else: return data def _generate_secure_token(self, length: int = 32) -> str: """生成安全令牌""" import secrets return secrets.token_hex(length) def _hash_password(self, password: str) -> str: """哈希密码""" import hashlib import os # 使用盐值 salt = os.urandom(32) key = hashlib.pbkdf2_hmac( 'sha256', password.encode('utf-8'), salt, 100000 # 迭代次数 ) return salt.hex() + key.hex() def _verify_password(self, password: str, hashed: str) -> bool: """验证密码""" import hashlib salt = bytes.fromhex(hashed[:64]) # 前64位是盐值 key = hashlib.pbkdf2_hmac( 'sha256', password.encode('utf-8'), salt, 100000 ) return hashed[64:] == key.hex() ``` ### 4.4 测试最佳实践 ```python import pytest import pytest_asyncio from unittest.mock import AsyncMock, Mock, patch class TestPlugin: """插件测试基类""" @pytest_asyncio.fixture async def plugin_instance(self): """创建插件实例""" config = { "name": "TestPlugin", "version": "1.0.0", "settings": {"enabled": True} } bridge_mock = AsyncMock() bridge_mock.service_manager = Mock() plugin = Plugin("test_plugin", config, bridge_mock) await plugin.initialize() yield plugin await plugin.shutdown() @pytest.mark.asyncio async def test_plugin_initialization(self, plugin_instance): """测试插件初始化""" assert plugin_instance.status.is_running == True assert plugin_instance.plugin_name == "test_plugin" assert plugin_instance.config["name"] == "TestPlugin" @pytest.mark.asyncio async def test_network_routes_registration(self, plugin_instance): """测试网络路由注册""" # 模拟网络桥接 network_bridge_mock = AsyncMock() plugin_instance.network_bridge = network_bridge_mock # 调用注册方法 await plugin_instance._register_network_routes() # 验证注册调用 assert network_bridge_mock.register_http_route.called assert network_bridge_mock.register_websocket.called @pytest.mark.asyncio async def test_command_execution(self, plugin_instance): """测试命令执行""" # 测试状态命令 result = await plugin_instance.cmd_status() assert "插件状态" in result assert plugin_instance.plugin_name in result @pytest.mark.asyncio async def test_error_handling(self, plugin_instance): """测试错误处理""" # 测试权限错误 with pytest.raises(PluginPermissionError): await plugin_instance._safe_execute( self._raise_permission_error ) # 测试网络错误 with pytest.raises(PluginNetworkError): await plugin_instance._safe_execute( self._raise_network_error ) def _raise_permission_error(self): raise PermissionError("测试权限错误") def _raise_network_error(self): raise ConnectionError("测试网络错误") @pytest.mark.asyncio async def test_rate_limiting(self, plugin_instance): """测试速率限制""" # 模拟多次调用 results = [] for i in range(15): # 超过10次限制 result = await plugin_instance._rate_limited_execute( self._dummy_function, rate_limit=10 ) results.append(result) # 验证所有调用都成功 assert len(results) == 15 assert all(r == "dummy_result" for r in results) async def _dummy_function(self): await asyncio.sleep(0.01) return "dummy_result" @pytest.mark.asyncio async def test_cache_functionality(self, plugin_instance): """测试缓存功能""" # 第一次调用应该缓存 result1 = await plugin_instance._cached_execute( self._expensive_function, ttl=60 ) # 第二次调用应该从缓存获取 result2 = await plugin_instance._cached_execute( self._expensive_function, ttl=60 ) assert result1 == result2 assert plugin_instance._performance_stats["cache_hits"] == 1 assert plugin_instance._performance_stats["cache_misses"] == 1 async def _expensive_function(self): await asyncio.sleep(0.1) return {"data": "expensive_result"} @pytest.mark.parametrize("input_data,expected", [ ("test@example.com", True), ("invalid-email", False), ("https://example.com", True), ("javascript:alert(1)", False), ("normal_text", True), ]) def test_input_validation(self, plugin_instance, input_data, expected): """测试输入验证""" validator = SecurePlugin() if expected: # 应该通过验证 assert validator._validate_input({"test": input_data}) == True else: # 应该抛出异常 with pytest.raises(PluginError): validator._validate_input({"test": input_data}) class IntegrationTest: """集成测试""" @pytest_asyncio.fixture async def framework_with_plugin(self): """创建带插件的框架实例""" from main import CatFramework framework = CatFramework() # 启动框架 await framework.initialize() # 加载测试插件 plugin_service = framework.service_manager.get_service("plugin") await plugin_service.load_plugin("test_plugin") yield framework # 关闭框架 await framework.shutdown() @pytest.mark.asyncio async def test_plugin_integration(self, framework_with_plugin): """测试插件与框架的集成""" # 获取插件服务 plugin_service = framework_with_plugin.service_manager.get_service("plugin") # 验证插件已加载 assert "test_plugin" in plugin_service.plugins # 验证插件命令已注册 command_service = framework_with_plugin.service_manager.get_service("command") command_list = command_service.get_command_list() plugin_commands = [ cmd for cmd in command_list if cmd["source"].startswith("plugin.test_plugin") ] assert len(plugin_commands) > 0 # 测试命令执行 result = await command_service.process_command("test_plugin_status", "test") assert "插件状态" in result @pytest.mark.asyncio async def test_plugin_network_integration(self, framework_with_plugin): """测试插件网络集成""" import aiohttp # 获取网络服务 internet_service = framework_with_plugin.service_manager.get_service("internet") # 测试HTTP API async with aiohttp.ClientSession() as session: url = f"http://localhost:{internet_service.http_port}/plugin/test_plugin/api/health" async with session.get(url) as response: assert response.status == 200 data = await response.json() assert data["status"] in ["healthy", "unhealthy"] ``` ## 五、插件发布与部署 ### 5.1 插件打包 ```yaml # setup.py 或 pyproject.toml 示例 """ 插件打包配置 """ # setup.py from setuptools import setup, find_packages setup( name="sensu-plugin-my-awesome-plugin", version="1.0.0", description="我的超棒插件", author="开发者名字", author_email="developer@example.com", packages=find_packages(), install_requires=[ "requests>=2.25.0", "pydantic>=1.8.0", ], extras_require={ "redis": ["redis>=3.5.0"], "mysql": ["aiomysql>=0.1.0"], }, package_data={ "": ["*.yaml", "*.json", "*.md"], }, entry_points={ "sensu.plugins": [ "my_awesome_plugin = my_awesome_plugin:Plugin", ], }, classifiers=[ "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", ], ) # pyproject.toml [build-system] requires = ["setuptools>=61.0"] build-backend = "setuptools.build_meta" [project] name = "sensu-plugin-my-awesome-plugin" version = "1.0.0" description = "我的超棒插件" authors = [ {name = "开发者名字", email = "developer@example.com"} ] dependencies = [ "requests>=2.25.0", "pydantic>=1.8.0" ] [project.optional-dependencies] redis = ["redis>=3.5.0"] mysql = ["aiomysql>=0.1.0"] [tool.setuptools] include-package-data = true [tool.setuptools.package-data] "*" = ["*.yaml", "*.json", "*.md"] [project.entry-points."sensu.plugins"] "my_awesome_plugin" = "my_awesome_plugin:Plugin" ``` ### 5.2 插件发布清单 ```markdown # 插件发布清单 ## 1. 代码质量检查 - [ ] 通过所有单元测试 - [ ] 通过集成测试 - [ ] 代码覆盖率 > 80% - [ ] 通过静态代码分析 - [ ] 通过安全扫描 ## 2. 文档检查 - [ ] README.md 完整 - [ ] 配置说明文档 - [ ] API文档 - [ ] 使用示例 - [ ] 更新日志 ## 3. 配置检查 - [ ] config.yaml 完整 - [ ] permissions.yaml 完整 - [ ] 默认配置合理 - [ ] 配置验证通过 ## 4. 依赖检查 - [ ] 依赖版本明确 - [ ] 无冲突依赖 - [ ] 可选依赖标注清楚 - [ ] 系统依赖说明 ## 5. 打包检查 - [ ] 打包脚本正确 - [ ] 包含所有必要文件 - [ ] 不包含敏感信息 - [ ] 版本号正确 ## 6. 性能检查 - [ ] 内存使用合理 - [ ] 启动时间 < 5秒 - [ ] API响应时间 < 1秒 - [ ] 支持并发请求 ## 7. 安全检查 - [ ] 输入验证完整 - [ ] 输出净化 - [ ] 权限控制 - [ ] 无硬编码密码 - [ ] 日志无敏感信息 ``` ### 5.3 持续集成配置 ```yaml # .github/workflows/ci.yml name: CI/CD on: push: branches: [ main ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest strategy: matrix: python-version: ["3.8", "3.9", "3.10"] steps: - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest pytest-asyncio pytest-cov pip install -e . - name: Run tests run: | pytest tests/ --cov=my_awesome_plugin --cov-report=xml - name: Upload coverage to Codecov uses: codecov/codecov-action@v3 with: file: ./coverage.xml fail_ci_if_error: true lint: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: "3.10" - name: Install linting tools run: | pip install black flake8 mypy pylint - name: Run black run: black --check . - name: Run flake8 run: flake8 . - name: Run mypy run: mypy my_awesome_plugin - name: Run pylint run: pylint my_awesome_plugin security: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Run bandit run: | pip install bandit bandit -r my_awesome_plugin -f json -o bandit-report.json - name: Run safety check run: | pip install safety safety check build: needs: [test, lint, security] runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: "3.10" - name: Build package run: | pip install build python -m build - name: Upload artifacts uses: actions/upload-artifact@v3 with: name: plugin-package path: dist/ ``` ## 六、插件调试与故障排除 ### 6.1 调试工具 ```python class DebugPlugin: """调试插件基类""" def __init__(self): self._debug_enabled = False self._debug_logs = [] self._performance_probes = {} def enable_debug(self): """启用调试模式""" self._debug_enabled = True logger.setLevel(logging.DEBUG) # 添加调试处理器 debug_handler = logging.StreamHandler() debug_handler.setLevel(logging.DEBUG) debug_handler.setFormatter(logging.Formatter( '%(asctime)s [DEBUG] %(name)s:%(lineno)d - %(message)s' )) logger.addHandler(debug_handler) def add_debug_log(self, message: str, data: dict = None): """添加调试日志""" if self._debug_enabled: log_entry = { "timestamp": datetime.now().isoformat(), "message": message, "data": data } self._debug_logs.append(log_entry) # 只保留最近的1000条日志 if len(self._debug_logs) > 1000: self._debug_logs.pop(0) async def _debug_probe(self, probe_name: str): """调试探针""" if not self._debug_enabled: return start_time = time.time() def finish(): end_time = time.time() duration = end_time - start_time if probe_name not in self._performance_probes: self._performance_probes[probe_name] = { "count": 0, "total_time": 0, "min_time": float('inf'), "max_time": 0, "last_time": 0 } stats = self._performance_probes[probe_name] stats["count"] += 1 stats["total_time"] += duration stats["min_time"] = min(stats["min_time"], duration) stats["max_time"] = max(stats["max_time"], duration) stats["last_time"] = duration self.add_debug_log( f"性能探针: {probe_name}", {"duration": duration, "stats": stats} ) return finish @plugin_command( name="debug", description="调试命令", permissions=["plugin.my_awesome_plugin.admin"] ) async def cmd_debug(self, *args): """调试命令""" try: if not args: return ( "🐛 **调试命令**\n" "💡 使用: debug enable 启用调试模式\n" "💡 使用: debug disable 禁用调试模式\n" "💡 使用: debug logs [count] 查看调试日志\n" "💡 使用: debug stats 查看性能统计\n" "💡 使用: debug memory 查看内存使用\n" "💡 使用: debug profile 性能分析\n" ) command = args[0].lower() if command == "enable": self.enable_debug() return "✅ 调试模式已启用" elif command == "disable": self._debug_enabled = False return "🛑 调试模式已禁用" elif command == "logs": count = 10 if len(args) > 1: try: count = min(int(args[1]), 100) except ValueError: return "❌ 日志数量必须是数字" if not self._debug_logs: return "📭 没有调试日志" result = [f"📋 **最近 {count} 条调试日志**"] for log in self._debug_logs[-count:]: result.append( f"[{log['timestamp']}] {log['message']}" ) if log['data']: result.append(f" 数据: {log['data']}") return "\n".join(result) elif command == "stats": if not self._performance_probes: return "📊 没有性能统计数据" result = ["📊 **性能统计**"] for probe_name, stats in self._performance_probes.items(): avg_time = stats["total_time"] / stats["count"] if stats["count"] > 0 else 0 result.append( f"{probe_name}: " f"调用{stats['count']}次, " f"平均{avg_time:.3f}秒, " f"最小{stats['min_time']:.3f}秒, " f"最大{stats['max_time']:.3f}秒" ) return "\n".join(result) elif command == "memory": import psutil import os process = psutil.Process(os.getpid()) memory_info = process.memory_info() result = ["💾 **内存使用**"] result.append(f"RSS: {memory_info.rss / 1024 / 1024:.2f} MB") result.append(f"VMS: {memory_info.vms / 1024 / 1024:.2f} MB") result.append(f"共享内存: {memory_info.shared / 1024 / 1024:.2f} MB") result.append(f"文本段: {memory_info.text / 1024 / 1024:.2f} MB") result.append(f"数据段: {memory_info.data / 1024 / 1024:.2f} MB") # 插件特定内存 result.append(f"缓存大小: {len(self.cache)} 项") result.append(f"WebSocket连接: {len(self.websocket_connections)} 个") return "\n".join(result) elif command == "profile": if len(args) < 2: return "❌ 请指定要分析的命令,如: debug profile status" sub_command = args[1] # 执行性能分析 import cProfile import io import pstats profiler = cProfile.Profile() profiler.enable() try: # 执行命令 command_func = getattr(self, f"cmd_{sub_command}", None) if command_func: result = await command_func(*args[2:]) else: result = f"❌ 找不到命令: {sub_command}" finally: profiler.disable() # 分析结果 s = io.StringIO() ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative') ps.print_stats(20) # 显示前20个最耗时的函数 profile_result = s.getvalue() return f"📈 **性能分析结果**\n```\n{profile_result}\n```\n\n**命令结果:**\n{result}" else: return f"❌ 未知调试命令: {command}" except Exception as e: logger.error(f"调试命令执行失败: {str(e)}") return f"❌ 调试命令错误: {str(e)}" ``` ### 6.2 故障排除指南 ```markdown # 插件故障排除指南 ## 1. 插件无法加载 ### 症状 - 插件没有出现在插件列表中 - 日志显示插件加载失败 ### 可能原因 1. 目录结构不正确 2. 配置文件缺失或格式错误 3. 权限文件格式错误 4. Python语法错误 5. 依赖包缺失 ### 解决方法 1. 检查插件目录结构: ``` plugins/ └── your_plugin/ ├── __init__.py ├── config.yaml └── permissions.yaml ``` 2. 验证配置文件: ```bash python -c "import yaml; yaml.safe_load(open('config.yaml'))" ``` 3. 检查Python语法: ```bash python -m py_compile __init__.py ``` 4. 查看详细日志: ```bash tail -f logs/runtime/*.log ``` ## 2. 权限申请失败 ### 症状 - 插件以受限模式运行 - 某些功能不可用 - 日志显示权限被拒绝 ### 可能原因 1. 权限名称格式错误 2. 权限描述不清晰 3. 申请了过多或不必要的权限 4. 用户拒绝了权限申请 ### 解决方法 1. 检查权限格式: ```yaml # 正确格式 permissions: - "plugin.your_plugin.read" - "plugin.your_plugin.write" # 错误格式 permissions: - "read" # 缺少插件名前缀 - "plugin.your_plugin.*" # 通配符可能被拒绝 ``` 2. 提供清晰的权限描述: ```yaml permission_descriptions: plugin.your_plugin.read: "读取插件数据,不会修改任何内容" plugin.your_plugin.write: "修改插件配置和数据" ``` 3. 分批申请权限: ```yaml # 第一次申请基础权限 permissions: - "plugin.your_plugin.read" # 后续根据需要申请更多权限 ``` ## 3. 网络功能不可用 ### 症状 - HTTP API返回404错误 - WebSocket连接失败 - 网络相关命令无法执行 ### 可能原因 1. 网络服务未启动 2. 端口被占用 3. 路由注册失败 4. 权限不足 ### 解决方法 1. 检查网络服务状态: ```bash # 在框架中执行 netdiag ``` 2. 检查端口占用: ```bash # Linux/Mac lsof -i :8000 # Windows netstat -ano | findstr :8000 ``` 3. 查看插件网络信息: ```bash # 在框架中执行 your_plugin network ``` 4. 重新注册网络路由: ```python # 在插件中 await self._register_network_routes() ``` ## 4. 性能问题 ### 症状 - 响应时间慢 - 内存使用率高 - CPU占用率高 ### 可能原因 1. 缓存未命中 2. 数据库查询效率低 3. 网络请求过多 4. 内存泄漏 ### 解决方法 1. 启用性能监控: ```bash # 在框架中执行 your_plugin debug stats ``` 2. 分析内存使用: ```bash your_plugin debug memory ``` 3. 优化数据库查询: ```python # 添加索引 # 使用连接池 # 批量操作 ``` 4. 实现缓存: ```python # 使用装饰器 @cached(ttl=300) async def get_data(self): # 耗时操作 pass ``` ## 5. 内存泄漏 ### 症状 - 内存使用持续增长 - 重启后恢复正常 - 长时间运行后变慢 ### 可能原因 1. 未关闭的资源(文件、连接等) 2. 循环引用 3. 缓存无限增长 4. 事件监听器未移除 ### 解决方法 1. 使用资源上下文管理器: ```python async with open_file() as f: # 使用文件 pass # 自动关闭 ``` 2. 定期清理缓存: ```python async def _cleanup_expired_cache(self): # 清理过期缓存 pass ``` 3. 使用弱引用: ```python import weakref self._callbacks = weakref.WeakSet() ``` 4. 监控内存使用: ```python import tracemalloc tracemalloc.start() # ... 运行代码 ... snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') ``` ## 6. 日志调试 ### 启用详细日志 ```python # 在插件配置中 settings: log_level: "DEBUG" ``` ### 查看插件特定日志 ```bash # 查找插件相关日志 grep "your_plugin" logs/runtime/*.log # 实时查看日志 tail -f logs/runtime/latest.log | grep "your_plugin" ``` ### 添加自定义日志 ```python logger.debug("详细调试信息", extra={"data": your_data}) logger.info("一般信息") logger.warning("警告信息") logger.error("错误信息", exc_info=True) ``` ## 7. 联系支持 如果以上方法都无法解决问题: 1. 收集以下信息: - 插件版本 - 框架版本 - 错误日志 - 复现步骤 - 系统信息 2. 提交问题报告: - GitHub Issues - 社区论坛 - 邮件支持 3. 提供最小复现示例: ```python # 简化的代码示例 # 能够重现问题的最小代码 ``` ``` ## 七、插件开发检查清单 ### 7.1 开发前检查清单 - [ ] **需求分析** - [ ] 明确插件功能需求 - [ ] 确定目标用户群体 - [ ] 分析使用场景 - [ ] 定义成功标准 - [ ] **技术选型** - [ ] 选择合适的技术栈 - [ ] 评估依赖包兼容性 - [ ] 确定性能要求 - [ ] 制定安全策略 - [ ] **架构设计** - [ ] 设计插件模块结构 - [ ] 规划API接口 - [ ] 设计数据模型 - [ ] 制定错误处理策略 ### 7.2 开发中检查清单 - [ ] **代码质量** - [ ] 遵循PEP 8编码规范 - [ ] 添加类型提示 - [ ] 编写文档字符串 - [ ] 实现单元测试 - [ ] **功能实现** - [ ] 实现核心功能 - [ ] 添加错误处理 - [ ] 实现日志记录 - [ ] 添加配置选项 - [ ] **安全性** - [ ] 验证所有输入 - [ ] 净化所有输出 - [ ] 实现权限控制 - [ ] 避免敏感信息泄露 ### 7.3 测试检查清单 - [ ] **单元测试** - [ ] 测试所有公开方法 - [ ] 测试错误处理 - [ ] 测试边界条件 - [ ] 测试异步方法 - [ ] **集成测试** - [ ] 测试插件加载 - [ ] 测试命令执行 - [ ] 测试网络功能 - [ ] 测试事件处理 - [ ] **性能测试** - [ ] 测试响应时间 - [ ] 测试内存使用 - [ ] 测试并发处理 - [ ] 测试资源清理 ### 7.4 发布检查清单 - [ ] **文档** - [ ] 编写README.md - [ ] 编写API文档 - [ ] 编写使用示例 - [ ] 编写更新日志 - [ ] **打包** - [ ] 创建打包配置 - [ ] 包含必要文件 - [ ] 设置版本号 - [ ] 添加依赖声明 - [ ] **验证** - [ ] 在新环境中测试 - [ ] 验证安装过程 - [ ] 测试升级流程 - [ ] 确认卸载清理 ## 八、文件管理器集成 (v0.6.0 新增) ### 8.1 文件管理器 Picker API 插件可通过 postMessage 接口调用文件管理器选择器,让用户在 WebUI 中快速选择目录或文件路径。无需插件自行实现文件浏览界面。 #### 8.1.1 权限要求 插件需要在 permissions.yaml 中声明文件管理相关权限: ```yaml permissions: - "filemanager.picker" # 调用文件选择器 - "filemanager.access" # 浏览/读取文件 - "filemanager.write" # 写入/删除文件 - "filemanager.*" # 完整权限 ``` | 权限 | 级别 | 说明 | |------|------|------| | `filemanager.picker` | 基础 | 唤起文件选择器悬浮窗 | | `filemanager.access` | 读取 | 浏览目录、读取文件内容 | | `filemanager.write` | 写入 | 创建/删除/重命名/上传/写入 | | `filemanager.*` | 管理 | 完整文件管理访问 | #### 8.1.2 调用文件选择器 插件 WebUI 页面通过 window.open 弹出选择器窗口: ```javascript // 选择目录 var picker = window.open( '/SenSu/static/pages/files.html?picker=1&mode=dir', 'fm-picker', 'width=680,height=520' ); // 选择文件 var picker = window.open( '/SenSu/static/pages/files.html?picker=1&mode=file', 'fm-picker', 'width=680,height=520' ); // 监听选择结果 window.addEventListener('message', function(e) { try { var data = JSON.parse(e.data); if (data.action === 'fm-picked' && data.path) { console.log('用户选择的路径:', data.path); // 将路径发送到插件后端进行处理 } } catch(ex) {} }); ``` 选择器关闭时通过 postMessage 返回: {"action":"fm-picked","path":"/选择的/路径"} #### 8.1.3 后端 REST API 参考 插件后端可直接调用文件管理 API(需声明对应权限): | 方法 | 端点 | 说明 | |------|------|------| | GET | /api/files/list?path=&show_hidden=0 | 列出目录内容 | | POST | /api/files/mkdir {path,name} | 创建目录 | | POST | /api/files/touch {path,name} | 创建空文件 | | GET | /api/files/read?path= | 读取文本(<=1MB) | | POST | /api/files/write {path,content} | 写入文本 | | POST | /api/files/delete {path} | 删除文件/目录 | | POST | /api/files/rename {path,new_name} | 重命名 | | POST | /api/files/upload (multipart) | 上传(<=50MB) | | GET | /api/files/download?path= | 下载文件 | | GET | /api/files/info?path= | 文件信息 | | GET | /api/files/picker?mode=file&path= | 选择器模式 | 安全: 文件管理器可浏览整个文件系统,路径穿越攻击会被拦截。 ## 九、总结 ### 9.1 成功插件的特点 1. **可靠性**:稳定运行,正确处理各种异常情况 2. **易用性**:简洁的API,清晰的文档,直观的配置 3. **安全性**:严格的输入验证,完善的权限控制 4. **性能**:高效的处理能力,合理的内存使用 5. **可维护性**:清晰的代码结构,完善的测试覆盖 6. **可扩展性**:支持插件间的协作,易于功能扩展 ### 8.2 持续改进 1. **收集反馈**:积极收集用户反馈,了解使用痛点 2. **监控使用**:通过日志和指标了解插件使用情况 3. **定期更新**:修复bug,添加新功能,优化性能 4. **保持兼容**:确保新版本与旧版本的兼容性 5. **社区参与**:参与插件生态建设,分享经验 ### 8.3 资源推荐 1. **学习资源** - Python官方文档 - asyncio官方文档 - Textual框架文档 - aiohttp文档 2. **工具推荐** - **代码质量**:black, flake8, mypy, pylint - **测试框架**:pytest, pytest-asyncio, coverage - **性能分析**:cProfile, memory_profiler, line_profiler - **打包工具**:setuptools, poetry, hatch 3. **社区支持** - GitHub Issues:报告问题和功能请求 - 论坛社区:交流开发经验 - Stack Overflow:解决具体技术问题 - 开发者群组:实时交流和协作 通过遵循本指南,您可以开发出高质量的SenSu插件,为用户提供有价值的功能,同时为插件生态系统做出贡献。祝您开发顺利! --- ## 九、v0.2.1 新增 API(已实现) ### 9.1 PluginStatus 状态枚举 ```python from core.plugin_status import PluginStatus # 插件生命周期状态 PluginStatus.UNLOADED # "unloaded" PluginStatus.LOADING # "loading" PluginStatus.LOADED # "loaded" PluginStatus.RUNNING # "running" PluginStatus.ERROR # "error" PluginStatus.STOPPING # "stopping" PluginStatus.STOPPED # "stopped" ``` 用法:`plugin_service.plugin_status["my_plugin"] = PluginStatus.RUNNING` ### 9.2 PluginError 异常层级 ```python from core.plugin_error import ( PluginError, # 基础异常 PluginLoadError, # 加载失败 PluginConfigError, # 配置错误 PluginPermissionError, # 权限不足 PluginCommandError, # 命令执行错误 ) # 抛出 raise PluginConfigError("缺少必需配置项: name", plugin_name="my_plugin") ``` ### 9.3 PluginBridge.subscribe_plugin() ```python # 便捷订阅方法(同时注册到 CoreBridge 和插件订阅表) self.bridge.subscribe_plugin("test.event", self._handler, "my_plugin") ``` ### 9.4 PluginNetworkBridge 跨插件数据 ```python # 设置数据接收通道 await self.network_bridge.setup_data_transfer("status_update", self._on_data) # 向其他插件发送数据 await self.network_bridge.send_data("target_plugin", "status_update", {"key": "val"}) ``` ### 9.5 配置验证 框架在插件加载时自动检查: - `config.yaml` 必须存在且可解析 - 必需字段:`name`, `version` - 版本号格式建议遵循 semver