204 lines
5.1 KiB
Python
204 lines
5.1 KiB
Python
import os
|
||
import toml
|
||
from abc import ABC, abstractmethod
|
||
from loguru import logger
|
||
from enum import Enum
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
|
||
|
||
class PluginStatus(Enum):
|
||
"""插件状态枚举"""
|
||
UNLOADED = 0 # 未加载
|
||
LOADED = 1 # 已加载但未启动
|
||
RUNNING = 2 # 运行中
|
||
STOPPED = 3 # 已停止
|
||
ERROR = 4 # 错误状态
|
||
|
||
|
||
class PluginInterface(ABC):
|
||
"""插件基础接口，所有插件必须实现此接口"""
|
||
|
||
@property
|
||
@abstractmethod
|
||
def name(self) -> str:
|
||
"""插件名称"""
|
||
pass
|
||
|
||
@property
|
||
@abstractmethod
|
||
def version(self) -> str:
|
||
"""插件版本"""
|
||
pass
|
||
|
||
@property
|
||
@abstractmethod
|
||
def description(self) -> str:
|
||
"""插件描述"""
|
||
pass
|
||
|
||
@property
|
||
@abstractmethod
|
||
def author(self) -> str:
|
||
"""插件作者"""
|
||
pass
|
||
|
||
@property
|
||
def dependencies(self) -> List[str]:
|
||
"""插件依赖，返回依赖的其他插件名称列表"""
|
||
return []
|
||
|
||
@property
|
||
def status(self) -> PluginStatus:
|
||
"""获取插件当前状态"""
|
||
return self._status
|
||
|
||
@status.setter
|
||
def status(self, value: PluginStatus):
|
||
"""设置插件状态"""
|
||
self._status = value
|
||
|
||
def __init__(self):
|
||
"""初始化插件"""
|
||
self._status = PluginStatus.UNLOADED
|
||
self._config = {}
|
||
self._plugin_path = ""
|
||
# 初始化日志记录器
|
||
self.LOG = logger
|
||
self.bot = None
|
||
|
||
def load_config(self) -> bool:
|
||
"""
|
||
从插件目录下的config.toml加载配置
|
||
|
||
Returns:
|
||
加载是否成功
|
||
"""
|
||
try:
|
||
config_path = os.path.join(self._plugin_path, "config.toml")
|
||
if os.path.exists(config_path):
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
plugin_config = toml.load(f)
|
||
self.LOG.debug(f"从 {config_path} 加载插件{self.name}配置:{plugin_config}")
|
||
self._config.update(plugin_config)
|
||
self.LOG.debug(f"从 {config_path} 加载插件配置成功")
|
||
return True
|
||
else:
|
||
self.LOG.info(f"插件配置文件 {config_path} 不存在，使用默认配置")
|
||
return True # 配置文件不存在也视为成功，使用默认配置
|
||
except Exception as e:
|
||
self.LOG.info(f"加载插件配置失败: {e}")
|
||
return False
|
||
|
||
def set_plugin_path(self, path: str) -> None:
|
||
"""
|
||
设置插件路径
|
||
|
||
Args:
|
||
path: 插件路径
|
||
"""
|
||
self._plugin_path = path
|
||
|
||
def get_plugin_path(self) -> str:
|
||
"""
|
||
获取插件路径
|
||
|
||
Returns:
|
||
插件路径
|
||
"""
|
||
return self._plugin_path
|
||
|
||
@abstractmethod
|
||
def initialize(self, context: Dict[str, Any]) -> bool:
|
||
"""
|
||
初始化插件
|
||
|
||
Args:
|
||
context: 插件上下文，包含系统环境和配置信息
|
||
|
||
Returns:
|
||
初始化是否成功
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def start(self) -> bool:
|
||
"""
|
||
启动插件
|
||
|
||
Returns:
|
||
启动是否成功
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def stop(self) -> bool:
|
||
"""
|
||
停止插件
|
||
|
||
Returns:
|
||
停止是否成功
|
||
"""
|
||
pass
|
||
|
||
def configure(self, config: Dict[str, Any]) -> bool:
|
||
"""
|
||
配置插件
|
||
|
||
Args:
|
||
config: 插件配置
|
||
|
||
Returns:
|
||
配置是否成功
|
||
"""
|
||
self._config.update(config)
|
||
return True
|
||
|
||
def get_config(self) -> Dict[str, Any]:
|
||
"""
|
||
获取插件配置
|
||
|
||
Returns:
|
||
插件配置
|
||
"""
|
||
return self._config
|
||
|
||
def cleanup(self) -> bool:
|
||
"""
|
||
清理插件资源，在卸载前调用
|
||
|
||
Returns:
|
||
清理是否成功
|
||
"""
|
||
return True
|
||
|
||
def get_config_path(self) -> str:
|
||
return os.path.join(self._plugin_path, 'config.toml')
|
||
|
||
def can_process(self, data: Any) -> bool:
|
||
"""检查插件是否可以处理给定的数据
|
||
|
||
这是一个通用方法，用于检查插件是否可以处理特定类型的数据。
|
||
子类可以根据需要重写此方法。
|
||
|
||
Args:
|
||
data: 要检查的数据
|
||
|
||
Returns:
|
||
如果插件可以处理该数据，则返回True，否则返回False
|
||
"""
|
||
return False
|
||
|
||
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
|
||
"""
|
||
处理消息
|
||
|
||
Args:
|
||
message: 消息字典，包含消息的各种属性，以及发送消息所需的对象
|
||
- wcf: WcfAPI对象，可用于发送消息
|
||
- message_util: 消息工具类，提供更高级的消息处理功能
|
||
|
||
Returns:
|
||
(是否已处理, 处理结果)
|
||
"""
|
||
raise NotImplementedError("子类必须实现此方法")
|