mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-14 21:52:56 +08:00
ee699fb345
* fix(plugin_store): 修复插件商店的安装与卸载逻辑 - 优化了插件安装、更新和移除的逻辑 - 调整了插件路径的处理方式，支持更灵活的安装位置 - 重构了 `install_plugin_with_repo` 方法，使用 `StorePluginInfo` 对象作为参数 - 修复了一些潜在的路径问题和模块命名问题 * refactor(zhenxun): 优化插件信息获取逻辑 - 将 PluginInfo.get_or_none 替换为 get_plugin 方法，简化插件信息获取逻辑 - 优化了插件移除操作中的插件信息获取流程 * refactor(zhenxun): 优化 sparse_checkout_clone 函数的实现 - 将 git 操作移至临时目录中执行，避免影响目标目录中的现有内容 - 简化了稀疏检出的配置和执行过程 - 改进了错误处理和回退逻辑 - 优化了文件移动和目录清理的操作 * 🐛 添加移除插件时二次查询 * ✨ plugin_info.get_plugin参数包含plugin_type时无效过滤 --------- Co-authored-by: HibiKier <45528451+HibiKier@users.noreply.github.com>
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
from abc import ABC, abstractmethod
|
|
from collections.abc import Callable
|
|
|
|
import nonebot
|
|
from nonebot.utils import is_coroutine_callable
|
|
from pydantic import BaseModel
|
|
|
|
from zhenxun.services.log import logger
|
|
from zhenxun.utils.manager.priority_manager import PriorityLifecycle
|
|
|
|
driver = nonebot.get_driver()
|
|
|
|
|
|
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 not cls.plugins:
|
|
return
|
|
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)
|
|
|
|
|
|
@PriorityLifecycle.on_startup(priority=5)
|
|
async def _():
|
|
await PluginInitManager.install_all()
|