diff --git a/docs/SenSu 插件开发详细指南.md b/docs/SenSu 插件开发详细指南.md index c8f7b95..c58a89e 100644 --- a/docs/SenSu 插件开发详细指南.md +++ b/docs/SenSu 插件开发详细指南.md @@ -1,5 +1,39 @@ # 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-发布检查清单) + - [八、总结](#八总结) + - [8.1 成功插件的特点](#81-成功插件的特点) + - [8.2 持续改进](#82-持续改进) + - [8.3 资源推荐](#83-资源推荐) + +--- + + ## 一、插件系统架构深度解析 ### 1.1 插件生命周期 diff --git a/docs/SenSu 插件开发详细指南.md.bak b/docs/SenSu 插件开发详细指南.md.bak new file mode 100644 index 0000000..c8f7b95 --- /dev/null +++ b/docs/SenSu 插件开发详细指南.md.bak @@ -0,0 +1,5044 @@ +# SenSu 插件开发超详细指南 + +## 一、插件系统架构深度解析 + +### 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 fmfuncs.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文档 + - [ ] 编写使用示例 + - [ ] 编写更新日志 + +- [ ] **打包** + - [ ] 创建打包配置 + - [ ] 包含必要文件 + - [ ] 设置版本号 + - [ ] 添加依赖声明 + +- [ ] **验证** + - [ ] 在新环境中测试 + - [ ] 验证安装过程 + - [ ] 测试升级流程 + - [ ] 确认卸载清理 + +## 八、总结 + +### 8.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插件，为用户提供有价值的功能，同时为插件生态系统做出贡献。祝您开发顺利！ \ No newline at end of file diff --git a/services/web_panel/utils/system_info.py b/services/web_panel/utils/system_info.py index 095e8a1..5770b34 100644 --- a/services/web_panel/utils/system_info.py +++ b/services/web_panel/utils/system_info.py @@ -51,7 +51,7 @@ class SystemInfoCollector: } # Android 估算：负载率 = (1分钟负载 / 核心数) * 100 try: - load = os.getloadavg() + load = os.getloadavg() if hasattr(os, "getloadavg") else [0.0, 0.0, 0.0] cores = os.cpu_count() or 1 percent = min(100.0, (load[0] / cores) * 100) return {"percent": round(percent, 1), "cores": cores, "load_avg": load}