diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index bee030b3..0e975574 100644 --- a/.github/workflows/test.yaml +++ b/.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} diff --git a/src/function/constant/general.py b/src/function/constant/general.py index cbaad9d0..9e1ed62f 100644 --- a/src/function/constant/general.py +++ b/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}") diff --git a/src/function/maintain/config.py b/src/function/maintain/config.py index 8e70cffd..dc3f1ead 100644 --- a/src/function/maintain/config.py +++ b/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,6 +275,7 @@ 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 成功") @@ -278,7 +283,7 @@ def 获取配置schema(版本: str) -> dict[str, Any] | None: 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 配置值: # 使用默认配置值 diff --git a/src/tools/ignore.py b/src/tools/ignore.py index ab732ac4..04ddd2d8 100644 --- a/src/tools/ignore.py +++ b/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 diff --git a/src/tools/maintain/config.py b/src/tools/maintain/config.py index 7412a251..52a17c68 100644 --- a/src/tools/maintain/config.py +++ b/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}): ", diff --git a/src/tools/modify.py b/src/tools/modify.py index 50eaedba..9c8b6dc5 100644 --- a/src/tools/modify.py +++ b/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): {新分支}") # 创建拉取请求 diff --git a/src/tools/prune.py b/src/tools/prune.py index ee863d9f..809be0ca 100644 --- a/src/tools/prune.py +++ b/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 diff --git a/src/tools/remove.py b/src/tools/remove.py index 76cfeda1..f68fa39f 100644 --- a/src/tools/remove.py +++ b/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("移除此包版本的理由: ") diff --git a/src/tools/sync.py b/src/tools/sync.py index 825d08b6..eb096e70 100644 --- a/src/tools/sync.py +++ b/src/tools/sync.py @@ -3,8 +3,10 @@ from catfood.constant import YES 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 读取配置 @@ -14,62 +16,72 @@ def main() -> int: if not isinstance(winget_pkgs目录, str): return 1 - # 入口 os.chdir(winget_pkgs目录) + + # 签出 master try: - subprocess.run(["git", "checkout", "master"], check=True) # 确保从 master 分支开始 - print(f"{Fore.BLUE} 已签出到 master 分支") + subprocess.run(["git", "checkout", "master"], check=True) + print(f"{MSHead.Information} 已签出到 master 分支") except subprocess.CalledProcessError as e: print(f"{MSHead.Error} 签出到 master 分支失败:\n{Fore.RED}{e}{Fore.RESET}") return 1 - while True: - try: - subprocess.run(["git", "fetch", "upstream"], check=True) # 拉取上游修改 - print(f"{Fore.BLUE} 已获取上游修改") - break - except subprocess.CalledProcessError as e: - print(f"{MSHead.Error} 获取上游修改失败:\n{Fore.RED}{e}{Fore.RESET}") - if input(f"{MSHead.Question} 是否重试？(默认为{Fore.GREEN}是{Fore.RESET}): ").lower() not in (*YES, ""): - print(f"{MSHead.Message} 已取消操作") - return 1 + # 获取上游 + if e := runCommand(["git", "fetch", "upstream"], retry=RETRY_INTERVAL): + print(f"{MSHead.Error} 获取上游修改失败: Git 返回退出代码 {e}") + return e + else: + print(f"{MSHead.Information} 已获取上游修改") - try: - subprocess.run(["git", "fetch", "origin"], check=True) # 拉取远程修改 - print(f"{Fore.BLUE} 已获取远程修改") - except subprocess.CalledProcessError as e: - print(f"{MSHead.Warning} 拉取远程修改失败:\n{Fore.YELLOW}{e}{Fore.RESET}") - # 不影响...不影响 + # 获取远程 + if e := runCommand(["git", "fetch", "origin"], retry=RETRY_INTERVAL): + print(f"{MSHead.Error} 获取远程修改失败: Git 返回退出代码 {e}") + return e + else: + print(f"{MSHead.Information} 已获取远程修改") try: - subprocess.run(["git", "rebase", "upstream/master"], check=True) # 变基合并上游修改 - print(f"{Fore.BLUE} 已变基上游修改") + subprocess.run( + ["git", "rebase", "upstream/master"], check=True + ) # 变基合并上游修改 + print(f"{MSHead.Information} 已变基上游修改") except subprocess.CalledProcessError as e: print(f"{MSHead.Error} 变基上游修改失败:\n{Fore.RED}{e}{Fore.RESET}") - if input(f"{MSHead.Question} 是否尝试替换 master 分支？(默认为{Fore.YELLOW}否{Fore.RESET}): ").lower() in YES: + if ( + input( + f"{MSHead.Question} 是否尝试替换 master 分支？(默认为{Fore.YELLOW}否{Fore.RESET}): " + ).lower() + in YES + ): try: - subprocess.run(["git", "checkout", "upstream/master"], check=True) # 签出到上游 master 分支 + subprocess.run( + ["git", "checkout", "upstream/master"], check=True + ) # 签出到上游 master 分支 print(f"{Fore.BLUE} 已签出到上游 master 分支") - subprocess.run(["git", "branch", "-D", "master"], check=True) # 移除旧的 master 分支 + subprocess.run( + ["git", "branch", "-D", "master"], check=True + ) # 移除旧的 master 分支 print(f"{Fore.BLUE} 已移除旧 master 分支") - subprocess.run(["git", "switch", "-c", "master"], check=True) # 创建并签出到 master 分支 + subprocess.run( + ["git", "switch", "-c", "master"], check=True + ) # 创建并签出到 master 分支 print(f"{Fore.BLUE} 已创建并签出到 master 分支") except subprocess.CalledProcessError as e1: - print(f"{MSHead.Error} 替换 master 分支失败:\n{Fore.RED}{e1}{Fore.RESET}") + print( + f"{MSHead.Error} 替换 master 分支失败:\n{Fore.RED}{e1}{Fore.RESET}" + ) return 1 else: - print(f"{MSHead.Message} 已取消操作") - return 1 + raise KeyboardInterrupt from e - try: - # 推送 master - subprocess.run(["git", "push", "origin", "master"], check=True) - print(f"{Fore.BLUE} 已推送 master 分支") - except subprocess.CalledProcessError as e: - print(f"{MSHead.Error} 推送 master 分支失败:\n{Fore.RED}{e}{Fore.RESET}") - return 1 + # 推送到远程 + if e := runCommand(["git", "push", "origin", "master"], retry=RETRY_INTERVAL): + print(f"{MSHead.Error} 推送到远程失败: Git 返回退出代码 {e}") + return e + else: + print(f"{MSHead.Information} 已推送到远程") - print(f"{Fore.GREEN}✓{Fore.RESET} 同步完成") + print(f"{MSHead.Success} 同步完成") except KeyboardInterrupt: print(f"{MSHead.Error} 用户已取消操作") return 1 diff --git a/tests/function/constant/test_general.py b/tests/function/constant/test_general.py new file mode 100644 index 00000000..ed2a1522 --- /dev/null +++ b/tests/function/constant/test_general.py @@ -0,0 +1,5 @@ +from function.constant.general import RETRY_INTERVAL + + +def test_RETRY_INTERVAL_is_int(): + assert isinstance(RETRY_INTERVAL, int)