from abc import ABCMeta, abstractmethod from pathlib import Path from typing import Any, List, Dict, Tuple, Optional from app.chain import ChainBase from app.core.config import settings from app.core.event import EventManager from app.db.plugindata_oper import PluginDataOper from app.db.systemconfig_oper import SystemConfigOper from app.helper.message import MessageHelper from app.schemas import Notification, NotificationType, MessageChannel class PluginChian(ChainBase): """ 插件处理链 """ pass class _PluginBase(metaclass=ABCMeta): """ 插件模块基类，通过继续该类实现插件功能 除内置属性外，还有以下方法可以扩展或调用： - stop_service() 停止插件服务 - get_config() 获取配置信息 - update_config() 更新配置信息 - init_plugin() 生效配置信息 - get_data_path() 获取插件数据保存目录 """ # 插件名称 plugin_name: Optional[str] = "" # 插件描述 plugin_desc: Optional[str] = "" # 插件顺序 plugin_order: Optional[int] = 9999 # 是否为插件分身 is_clone: bool = False def __init__(self): # 插件数据 self.plugindata = PluginDataOper() # 处理链 self.chain = PluginChian() # 系统配置 self.systemconfig = SystemConfigOper() # 系统消息 self.systemmessage = MessageHelper() # 事件管理器 self.eventmanager = EventManager() @abstractmethod def init_plugin(self, config: dict = None): """ 生效配置信息 :param config: 配置信息字典 """ pass def get_name(self) -> str: """ 获取插件名称 :return: 插件名称 """ return self.plugin_name @abstractmethod def get_state(self) -> bool: """ 获取插件运行状态 """ pass @staticmethod def get_command() -> List[Dict[str, Any]]: """ 注册插件远程命令 [{ "cmd": "/xx", "event": EventType.xx, "desc": "名称", "category": "分类，需要注册到Wechat时必须有分类", "data": {} }] """ pass @staticmethod def get_render_mode() -> Tuple[str, Optional[str]]: """ 获取插件渲染模式 :return: 1、渲染模式，支持：vue/vuetify，默认vuetify；2、vue模式下编译后文件的相对路径，默认为`dist/asserts`，vuetify模式下为None """ return "vuetify", None @abstractmethod def get_api(self) -> List[Dict[str, Any]]: """ 注册插件API [{ "path": "/xx", "endpoint": self.xxx, "methods": ["GET", "POST"], "auth: "apikey", # 鉴权类型：apikey/bear "summary": "API名称", "description": "API说明" }] """ pass @abstractmethod def get_form(self) -> Tuple[Optional[List[dict]], Dict[str, Any]]: """ 拼装插件配置页面，插件配置页面使用Vuetify组件拼装，参考：https://vuetifyjs.com/ :return: 1、页面配置（vuetify模式）或 None（vue模式）；2、默认数据结构 """ pass @abstractmethod def get_page(self) -> Optional[List[dict]]: """ 拼装插件详情页面，需要返回页面配置，同时附带数据 插件详情页面使用Vuetify组件拼装，参考：https://vuetifyjs.com/ :return: 页面配置（vuetify模式）或 None（vue模式） """ pass def get_service(self) -> List[Dict[str, Any]]: """ 注册插件公共服务 [{ "id": "服务ID", "name": "服务名称", "trigger": "触发器：cron/interval/date/CronTrigger.from_crontab()", "func": self.xxx, "kwargs": {} # 定时器参数 }] """ pass def get_dashboard(self, key: str, **kwargs) -> Optional[Tuple[Dict[str, Any], Dict[str, Any], Optional[List[dict]]]]: """ 获取插件仪表盘页面，需要返回：1、仪表板col配置字典；2、全局配置（布局、自动刷新等）；3、仪表板页面元素配置含数据json（vuetify）或 None（vue模式） 1、col配置参考： { "cols": 12, "md": 6 } 2、全局配置参考： { "refresh": 10, // 自动刷新时间，单位秒 "border": True, // 是否显示边框，默认True，为False时取消组件边框和边距，由插件自行控制 "title": "组件标题", // 组件标题，如有将显示该标题，否则显示插件名称 "subtitle": "组件子标题", // 组件子标题，缺省时不展示子标题 } 3、vuetify模式页面配置使用Vuetify组件拼装，参考：https://vuetifyjs.com/；vue模式为None kwargs参数可获取的值：1、user_agent：浏览器UA :param key: 仪表盘key，根据指定的key返回相应的仪表盘数据，缺省时返回一个固定的仪表盘数据（兼容旧版） """ pass def get_dashboard_meta(self) -> Optional[List[Dict[str, str]]]: """ 获取插件仪表盘元信息 返回示例： [{ "key": "dashboard1", // 仪表盘的key，在当前插件范围唯一 "name": "仪表盘1" // 仪表盘的名称 }, { "key": "dashboard2", "name": "仪表盘2" }] """ pass def get_module(self) -> Dict[str, Any]: """ 获取插件模块声明，用于胁持系统模块实现（方法名：方法实现） { "id1": self.xxx1, "id2": self.xxx2, } """ pass def get_actions(self) -> List[Dict[str, Any]]: """ 获取插件工作流动作 [{ "id": "动作ID", "name": "动作名称", "func": self.xxx, "kwargs": {} # 需要附加传递的参数 }] 对实现函数的要求： 1、函数的第一个参数固定为 ActionContent 实例，如需要传递额外参数，在kwargs中定义 2、函数的返回：执行状态 True / False，更新后的 ActionContent 实例 """ pass @abstractmethod def stop_service(self): """ 停止插件 """ pass def update_config(self, config: dict, plugin_id: Optional[str] = None) -> bool: """ 更新配置信息 :param config: 配置信息字典 :param plugin_id: 插件ID """ if not plugin_id: plugin_id = self.__class__.__name__ return self.systemconfig.set(f"plugin.{plugin_id}", config) def get_config(self, plugin_id: Optional[str] = None) -> Any: """ 获取配置信息 :param plugin_id: 插件ID """ if not plugin_id: plugin_id = self.__class__.__name__ return self.systemconfig.get(f"plugin.{plugin_id}") def get_data_path(self, plugin_id: Optional[str] = None) -> Path: """ 获取插件数据保存目录 """ if not plugin_id: plugin_id = self.__class__.__name__ data_path = settings.PLUGIN_DATA_PATH / f"{plugin_id}" if not data_path.exists(): data_path.mkdir(parents=True) return data_path def save_data(self, key: str, value: Any, plugin_id: Optional[str] = None): """ 保存插件数据 :param key: 数据key :param value: 数据值 :param plugin_id: 插件ID """ if not plugin_id: plugin_id = self.__class__.__name__ self.plugindata.save(plugin_id, key, value) def get_data(self, key: Optional[str] = None, plugin_id: Optional[str] = None) -> Any: """ 获取插件数据 :param key: 数据key :param plugin_id: plugin_id """ if not plugin_id: plugin_id = self.__class__.__name__ return self.plugindata.get_data(plugin_id, key) def del_data(self, key: str, plugin_id: Optional[str] = None) -> Any: """ 删除插件数据 :param key: 数据key :param plugin_id: plugin_id """ if not plugin_id: plugin_id = self.__class__.__name__ return self.plugindata.del_data(plugin_id, key) def post_message(self, channel: MessageChannel = None, mtype: NotificationType = None, title: Optional[str] = None, text: Optional[str] = None, image: Optional[str] = None, link: Optional[str] = None, userid: Optional[str] = None, username: Optional[str] = None, **kwargs): """ 发送消息 """ if not link: link = settings.MP_DOMAIN(f"#/plugins?tab=installed&id={self.__class__.__name__}") self.chain.post_message(Notification( channel=channel, mtype=mtype, title=title, text=text, image=image, link=link, userid=userid, username=username, **kwargs )) def close(self): pass