refactor(func/github/pr): 改用统一的修改拉取请求提交函数

src/function/github/pr.py

@@ -1,4 +1,86 @@
 import subprocess
+from colorama import Fore
+from requests import HTTPError
+from catfood.functions.print import 消息头
+from function.github.token import read_token
+from function.maintain.config import 读取配置
+from function.constant.general import PR_TOOL_NOTE
+from catfood.functions.github.api import 请求GitHubAPI
+from catfood.exceptions.operation import OperationFailed
+
+def submitPR(
+    branch: str,
+    packageIdentifier: str,
+    packageVersion: str,
+    doWhat: str,
+    information: str | None = None,
+    token: str | None = None
+) -> bool:
+    """
+    向上游仓库（microsoft/winget-pkgs）提交修改拉取请求。
+
+    :param branch: 修改位于的分支名
+    :type branch: str
+    :param packageIdentifier: 修改的包的标识符
+    :type packageIdentifier: str
+    :param packageVersion: 修改的包的版本
+    :type packageVersion: str
+    :param doWhat: 做什么修改，该内容作为拉取请求标题开头
+    :type doWhat: str
+    :param information: 要在拉取请求正文中添加的内容
+    :type information: str | None
+    :param token: 创建拉取请求时使用的 GitHub Token
+    :type token: str | None
+    :return: 是否成功创建拉取请求
+    :rtype: bool
+    """
+
+    pkgs仓库 = 读取配置("repos.winget-pkgs")
+    if not isinstance(pkgs仓库, tuple):
+        print(f"{消息头.错误} 未能获取配置文件中的 repos.winget-pkgs")
+        return False
+    owner: str = pkgs仓库[0]
+
+    if not token:
+        token = read_token()
+        if not token:
+            print(f"{消息头.错误} 未能读取到 Token")
+
+    jsonData: dict[str, str | bool] = {
+        "title": f"{doWhat}: {packageIdentifier} version {packageVersion}",
+        "head": f"{owner}:{branch}",
+        "base": "master",
+        "body": PR_TOOL_NOTE
+    }
+
+    if information:
+        jsonData["body"] = f"{jsonData['body']}\n\n{information}"
+
+    # 保证 PR body 最后是分割线
+    if isinstance(jsonData["body"], str) and (not jsonData["body"].rstrip('\n').endswith("---")):
+        jsonData["body"] = f"{jsonData["body"].rstrip('\n')}\n\n---\n\n"
+
+    if 读取配置("github.pr.maintainer_can_modify") == False:
+        jsonData["maintainer_can_modify"] = False
+
+    try:
+        try:
+            response = 请求GitHubAPI(
+                api="https://api.github.com/repos/microsoft/winget-pkgs/pulls",
+                json=jsonData,
+                token=token,
+                method="POST",
+                raiseException=True
+            )
+            if not response:
+                raise ValueError(f"catfood 的 请求GitHubAPI 函数没有返回有效的预期值，实际返回 {repr(response)}")
+            print(f"{消息头.成功} 拉取请求创建成功: {response.get("html_url", f"{Fore.YELLOW}未能获取到拉取请求链接{Fore.RESET}")}")
+            return True
+        except HTTPError as e:
+            raise OperationFailed(f"GitHub API 响应 {e.response.status_code} 错误: {e.response.text}") from e
+    except Exception as e:
+        print(f"{消息头.错误} 创建拉取请求失败: {Fore.RED}{e}{Fore.RESET}")
+        return False
 
 def 检查重复拉取请求(包标识符: str, 包版本: str) -> bool:
     """

src/tools/modify.py

@@ -3,23 +3,22 @@
 import csv
 import time
 import random
-import requests
 import subprocess
 from colorama import Fore
+from typing import Literal
 from datetime import datetime
-from catfood.constant import YES
 from catfood.functions.print import 消息头
 from function.git.format import branchName
+from function.github.pr import submitPR
 from function.maintain.config import 读取配置
 from catfood.functions.files import open_file
 from function.files.manifest import 获取清单目录
 from function.files.manifest import FormatManifest
-from function.constant.general import PR_TOOL_NOTE
 from function.github.token import read_token, 这是谁的Token
 
-def main(args: list[str]):
+def main(args: list[str]) -> Literal[1, 0]:
     global 包标识符, 包版本, 日志文件路径
-    global 解决, 清单目录, 首个_PR, 格式化审查者, 程序所在目录
+    global 解决, 清单目录, 格式化审查者, 程序所在目录
     global owner
 
     # 目录路径
@@ -85,7 +84,6 @@ def main(args: list[str]):
                         return 1
 
                 格式化审查者 = ' , '.join([f"@{审查者}" for 审查者 in 审查者列表])
-                首个_PR = "是"
 
     # ========= 日志 配置 开始 =========
     os.chdir(程序所在目录)
@@ -189,52 +187,8 @@ def 写入日志(消息: str, 等级: str="INFO"):
         for 行 in 消息.split("\n"):
             日志文件.write(f"{写入时间} {等级} {行}\n")
 
-# 创建拉取请求
-def 创建拉取请求(分支名: str, 版本文件夹: str, 审查: str="") -> str | int:
-    # 审查:
-    # "" -> 不请求审查
-    # 带 @ 的字符串 -> 在 PR body 中 @ 审查者
-    # 不带 @ 的字符串 -> 在 PR body 中引用首个拉取请求
-    global 解决
-    while True: # 不 break 直接 return
-        github_token = read_token()
-        if not github_token:
-            print(f"{消息头.错误} 拉取请求创建失败: Token 读取失败")
-            return 1
-        api = "https://api.github.com/repos/microsoft/winget-pkgs/pulls"
-        请求头 = {
-            "Authorization": f"token {github_token}",
-            "Accept": "application/vnd.github.v3+json"
-        }
-        数据: dict[str, str | bool]
-        数据 = {
-            "title": f"Modify: {包标识符} version {版本文件夹} (Auto)",
-            "head": f"{owner}:{分支名}",
-            "base": "master",
-            "body": f"{PR_TOOL_NOTE}\n\n{审查}\n{解决}\n\n---\n"
-        }
-        if 读取配置("github.pr.maintainer_can_modify") == False:
-            数据["maintainer_can_modify"] = False
-
-        response = requests.post(api, headers=请求头, json=数据)
-        if response.status_code == 201:
-            print(f"    {Fore.GREEN}拉取请求创建成功: {response.json()["html_url"]}")
-            写入日志(f"    Pull request created successfully: {response.json()["html_url"]}")
-            return response.json()["html_url"]
-        else:
-            print(f"    {Fore.RED}拉取请求创建失败: {response.status_code} - {response.text}")
-            写入日志(f"    Failed to create pull request: {response.status_code} - {response.text}", "ERROR")
-            try:
-                if input(f"{消息头.问题} 我应该重试吗[Y/N]: ").lower() not in (*YES, "应该", "重试", "retry"):
-                    return 1
-                print("正在重试...")
-                写入日志("    Retrying to create a pull request...")
-            except KeyboardInterrupt:
-                return 1
-
 # Git 操作部分
-def 修改版本(版本文件夹: str):
-    global 首个_PR
+def 修改版本(版本文件夹: str) -> Literal[1, 0]:
     print(f"\n正在处理版本文件夹: {版本文件夹}")
     写入日志(f"Processing version folder: {版本文件夹}")
     版本文件夹路径 = os.path.join(清单目录, 版本文件夹)
@@ -326,16 +280,13 @@ def 修改版本(版本文件夹: str):
     写入日志(f"    Successfully pushed to remote (origin): {新分支}")
 
     # 创建拉取请求
-    if 格式化审查者:
-        if 首个_PR == "是":
-            首个_PR = 创建拉取请求(新分支, 版本文件夹, f"{格式化审查者} PTAL")
-            if 首个_PR == 1:
-                return 1 # 创建拉取请求时出错
-        else:
-            if 创建拉取请求(新分支, 版本文件夹, f"Review has been requested in {首个_PR}") == 1:
-                return 1 # 创建拉取请求时出错
+    if submitPR(
+        branch=新分支,
+        packageIdentifier=包标识符,
+        packageVersion=包版本,
+        doWhat="Modify",
+        information=(f"\n\n{格式化审查者} PTAL" if 格式化审查者 else "") + (f"\n\n{解决}" if 解决 else "")
+    ):
+        return 0
     else:
-        if 创建拉取请求(新分支, 版本文件夹) == 1:
-            return 1 # 创建拉取请求时出错
-
-    return 0 # 成功处理版本文件夹
+        return 1

src/tools/remove.py

@@ -4,7 +4,6 @@
 import time
 import shutil
 import tempfile
-import requests
 import subprocess
 import webbrowser
 import tools.cat as cat
@@ -13,50 +12,13 @@
 from catfood.constant import YES, NO
 from catfood.functions.print import 消息头
 from function.git.format import branchName
+from function.github.pr import submitPR
 from function.maintain.config import 读取配置
 from translate import Translator # type: ignore
-from function.constant.general import PR_TOOL_NOTE
 from catfood.exceptions.operation import OperationFailed
 from function.github.token import read_token, 这是谁的Token
 from function.files.manifest import 获取清单目录, 获取现有包版本
 
-# 创建拉取请求
-def 创建拉取请求(包标识符: str, 分支名: str, 版本文件夹: str, 理由: str):
-    global owner, 手动验证结果
-    while True: # 不 break 直接 return
-        github_token = read_token()
-        if not github_token:
-            print(f"{消息头.错误} 拉取请求创建失败: Token 读取失败")
-            return 1
-
-        api = "https://api.github.com/repos/microsoft/winget-pkgs/pulls"
-        请求头 = {
-            "Authorization": f"token {github_token}",
-            "Accept": "application/vnd.github.v3+json"
-        }
-        数据: dict[str, str | bool] = {
-            "title": f"Remove version: {包标识符} version {版本文件夹} (Auto)",
-            "head": f"{owner}:{分支名}",
-            "base": "master",
-            "body": f"{PR_TOOL_NOTE}\n\n{理由}{f'\n{手动验证结果}' if 手动验证结果 else ''}\n\n---\n"
-        }
-
-        if 读取配置("github.pr.maintainer_can_modify") == False:
-            数据["maintainer_can_modify"] = False
-
-        response = requests.post(api, headers=请求头, json=数据)
-        if response.status_code == 201:
-            print(f"    {Fore.GREEN}拉取请求创建成功: {response.json()["html_url"]}")
-            return response.json()["html_url"]
-        else:
-            print(f"    {Fore.RED}拉取请求创建失败: {response.status_code} - {response.text}")
-            try:
-                if input(f"{消息头.问题} 我应该重试吗[Y/N]: ").lower() not in (*YES, "应该", "重试", "retry"):
-                    return 1
-                print("正在重试...")
-            except KeyboardInterrupt:
-                return 1
-
 def main(args: list[str]) -> int:
     global 手动验证结果, owner
 
@@ -210,7 +172,13 @@ def main(args: list[str]) -> int:
     while (not 理由):
         理由 = input("移除此包版本的理由: ")
 
-    if 创建拉取请求(包标识符, 新分支名, 包版本, 理由) == 1:
+    if not submitPR(
+        branch=新分支名,
+        packageIdentifier=包标识符,
+        packageVersion=包版本,
+        doWhat="Remove version",
+        information=f"{理由}{f'\n{手动验证结果}' if 手动验证结果 else ''}"
+    ):
         return 1 # 拉取请求创建失败
 
     print(f"{Fore.GREEN} 成功移除 {包标识符} 版本 {包版本}")