feat(config/1.4): 支持重试部分命令，并允许指定命令重试间隔

.github/workflows/test.yaml

@@ -232,8 +232,8 @@ jobs:
           cd "$GITHUB_WORKSPACE"
           echo "[DEBUG] 运行 python ./Release/src/sundry.py sync"
           python ./Release/src/sundry.py sync || exit_code=$?
-          if [ "${exit_code:-0}" -eq 1 ]; then
-            # 403
+          if [ "${exit_code:-0}" -eq 128 ]; then
+            # fatal: could not read Username for 'https://github.com': No such device or address
             exit 0
           fi
           exit ${exit_code:-0}

src/function/constant/general.py

@@ -4,6 +4,8 @@
 
 from typing import Final
 
+from function.maintain.config import 读取配置
+
 UNEXPECTED_TYPES: Final[set[str]] = {"xml", "json", "html"}
 """
 InstallerUrl 常见的意外响应类型。
@@ -29,8 +31,52 @@
 > This PR is automatically created by [Sundry](https://github.com/DuckDuckStudio/Sundry/) version x.x.x 🚀.
 """
 
+def _get_retry_interval() -> int:
+    """
+    尝试从 git.retry_interval 配置获取 Git 操作重试间隔秒数，
+    如果没能获取到返回默认值 50 秒。
+
+    Returns:
+        int: Git 操作重试间隔秒数，默认为 50 秒。
+    """
+
+    _config_value = 读取配置("git.retry_interval")
+    if not isinstance(_config_value, int):
+        _config_value = 50
+
+    return _config_value
+
+RETRY_INTERVAL: int
+"""
+Git 操作重试间隔秒数
+
+默认为 50 秒
+"""
+
 # TODO 通过配置文件设置请求超时时间
 REQUEST_TIMEOUT: Final = 30
 """
 请求超时时间（秒）
 """
+
+def __getattr__(name: str) -> int:
+    """
+    初始化需要读取配置的常量
+
+    - RETRY_INTERVAL
+
+    Args:
+        name: 读取的属性的名称
+
+    Returns:
+        int: 初始化后的值
+
+    Raises:
+        AttributeError: 意外的名称
+    """
+
+    if name == "RETRY_INTERVAL":
+        _retry_interval = _get_retry_interval()
+        globals()["RETRY_INTERVAL"] = _retry_interval
+        return _retry_interval
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/function/maintain/config.py

@@ -8,7 +8,6 @@
 from catfood.functions.print import MSHead
 from colorama import Fore
 
-from function.constant.general import REQUEST_TIMEOUT
 from function.constant.paths import CONFIG_FILE_PATH
 
 type 所有配置项 = Literal[
@@ -17,7 +16,7 @@
     "debug",
     "paths.winget-pkgs", "paths.winget-tools",
     "repos.winget-pkgs", "repos.winget-tools",
-    "git.signature",
+    "git.retry_interval", "git.signature",
     "github.pr.maintainer_can_modify", "github.pr.mention_self_when_reviewer",
     "github.token",
     "tools.autoremove.open_in_browser",
@@ -41,6 +40,7 @@ class 配置信息:
             "winget-tools": ""
         },
         "git": {
+            "retry_interval": 50,
             "signature": False
         },
         "github": {
@@ -96,7 +96,7 @@ class 配置信息:
     所在位置: str = CONFIG_FILE_PATH
     """等同于 `from function.constant.paths import CONFIG_FILE_PATH`。"""
 
-def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool) -> str | None:
+def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool | int) -> str | None:
     """
     [验证配置]
 
@@ -115,7 +115,7 @@ def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool) -> str
     if 配置项.startswith("paths.") and isinstance(配置值, str):
         配置值 = os.path.normpath(配置值)
         if (not os.path.exists(配置值)):
-            return f"配置文件中的目录 {Fore.BLUE}{配置值}{Fore.RESET} 不存在"
+            return f"目录 {Fore.BLUE}{配置值}{Fore.RESET} 不存在"
         return None
 
     elif 配置项.startswith("repos.") and isinstance(配置值, str):
@@ -124,6 +124,7 @@ def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool) -> str
             owner, repo = parts
             api_url = f"https://api.github.com/repos/{owner}/{repo}"
             try:
+                from function.constant.general import REQUEST_TIMEOUT
                 response = requests.head(api_url, timeout=REQUEST_TIMEOUT)
                 if response.status_code < 400:
                     return None
@@ -145,6 +146,9 @@ def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool) -> str
     elif (配置项 == "github.token") and (配置值 not in ["glm", "komac", "env"]):
         return "未知的 Token 读取源"
 
+    elif (配置项 == "git.retry_interval") and (not isinstance(配置值, int)):
+        return f"应是整数，但实际是 {Fore.BLUE}{type(配置值)}{Fore.RESET}"
+
     elif (配置项 == "version"):
         if isinstance(配置值, str):
             if not (
@@ -166,7 +170,7 @@ def 验证配置(配置项: 所有配置项 | str, 配置值: str | bool) -> str
     else:
         return None
 
-def 读取配置(配置项: 所有配置项 | str, 静默: bool = False) -> None | str | tuple[str, str] | bool:
+def 读取配置(配置项: 所有配置项 | str, 静默: bool = False) -> None | str | tuple[str, str] | bool | int:
     """
     [验证/转换后的配置值]
 
@@ -181,7 +185,7 @@ def 读取配置(配置项: 所有配置项 | str, 静默: bool = False) -> None
         静默 = True
 
     try:
-        配置值: str | bool | None = 读取配置项(配置项, 静默)
+        配置值: str | bool | int | None = 读取配置项(配置项, 静默)
 
         if 配置值 is None:
             return None
@@ -216,11 +220,11 @@ def 读取配置(配置项: 所有配置项 | str, 静默: bool = False) -> None
             print(f"{MSHead.Error} 读取配置 {配置项} 失败: {Fore.RED}{e}{Fore.RESET}")
         return None
 
-def 读取配置项(配置项: 所有配置项 | str, 静默: bool = False) -> str | bool | None:
+def 读取配置项(配置项: 所有配置项 | str, 静默: bool = False) -> str | bool | int | None:
     """
     [原始字符串]
     读取指定配置项的值，并返回配置项值。
-    预期返回非空 str 或 bool，读取失败返回 None。
+    读取失败返回 None。
     """
 
     if os.path.exists(配置信息.所在位置):
@@ -271,14 +275,15 @@ def 获取配置schema(版本: str) -> dict[str, Any] | None:
     except Exception as e:
         try:
             print(f"{MSHead.Warning} 获取配置文件 schema 失败 ({e})，通过 https://duckduckstudio.github.io/yazicbs.github.io/Tools/Sundry/config/schema/{版本}.json 重试...")
+            from function.constant.general import REQUEST_TIMEOUT
             响应 = requests.get(f"https://duckduckstudio.github.io/yazicbs.github.io/Tools/Sundry/config/schema/{版本}.json", timeout=REQUEST_TIMEOUT)
             响应.raise_for_status()
             print(f"{MSHead.Information} 获取配置文件 schema 成功")
             return 响应.json()
         except Exception:
             return None
 
-def 转换配置值(配置项: 所有配置项 | str, 配置值: str) -> str | bool:
+def 转换配置值(配置项: 所有配置项 | str, 配置值: str) -> str | bool | int:
     """
     尝试将输入的配置值转换为符合配置文件要求的格式，如 y → true
     遇到无法转换的会 raise OperationFailed(原因)，我假设调用这个函数的地方会用红色显示错误消息
@@ -302,6 +307,14 @@ def 转换配置值(配置项: 所有配置项 | str, 配置值: str) -> str | b
                 return False
             else:
                 raise OperationFailed(f"{Fore.BLUE}{配置值}{Fore.RED} 不能代表是或否，请使用 y / n")
+    elif 配置项 in ("git.retry_interval"):
+        if not 配置值:
+            配置值 = "50"
+
+        try:
+            return int(配置值)
+        except ValueError as e:
+            raise OperationFailed("指定的配置值不是整数") from e
     else:
         if not 配置值:
             # 使用默认配置值

src/tools/ignore.py

@@ -6,9 +6,10 @@
 import requests
 from catfood.functions.files import open_file
 from catfood.functions.print import MSHead
+from catfood.functions.terminal import runCommand
 from colorama import Fore
 
-from function.constant.general import REQUEST_TIMEOUT
+from function.constant.general import REQUEST_TIMEOUT, RETRY_INTERVAL
 from function.git.format import branchName
 from function.github.token import read_token
 from function.maintain.config import 读取配置
@@ -194,7 +195,8 @@ def main(args: list[str]) -> int:
         print(f"{MSHead.Information} 开始同步...")
         subprocess.run(["git", "checkout", "main"], check=True)
         print(f"{MSHead.Information}     已签出到 main 分支")
-        subprocess.run(["git", "pull"], check=True)
+        if e := runCommand(["git", "pull"], retry=RETRY_INTERVAL):
+            return 1
         print(f"{MSHead.Information}     已拉取远程修改")
     except subprocess.CalledProcessError as e:
         print(f"{MSHead.Error} 同步失败:\n{Fore.RED}{e}{Fore.RESET}")
@@ -233,7 +235,8 @@ def main(args: list[str]) -> int:
         print(f"{MSHead.Information} 开始提交和推送...")
         subprocess.run(["git", "add", 检测程序], check=True)
         subprocess.run(["git", "commit", "-m", f"[Auto] 自动忽略追加 - {忽略字段}"], check=True)
-        subprocess.run(["git", "push", "--set-upstream", "origin", 新分支名], check=True)
+        if e := runCommand(["git", "push", "--set-upstream", "origin", 新分支名], retry=RETRY_INTERVAL):
+            return 1
         print(f"{MSHead.Success} 成功推送到远程")
         if not 创建拉取请求(新分支名, owner, 忽略字段):
             失败 = True
@@ -249,7 +252,8 @@ def main(args: list[str]) -> int:
         print(f"{MSHead.Information} 开始提交和推送...")
         subprocess.run(["git", "add", 检测程序], check=True)
         subprocess.run(["git", "commit", "-m", f"[Auto] 自动忽略移除 - {格式化忽略字段}"], check=True)
-        subprocess.run(["git", "push", "--set-upstream", "origin", 新分支名], check=True)
+        if e := runCommand(["git", "push", "--set-upstream", "origin", 新分支名], retry=RETRY_INTERVAL):
+            return 1
         print(f"{MSHead.Success} 成功推送到远程")
         if not 创建拉取请求(新分支名, owner, 忽略字段, 理由):
             失败 = True
@@ -262,7 +266,8 @@ def main(args: list[str]) -> int:
         print(f"{MSHead.Information} 开始提交和推送...")
         subprocess.run(["git", "add", 检测程序], check=True)
         subprocess.run(["git", "commit", "-m", "chore(checker): 更新忽略字段"], check=True)
-        subprocess.run(["git", "push", "--set-upstream", "origin", 新分支名], check=True)
+        if e := runCommand(["git", "push", "--set-upstream", "origin", 新分支名], retry=RETRY_INTERVAL):
+            return 1
         print(f"{MSHead.Success} 成功推送到远程")
         if not 创建拉取请求(新分支名, owner, 理由="edit"):
             失败 = True

src/tools/maintain/config.py

@@ -23,7 +23,7 @@
 )
 
 
-def 获取用户输入(配置项: str) -> str | bool:
+def 获取用户输入(配置项: str) -> str | bool | int:
     """
     获取用户输入的配置值，并返回转换后的配置值。
 
@@ -42,6 +42,7 @@ def 获取用户输入(配置项: str) -> str | bool:
         "repos.winget-pkgs": f"{MSHead.Question} 您的远程 winget-{Fore.YELLOW}pkgs{Fore.RESET} 仓库是什么 (owner/winget-pkgs): ",
         "repos.winget-tools": f"{MSHead.Question} 您的远程 winget-{Fore.YELLOW}tools{Fore.RESET} 仓库是什么 (owner/winget-tools): ",
         # git.*
+        "git.retry_interval": f"{MSHead.Question} 重试命令的间隔是? [间隔应为{Fore.GREEN}整数{Fore.RESET}，{Fore.RED}负数{Fore.RESET}为不重试，{Fore.YELLOW}零{Fore.RESET}为立即重试] (默认为{Fore.BLUE} 50 {Fore.RESET}秒): ",
         "git.signature": f"{MSHead.Question} 是否要为 Git 提交签名? (默认为{Fore.YELLOW}否{Fore.RESET}): ",
         # github.pr.*
         "github.pr.maintainer_can_modify": f"{MSHead.Question} 是否允许维护者修改您的 PR 内容? (默认为{Fore.YELLOW}否{Fore.RESET}): ",

src/tools/modify.py

@@ -12,9 +12,11 @@
 from catfood.functions.files import open_file
 from catfood.functions.github.api import get_github_token_owner
 from catfood.functions.print import MSHead
+from catfood.functions.terminal import runCommand
 from colorama import Fore
 
-from function.constant.general import PR_TOOL_NOTE, REQUEST_TIMEOUT
+import tools.sync
+from function.constant.general import PR_TOOL_NOTE, REQUEST_TIMEOUT, RETRY_INTERVAL
 from function.files.manifest import FormatManifest, 获取清单目录
 from function.git.format import branchName
 from function.github.token import read_token
@@ -247,9 +249,8 @@ def 修改版本(版本文件夹: str):
     新分支 = branchName(f"Modify-S-{包标识符}-{版本文件夹}-{int(time.time())}")
     print(f"  创建并切换到新分支: {新分支}")
     写入日志(f"  Create and checkout to a new branch: {新分支}")
-    subprocess.run(["git", "checkout", "master"], check=True) # 确保从 master 分支开始
-    subprocess.run(["git", "fetch", "upstream"], check=True) # 拉取上游修改
-    subprocess.run(["git", "rebase", "upstream/master"], check=True) # 变基上游修改
+    if e := tools.sync.main():
+        return e
     subprocess.run(["git", "checkout", "-b", 新分支], check=True) # 创建并切换到新的分支
 
     # 遍历该版本文件夹中的所有文件
@@ -325,8 +326,11 @@ def 修改版本(版本文件夹: str):
     # 推送更改到远程仓库
     print("  推送更改到远程仓库")
     写入日志("  Pushing changes to remote (origin) repository")
-    subprocess.run(["git", "push", "origin", 新分支], check=True)
-    print(f"    {Fore.GREEN}推送到远程成功: {新分支}")
+    if e := runCommand(["git", "push", "origin", 新分支], retry=RETRY_INTERVAL):
+        print(f"{MSHead.Error} 推送到远程失败: Git 返回退出代码 {e}")
+        return e
+    else:
+        print(f"    {Fore.GREEN}推送到远程成功: {新分支}")
     写入日志(f"    Successfully pushed to remote (origin): {新分支}")
 
     # 创建拉取请求

src/tools/prune.py

@@ -1,10 +1,10 @@
 import os
-import subprocess
 
-from catfood.exceptions.operation import OperationFailed
 from catfood.functions.print import MSHead
+from catfood.functions.terminal import runCommand
 from colorama import Fore
 
+from function.constant.general import RETRY_INTERVAL
 from function.maintain.config import 读取配置
 
 
@@ -13,23 +13,18 @@ def main() -> int:
         for 仓库 in ["winget-pkgs", "winget-tools"]:
             仓库路径 = 读取配置(f"paths.{仓库}")
             if isinstance(仓库路径, str):
-                清理远程(仓库, 仓库路径)
+                os.chdir(仓库路径)
+                print(f"清理 {Fore.BLUE}{仓库}{Fore.RESET} 仓库的远程已删除分支...")
+                if e := runCommand(["git", "remote", "prune", "origin"], retry=RETRY_INTERVAL):
+                    print(f"{MSHead.Error} 清理 {Fore.BLUE}{仓库}{Fore.RESET} 的远程已删除分支时出错: Git 返回退出代码 {e}")
+                    return e
+                else:
+                    print(f"{Fore.GREEN}✓{Fore.RESET} 清理完毕")
             else:
-                raise OperationFailed
+                raise ValueError
     except KeyboardInterrupt:
         print(f"{MSHead.Error} 已取消操作")
         return 1
-    except OperationFailed:
+    except ValueError:
         return 1
     return 0
-
-
-def 清理远程(仓库: str, 仓库路径: str):
-    try:
-        print(f"清理 {Fore.BLUE}{仓库}{Fore.RESET} 仓库的远程已删除分支...")
-        os.chdir(仓库路径)
-        subprocess.run(["git", "remote", "prune", "origin"], check=True)
-        print(f"{Fore.GREEN}✓{Fore.RESET} 清理完毕")
-    except subprocess.CalledProcessError as e:
-        print(f"{MSHead.Error} 清理 {Fore.BLUE}{仓库}{Fore.RESET} 的远程已删除分支时出错，git 返回 {Fore.BLUE}{e.returncode}{Fore.RESET}")
-        raise OperationFailed

src/tools/remove.py

@@ -12,10 +12,11 @@
 from catfood.exceptions.operation import OperationFailed
 from catfood.functions.github.api import get_github_token_owner
 from catfood.functions.print import MSHead
+from catfood.functions.terminal import runCommand
 from colorama import Fore
-from translate import Translator  # type: ignore
+from translate import Translator  # pyright: ignore[reportMissingTypeStubs]
 
-from function.constant.general import PR_TOOL_NOTE, REQUEST_TIMEOUT
+from function.constant.general import PR_TOOL_NOTE, REQUEST_TIMEOUT, RETRY_INTERVAL
 from function.files.manifest import 获取清单目录, 获取现有包版本
 from function.git.format import branchName
 from function.github.token import read_token
@@ -207,8 +208,11 @@ def main(args: list[str]) -> int:
         subprocess.run(["git", "commit", "-m", f"Remove version: {包标识符} version {包版本} (Auto)"], check=True)
     print(f"{Fore.BLUE}  已提交修改")
 
-    subprocess.run(["git", "push"], check=True)
-    print(f"{Fore.BLUE}  已推送修改")
+    if e := runCommand(["git", "push"], retry=RETRY_INTERVAL):
+        print(f"{MSHead.Error} 推送到远程失败: Git 返回退出代码 {e}")
+        return e
+    else:
+        print(f"{Fore.BLUE}  已推送修改")
 
     while (not 理由):
         理由 = input("移除此包版本的理由: ")