添加.update指令 配合.m reboot完成更新

src/plugins/DicePP/adapter/nonebot_adapter.py

@@ -32,7 +32,7 @@
 except ValueError:
     dice_log("DPP API is not amounted because NoneBot has not been initialized")
 
-command_matcher = on_message()
+command_matcher = on_message(block=False)
 notice_matcher = on_notice()
 request_matcher = on_request()
 

src/plugins/DicePP/module/misc/__init__.py

@@ -1,3 +1,4 @@
 from .jrrp_command import JrrpCommand
 from .dnd_command import UtilsDNDCommand
 from .statistics_cmd import StatisticsCommand
+from .update_command import UpdateCommand

src/plugins/DicePP/module/misc/update_command.py

@@ -0,0 +1,133 @@
+import os
+from typing import List, Tuple, Any, Literal, Callable, Optional
+
+from core.bot import Bot
+from core.command.const import *
+from core.command import UserCommandBase, custom_user_command
+from core.command import BotCommandBase, BotSendMsgCommand
+from core.communication import MessageMetaData, PrivateMessagePort, GroupMessagePort
+from core.config import PROJECT_PATH
+from utils.dicegit import GitRepository
+
+BOT_ROOT_PATH = os.path.dirname(os.path.dirname(os.path.dirname(PROJECT_PATH)))
+CFG_UPDATE_SOURCE = "update_source"
+CFG_DAILY_CHECK = "update_daily_check"
+LOC_UPDATE = "update"
+
+# DicePP Source
+dpp_github_path = "https://github.com/pear-studio/nonebot-dicepp.git"
+dpp_gitee_path = "https://gitee.com/pear_studio/nonebot-dicepp.git"
+
+
+@custom_user_command(readable_name="升级检查", priority=DPP_COMMAND_PRIORITY_DEFAULT)
+class UpdateCommand(UserCommandBase):
+
+    def __init__(self, bot: Bot):
+        super().__init__(bot)
+        bot.cfg_helper.register_config(CFG_UPDATE_SOURCE, "gitee", "可自定义的更新源, 目前支持gitee(推荐国内使用)与github.")
+        bot.cfg_helper.register_config(CFG_DAILY_CHECK, "0", "每日更新检查, 1为检查, 0为不检查.")
+
+        self.update_source: Literal["gitee", "github"] = "gitee"
+        self.daily_check: bool = False
+        self.git_repo = None
+        self.master_list = []
+
+    def delay_init(self) -> List[str]:
+        try:
+            self.update_source = self.bot.cfg_helper.get_config(key=CFG_UPDATE_SOURCE)[0].lower()
+            assert self.update_source in ["gitee", "github"]
+        except (AssertionError, IndexError):
+            self.update_source = "gitee"
+        try:
+            daily_check_val = int(self.bot.cfg_helper.get_config(key=CFG_DAILY_CHECK)[0])
+            self.daily_check = (daily_check_val != 0)
+        except (ValueError, IndexError):
+            self.daily_check = False
+        try:
+            self.master_list = self.bot.get_master_ids()
+        finally:
+            pass
+        return []
+
+    def can_process_msg(self, msg_str: str, meta: MessageMetaData) -> Tuple[bool, bool, Any]:
+        should_proc: bool = False
+        should_pass: bool = False
+        if meta.user_id not in self.master_list[0]:
+            return should_proc, should_pass, None
+
+        should_proc: bool = msg_str.startswith(".update")
+        return should_proc, should_pass, msg_str[7:].strip()
+
+    def process_msg(self, msg_str: str, meta: MessageMetaData, hint: Any) -> List[BotCommandBase]:
+        port = GroupMessagePort(meta.group_id) if meta.group_id else PrivateMessagePort(meta.user_id)
+        # 解析语句
+        arg_str = hint
+        feedback = ""
+
+        if self.update_source.lower() == "github":
+            git_path = dpp_github_path
+        elif self.update_source.lower() == "gitee":
+            git_path = dpp_gitee_path
+        else:
+            feedback = "初始化Git仓库失败, 请检查您的配置项是否正确。"
+            return [BotSendMsgCommand(self.bot.account, feedback, [port])]
+        try:
+            self.git_repo = GitRepository(BOT_ROOT_PATH, git_path, self.update_source)
+        except AssertionError:
+            raise NotImplementedError("未能成功初始化git仓库。")
+
+        if arg_str == "更新":
+            self.bot.register_task(self.get_update, is_async=False, timeout=0, timeout_callback=self.lose)
+        elif arg_str == "初始化":
+            self.bot.register_task(self.refresh, is_async=False, timeout=0, timeout_callback=self.lose)
+        elif not arg_str:
+            self.bot.register_task(self.update, is_async=False, timeout=0, timeout_callback=self.lose)
+        else:
+            feedback: str = "请输入.help update查看指令说明"
+        if feedback is None:
+            return []
+        return [BotSendMsgCommand(self.bot.account, feedback, [port])]
+
+    def tick_daily(self) -> List[BotCommandBase]:
+        if self.bot.cfg_helper.get_config(CFG_DAILY_CHECK) == 0:
+            return []
+        if self.update_source.lower() == "github":
+            git_path = dpp_github_path
+        elif self.update_source.lower() == "gitee":
+            git_path = dpp_gitee_path
+        else:
+            feedback: str = "初始化Git仓库失败, 请检查您的配置项是否正确。"
+            return [BotSendMsgCommand(self.bot.account, feedback, [PrivateMessagePort(self.master_list[0])])]
+        try:
+            self.git_repo = GitRepository(BOT_ROOT_PATH, git_path, self.update_source)
+        except AssertionError:
+            raise NotImplementedError("未能成功初始化git仓库。")
+        self.bot.register_task(self.update, is_async=False, timeout=0, timeout_callback=self.lose)
+        self.bot.register_task(self.is_dirty_check, is_async=False, timeout=0, timeout_callback=self.lose)
+        return []
+
+    def update(self) -> List[BotCommandBase]:
+        return [BotSendMsgCommand(self.bot.account, self.git_repo.get_update(), [PrivateMessagePort(self.master_list[0])])]
+
+    def get_update(self) -> List[BotCommandBase]:
+        return [BotSendMsgCommand(self.bot.account, self.git_repo.update(), [PrivateMessagePort(self.master_list[0])])]
+
+    def refresh(self) -> List[BotCommandBase]:
+        return [BotSendMsgCommand(self.bot.account, self.git_repo.refresh(), [PrivateMessagePort(self.master_list[0])])]
+
+    def is_dirty_check(self) -> List[BotCommandBase]:
+        return [BotSendMsgCommand(self.bot.account, self.git_repo.is_dirty_check(), [PrivateMessagePort(self.master_list[0])])]
+
+    def lose(self) -> List[BotCommandBase]:
+        return [BotSendMsgCommand(self.bot.account, "失败.", [PrivateMessagePort(self.master_list[0])])]
+
+    def get_help(self, keyword: str, meta: MessageMetaData) -> str:
+        if keyword == "update":  # help后的接着的内容
+            feedback: str = ".update     # 手动检查更新\n" \
+                            ".update 更新 # 下载并应用更新(需重启生效)\n" \
+                            ".update 初始化 # 恢复被修改的源代码\n"
+            return feedback
+        return ""
+
+    def get_description(self) -> str:
+        return ".update 更新相关指令"  # help指令中返回的内容

src/plugins/DicePP/utils/dicegit.py

@@ -0,0 +1,85 @@
+from typing import Union
+from git import InvalidGitRepositoryError, NoSuchPathError, GitCommandError, Repo
+from utils.logger import dice_log
+from urllib.parse import quote
+
+IS_NEWEST = False
+
+
+def is_network_error(err_str: str) -> bool:
+    return "errno 10054" in err_str or "Timed out" in err_str or "Could not resolve host" in err_str
+
+
+class GitRepository:
+    """
+    git仓库管理
+    """
+
+    def __init__(self, local_path, repo_url, update_source):
+        try:
+            import os
+            import git
+        except ImportError as e:
+            raise AssertionError(f"当前Python环境缺少必要库\n{e}")
+
+        try:
+            assert git.GIT_OK
+        except AssertionError:
+            git_path = os.path.abspath("../PortableGit/bin/git.exe")
+            assert os.path.exists(git_path)
+            dice_log(f"找不到系统已安装的Git工具, 使用PortableGit: {git_path}")
+            git.refresh(git_path)
+        assert git.GIT_OK, "Git不可用..."
+        self.local_path = local_path
+        self.repo_url = repo_url
+        self.update_source = update_source
+        self.repo = self.get_git_repo()
+
+    def is_dirty_check(self) -> str:
+        if self.repo.is_dirty() and self.repo.head.name == "HEAD" and self.repo.remote().name == "origin":
+            return "检测到DicePP源码被修改.若需清除该修改,请键入 .update 初始化 以清除其它更改."
+        else:
+            return ""
+
+    def update(self) -> str:
+        marking = self.get_update()
+        if marking == "已是最新. ":
+            return "已是最新."
+        try:
+            self.repo.git.fetch(self.repo_url + " " + "refs/heads/master:refs/heads/origin/master")
+        except ValueError as e:
+            return str(e)
+        except GitCommandError as e:
+            if e.stderr and is_network_error(e.stderr):
+                return "网络连接异常, 请确认没有开启VPN或代理服务, 并再次重试"
+            return str(e)
+        try:
+            self.repo.git.merge("origin/master")
+        except GitCommandError as e:
+            return str(e)
+        return "更新完成,请输入.m reboot重启bot以应用更新."
+
+    def get_update(self) -> str:
+        try:
+            c: str = self.repo.git.log("master..origin/master", "-1", "--pretty=hash:%H\n更新内容:%s")
+        except Exception as e:
+            return "检查更新失败. 原因: \n" + str(e)
+        if c:
+            return "检测到更新, 内容如下: \n" + str(c)
+        return "已是最新. "
+
+    def refresh(self) -> str:
+        try:
+            self.repo.git.refresh()
+        except Exception as e:
+            return "初始化代码失败：" + str(e)
+        return "已成功初始化DicePP代码."
+
+    def get_git_repo(self) -> Union[Repo, str]:
+        try:
+            git_repo = Repo(self.local_path)
+        except InvalidGitRepositoryError as e:
+            return "git仓库初始化失败, 原因如下: \n" + str(e)
+        except NoSuchPathError as e:
+            return "git仓库初始化失败, 原因如下: \n" + str(e)
+        return git_repo