Files
abot/plugin_common/plugin_interface.py

167 lines
3.9 KiB
Python

import os
import toml
from abc import ABC, abstractmethod
from enum import Enum
from typing import Dict, Any, List, Optional
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 = ""
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._config.update(plugin_config)
print(f"{config_path} 加载插件配置成功")
return True
else:
print(f"插件配置文件 {config_path} 不存在,使用默认配置")
return True # 配置文件不存在也视为成功,使用默认配置
except Exception as e:
print(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