feat: 基于元数据记录的插件更新实现

pkg/plugin/host.py

@@ -7,6 +7,7 @@ import pkgutil
 import sys
 import shutil
 import traceback
+import time
 import re
 
 import pkg.utils.updater as updater
@@ -15,6 +16,7 @@ import pkg.plugin.switch as switch
 import pkg.plugin.settings as settings
 import pkg.qqbot.adapter as msadapter
 import pkg.utils.network as network
+import pkg.plugin.metadata as metadata
 
 from mirai import Mirai
 import requests
@@ -116,10 +118,15 @@ def load_plugins():
     # 加载插件顺序
     settings.load_settings()
 
+    logging.debug("registered plugins: {}".format(__plugins__))
+
     # 输出已注册的内容函数列表
     logging.debug("registered content functions: {}".format(__callable_functions__))
     logging.debug("function instance map: {}".format(__function_inst_map__))
 
+    # 迁移插件源地址记录
+    metadata.do_plugin_git_repo_migrate()
+
 
 def initialize_plugins():
     """初始化插件"""
@@ -170,13 +177,15 @@ def get_github_plugin_repo_label(repo_url: str) -> list[str]:
         return None
 
 
-def download_plugin_source_coder(repo_url: str, target_path: str):
+def download_plugin_source_code(repo_url: str, target_path: str) -> str:
     """下载插件源码"""
     # 检查源类型
     
     # 提取 username/repo , 正则表达式
     repo =  get_github_plugin_repo_label(repo_url)
 
+    target_path += repo[1]
+
     if repo is not None: # github
         logging.info("从 GitHub 下载插件源码...")
 
@@ -225,28 +234,29 @@ def download_plugin_source_coder(repo_url: str, target_path: str):
         logging.info("解压完成")
     else:
         raise Exception("暂不支持的源类型，请使用 GitHub 仓库发行插件。")
+    
+    return repo[1]
+
+
+def check_requirements(path: str):
+    # 检查此目录是否包含requirements.txt
+    if os.path.exists(path+"/requirements.txt"):
+        logging.info("检测到requirements.txt，正在安装依赖")
+        import pkg.utils.pkgmgr
+        pkg.utils.pkgmgr.install_requirements(path+"/requirements.txt")
+
+        import pkg.utils.log as log
+        log.reset_logging()
 
 
 def install_plugin(repo_url: str):
     """安装插件，从git储存库获取并解决依赖"""
 
-    repo_label = None
+    repo_label = download_plugin_source_code(repo_url, "plugins/")
 
-    repo_label = get_github_plugin_repo_label(repo_url)
+    check_requirements("plugins/"+repo_label)
 
-    if repo_label is None:
+    metadata.set_plugin_metadata(repo_label, repo_url, int(time.time()), "HEAD")
-        raise Exception("暂不支持的源类型，请使用 GitHub 仓库发行插件。")
-
-    download_plugin_source_coder(repo_url, "plugins/"+repo_label[1])
-
-    # 检查此目录是否包含requirements.txt
-    if os.path.exists("plugins/"+repo_label[1]+"/requirements.txt"):
-        logging.info("检测到requirements.txt，正在安装依赖")
-        import pkg.utils.pkgmgr
-        pkg.utils.pkgmgr.install_requirements("plugins/"+repo_label[1]+"/requirements.txt")
-
-        import pkg.utils.log as log
-        log.reset_logging()
 
 
 def uninstall_plugin(plugin_name: str) -> str:
@@ -268,39 +278,43 @@ def uninstall_plugin(plugin_name: str) -> str:
 def update_plugin(plugin_name: str):
     """更新插件"""
     # 检查是否有远程地址记录
-    target_plugin_dir = "plugins/" + __plugins__[plugin_name]['path'].replace("\\", "/").split("plugins/")[1].split("/")[0]
+    plugin_path_name = get_plugin_path_name_by_plugin_name(plugin_name)
 
-    remote_url = updater.get_remote_url(target_plugin_dir)
+    meta = metadata.get_plugin_metadata(plugin_path_name)
+
+    if meta == {}:
+        raise Exception("没有此插件元数据信息，无法更新")
+
+    remote_url = meta['source']
     if remote_url == "https://github.com/RockChinQ/QChatGPT" or remote_url == "https://gitee.com/RockChin/QChatGPT" \
         or remote_url == "" or remote_url is None or remote_url == "http://github.com/RockChinQ/QChatGPT" or remote_url == "http://gitee.com/RockChin/QChatGPT":
         raise Exception("插件没有远程地址记录，无法更新")
     
-    # 把远程clone到temp/plugins/update/插件名
+    # 重新安装插件
-    logging.info("克隆插件储存库: {}".format(remote_url))
+    logging.info("正在重新安装插件以进行更新...")
 
-    from dulwich import porcelain
+    install_plugin(remote_url)
-    clone_target_dir = "temp/plugins/update/"+target_plugin_dir.split("/")[-1]+"/"
 
-    if os.path.exists(clone_target_dir):
-        shutil.rmtree(clone_target_dir)
 
-    if not os.path.exists(clone_target_dir):
+def get_plugin_name_by_path_name(plugin_path_name: str) -> str:
-        os.makedirs(clone_target_dir)
+    for k, v in __plugins__.items():
-    repo = porcelain.clone(remote_url, clone_target_dir, checkout=True)
+        if v['path'] == "plugins/"+plugin_path_name+"/main.py":
+            return k
+    return None
 
-    # 检查此目录是否包含requirements.txt
-    if os.path.exists(clone_target_dir+"requirements.txt"):
-        logging.info("检测到requirements.txt，正在安装依赖")
-        import pkg.utils.pkgmgr
-        pkg.utils.pkgmgr.install_requirements(clone_target_dir+"requirements.txt")
 
-        import pkg.utils.log as log
+def get_plugin_path_name_by_plugin_name(plugin_name: str) -> str:
-        log.reset_logging()
+    if plugin_name not in __plugins__:
+        return None
+    
+    plugin_main_module_path = __plugins__[plugin_name]['path']
 
-    # 将temp/plugins/update/插件名 覆盖到 plugins/插件名
+    plugin_main_module_path = plugin_main_module_path.replace("\\", "/")
-    shutil.rmtree(target_plugin_dir)
+
+    spt = plugin_main_module_path.split("/")
+
+    return spt[1]
 
-    shutil.copytree(clone_target_dir, target_plugin_dir)
 
 class EventContext:
     """事件上下文"""

pkg/plugin/metadata.py

@@ -0,0 +1,87 @@
+import os
+import shutil
+import json
+import time
+
+import dulwich.errors as dulwich_err
+
+from ..utils import updater
+
+
+def read_metadata_file() -> dict:
+    # 读取 plugins/metadata.json 文件
+    if not os.path.exists('plugins/metadata.json'):
+        return {}
+    with open('plugins/metadata.json', 'r') as f:
+        return json.load(f)
+
+
+def write_metadata_file(metadata: dict):
+    if not os.path.exists('plugins'):
+        os.mkdir('plugins')
+
+    with open('plugins/metadata.json', 'w') as f:
+        json.dump(metadata, f, indent=4, ensure_ascii=False)
+
+
+def do_plugin_git_repo_migrate():
+    # 仅在 plugins/metadata.json 不存在时执行
+    if os.path.exists('plugins/metadata.json'):
+        return
+
+    metadata = read_metadata_file()
+
+    # 遍历 plugins 下所有目录，获取目录的git远程地址
+    for plugin_name in os.listdir('plugins'):
+        plugin_path = os.path.join('plugins', plugin_name)
+        if not os.path.isdir(plugin_path):
+            continue
+        
+        remote_url = None
+        try:
+            remote_url = updater.get_remote_url(plugin_path)
+        except dulwich_err.NotGitRepository:
+            continue
+        if remote_url == "https://github.com/RockChinQ/QChatGPT" or remote_url == "https://gitee.com/RockChin/QChatGPT" \
+            or remote_url == "" or remote_url is None or remote_url == "http://github.com/RockChinQ/QChatGPT" or remote_url == "http://gitee.com/RockChin/QChatGPT":
+            continue
+
+        from . import host
+
+        if plugin_name not in metadata:
+            metadata[plugin_name] = {
+                'source': remote_url,
+                'install_timestamp': int(time.time()),
+                'ref': 'HEAD',
+            }
+
+    write_metadata_file(metadata)
+    
+
+def set_plugin_metadata(
+    plugin_name: str,
+    source: str,
+    install_timestamp: int,
+    ref: str,
+):
+    metadata = read_metadata_file()
+    metadata[plugin_name] = {
+        'source': source,
+        'install_timestamp': install_timestamp,
+        'ref': ref,
+    }
+    write_metadata_file(metadata)
+
+
+def remove_plugin_metadata(plugin_name: str):
+    metadata = read_metadata_file()
+    if plugin_name in metadata:
+        del metadata[plugin_name]
+        write_metadata_file(metadata)
+
+
+def get_plugin_metadata(plugin_name: str) -> dict:
+    metadata = read_metadata_file()
+    if plugin_name in metadata:
+        return metadata[plugin_name]
+    return {}

pkg/qqbot/cmds/plugin/plugin.py

@@ -84,7 +84,7 @@ class PluginGetCommand(AbstractCommandNode):
 @AbstractCommandNode.register(
     parent=PluginCommand,
     name="update",
-    description="更新所有插件",
+    description="更新插件",
     usage="!plugin update",
     aliases=[],
     privilege=2
@@ -110,7 +110,9 @@ class PluginUpdateCommand(AbstractCommandNode):
                             plugin_host.update_plugin(key)
                             updated.append(key)
                     else:
-                        if ctx.crt_params[0] in plugin_list:
+                        plugin_path_name = plugin_host.get_plugin_path_name_by_plugin_name(ctx.crt_params[0])
+
+                        if ctx.crt_params[0] is not None:
                             plugin_host.update_plugin(ctx.crt_params[0])
                             updated.append(ctx.crt_params[0])
                         else:
@@ -119,7 +121,7 @@ class PluginUpdateCommand(AbstractCommandNode):
                     pkg.utils.context.get_qqbot_manager().notify_admin("已更新插件: {}, 请发送 !reload 重载插件".format(", ".join(updated)))
                 except Exception as e:
                     logging.error("插件更新失败:{}".format(e))
-                    pkg.utils.context.get_qqbot_manager().notify_admin("插件更新失败:{} 请尝试手动更新插件".format(e))
+                    pkg.utils.context.get_qqbot_manager().notify_admin("插件更新失败:{} 请使用 !plugin 命令确认插件名称或尝试手动更新插件".format(e))
 
             reply = ["[bot]正在更新插件，请勿重复发起..."]
             threading.Thread(target=closure).start()