✨ 安装/移除插件提供插件安装/卸载方法用于插件初始化 (#1654)

__version__

@@ -1 +1 @@
-__version__: v0.2.2-4a8ef85
+__version__: v0.2.2-e08b388

zhenxun/builtin_plugins/plugin_store/data_source.py

@@ -10,6 +10,7 @@ from zhenxun.utils.http_utils import AsyncHttpx
 from zhenxun.models.plugin_info import PluginInfo
 from zhenxun.utils.github_utils import GithubUtils
 from zhenxun.utils.github_utils.models import RepoAPI
+from zhenxun.services.plugin_init import PluginInitManager
 from zhenxun.builtin_plugins.plugin_store.models import StorePluginInfo
 from zhenxun.utils.image_utils import RowStyle, BuildImage, ImageTemplate
 from zhenxun.builtin_plugins.auto_update.config import REQ_TXT_FILE_STRING
@@ -292,6 +293,7 @@ class ShopManage:
             shutil.rmtree(path)
         else:
             path.unlink()
+        await PluginInitManager.remove(f"zhenxun.{plugin_info.module_path}")
         return f"插件 {plugin_key} 移除成功! 重启后生效"
 
     @classmethod

zhenxun/services/plugin_init.py

@@ -0,0 +1,107 @@
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+
+import nonebot
+from pydantic import BaseModel
+from nonebot.utils import is_coroutine_callable
+
+from zhenxun.services.log import logger
+
+driver = nonebot.get_driver()
+
+PLUGINS_METHOD = []
+
+
+class PluginInit(ABC):
+    """
+    插件安装与卸载模块
+    """
+
+    def __init_subclass__(cls, **kwargs):
+        module_path = cls.__module__
+        install_func = getattr(cls, "install", None)
+        remove_func = getattr(cls, "remove", None)
+        if install_func or remove_func:
+            PluginInitManager.plugins[module_path] = PluginInitData(
+                module_path=module_path,
+                install=install_func,
+                remove=remove_func,
+                class_=cls,
+            )
+
+    @abstractmethod
+    async def install(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    async def remove(self):
+        raise NotImplementedError
+
+
+class PluginInitData(BaseModel):
+    module_path: str
+    """模块名"""
+    install: Callable | None
+    """安装方法"""
+    remove: Callable | None
+    """卸载方法"""
+    class_: type[PluginInit]
+    """类"""
+
+
+class PluginInitManager:
+    plugins: dict[str, PluginInitData] = {}  # noqa: RUF012
+
+    @classmethod
+    async def install_all(cls):
+        """运行所有插件安装方法"""
+        if cls.plugins:
+            for module_path, model in cls.plugins.items():
+                if model.install:
+                    class_ = model.class_()
+                    try:
+                        logger.debug(f"开始执行: {module_path}:install 方法")
+                        if is_coroutine_callable(class_.install):
+                            await class_.install()
+                        else:
+                            class_.install()  # type: ignore
+                            logger.debug(f"执行: {module_path}:install 完成")
+                    except Exception as e:
+                        logger.error(f"执行: {module_path}:install 失败", e=e)
+
+    @classmethod
+    async def install(cls, module_path: str):
+        """运行指定插件安装方法"""
+        if model := cls.plugins.get(module_path):
+            if model.install:
+                class_ = model.class_()
+                try:
+                    logger.debug(f"开始执行: {module_path}:install 方法")
+                    if is_coroutine_callable(class_.install):
+                        await class_.install()
+                    else:
+                        class_.install()  # type: ignore
+                        logger.debug(f"执行: {module_path}:install 完成")
+                except Exception as e:
+                    logger.error(f"执行: {module_path}:install 失败", e=e)
+
+    @classmethod
+    async def remove(cls, module_path: str):
+        """运行指定插件安装方法"""
+        if model := cls.plugins.get(module_path):
+            if model.remove:
+                class_ = model.class_()
+                try:
+                    logger.debug(f"开始执行: {module_path}:remove 方法")
+                    if is_coroutine_callable(class_.remove):
+                        await class_.remove()
+                    else:
+                        class_.remove()  # type: ignore
+                        logger.debug(f"执行: {module_path}:remove 完成")
+                except Exception as e:
+                    logger.error(f"执行: {module_path}:remove 失败", e=e)
+
+
+@driver.on_startup
+async def _():
+    await PluginInitManager.install_all()