diff --git a/src/function/github/pr.py b/src/function/github/pr.py index 33d0c22..8e7ba65 100644 --- a/src/function/github/pr.py +++ b/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: """ diff --git a/src/tools/modify.py b/src/tools/modify.py index db2e057..6779157 100644 --- a/src/tools/modify.py +++ b/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 diff --git a/src/tools/remove.py b/src/tools/remove.py index 9b7e54e..129a64c 100644 --- a/src/tools/remove.py +++ b/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} 成功移除 {包标识符} 版本 {包版本}")