pbf.controller.PluginsManager

  1import os
  2import traceback
  3
  4try:
  5    from ..config import plugins_directory, plugins_disabled
  6    from ..utils.Logging import Logger
  7    from ..utils import Path, MetaData
  8    from ..setup import ListenerManager
  9    from ..error import NoApiError
 10except ImportError:
 11    from pbf.config import plugins_directory, plugins_disabled
 12    from pbf.utils.Logging import Logger
 13    from pbf.utils import Path, MetaData
 14    from pbf.setup import ListenerManager
 15    from pbf.error import NoApiError
 16
 17logger = Logger(__name__)
 18
 19
 20class PluginsManager:
 21    plugins: dict = {}
 22    api: dict = {}
 23    plugin_name: str = None
 24
 25    def __init__(self, path: str = plugins_directory):
 26        """
 27        初始化PluginsManager。
 28        :param path: str (可选)插件目录，默认为`pbf.config.plugins_directory`
 29        :return: None
 30        """
 31        self.path = Path.replace(path)
 32
 33    def clearPlugins(self):
 34        """
 35        清空所有插件。
 36        :return: None
 37        """
 38        # 销毁所有self.plugins中的对象
 39        for key, _ in self.plugins.items():
 40            self.disable(key, pop_flag=False)
 41        # 清空self.plugins
 42        self.plugins.clear()
 43        # 销毁Api
 44        self.api.clear()
 45        ListenerManager.clear()
 46
 47    def loadPlugins(self, enter: bool = True):
 48        """
 49        Load all plugins in the directory.
 50        :param enter: bool (可选)是否调用`_enter`函数，默认为`True`
 51        :return: None
 52        """
 53        self.clearPlugins()
 54        for plugin in os.listdir(self.path):
 55            if plugin.startswith("__") or plugin.startswith("."):
 56                continue
 57            if plugin.endswith(".py"):
 58                plugin = plugin[:-3]
 59            logger.info(f"Loading plugin: `{plugin}`")
 60            if plugin in plugins_disabled:
 61                logger.warning(f"Plugin `{plugin}` is disabled")
 62                continue
 63            self.plugin_name = plugin
 64            self.loadPlugin(plugin, enter=enter)
 65
 66    def has(self, plugin: str):
 67        """
 68        Check if the plugin exists.
 69        :param plugin: str Plugin name
 70        :return: bool
 71        """
 72        return plugin in self.plugins
 73
 74    def hasApi(self, plugin: str):
 75        """
 76        Check if the plugin has an API.
 77        :param plugin: str Plugin name
 78        :return: bool
 79        """
 80        return plugin in self.api
 81
 82    def checkDependency(self, plugins: dict):
 83        """
 84        Plugins dependencies struct:
 85        ```
 86        {
 87            "plugin_name": {
 88                "upper": 1,
 89                "lower": 1
 90            }
 91        }
 92        ```
 93        :param plugins: dict Plugins dependencies
 94        :return: bool, str
 95        """
 96        for key, value in plugins:
 97            upper = value.get("upper")
 98            lower = value.get("lower")
 99            if self.plugins.get(key) is None:
100                return False, f"Plugin `{key}` not found"
101            meta_data: MetaData = self.plugins.get(key).meta_data
102            if upper is not None:
103                if meta_data.versionCode > upper:
104                    return False, f"Plugin `{key}` version is too high"
105            if lower is not None:
106                if meta_data.versionCode < lower:
107                    return False, f"Plugin `{key}` version is too low"
108        return True, "All dependencies are met"
109
110    def require(self, plugin: str):
111        """
112        Require the plugin API.
113        :param plugin: str Plugin name
114        :return: Api Object
115        """
116        if self.hasApi(plugin):
117            return self.api[plugin]
118        raise NoApiError(f"Plugin `{plugin}` not found or has no API")
119
120    def disable(self, plugin: str, pop_flag: bool=True):
121        """
122        Disable the plugin.
123        :param plugin: str Plugin name
124        :param pop_flag: bool (可选)是否从`self.api`中删除，默认为`True`
125        :return: None
126        """
127        if self.has(plugin):
128            try:
129                self.plugins[plugin]._exit()
130            except AttributeError:
131                pass
132            except Exception:
133                logger.warning(f"Plugin `{plugin}` failed to load `_exit` function: {traceback.format_exc()}")
134            if pop_flag:
135                self.plugins.pop(plugin)
136                if self.hasApi(plugin):
137                    self.api.pop(plugin)
138            ListenerManager.remove_listeners_by_plugin_name(plugin)
139            logger.info(f"Plugin `{plugin}` disabled")
140        else:
141            logger.warning(f"Plugin `{plugin}` not found")
142
143    def loadPlugin(self, plugin: str, enter: bool = True):
144        """
145        Load the plugin.
146        :param plugin: str Plugin name
147        :param enter: bool (可选)是否调用`_enter`函数，默认为`True`
148        :return: None
149        """
150        try:
151            plugin_distance = __import__(f"{plugin}")
152            logger.info(plugin_distance.meta_data)
153            if enter:
154                try:
155                    plugin_distance._enter()
156                except AttributeError:
157                    pass
158                except Exception:
159                    logger.warning(f"Plugin `{plugin}` failed to load `_enter` function: {traceback.format_exc()}")
160            try:
161                self.api[plugin] = plugin_distance.Api
162            except Exception:
163                pass
164            self.plugins[plugin] = plugin_distance
165            logger.info(f"Plugin `{plugin}` loaded")
166        except ImportError:
167            logger.error(f"Plugin `{plugin}` failed to load: {traceback.format_exc()}")
168        except Exception:
169            logger.error(f"Plugin `{plugin}` failed to load: {traceback.format_exc()}")
170
171    def enable(self, plugin: str):
172        """
173        Enable the plugin.
174        :param plugin: str Plugin name
175        :return: None
176        """
177        self.loadPlugin(plugin)
178
179    def getAllPlugins(self):
180        """
181        Get all plugins.
182        :return: dict
183        """
184        ret_dict: dict = {}
185        for key, value in self.plugins.items():
186            ret_dict[key] = value.meta_data.toDict()
187        return ret_dict
188
189
190if __name__ == "__main__":
191    PluginsManager()