From d600250106438ffe08e75376544642b15599fd13 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?F=C3=A9lix=20Pi=C3=A9dallu?= Date: Thu, 12 Sep 2024 15:50:58 +0200 Subject: [PATCH] autopatch refactor 1 --- tools/autopatches/autopatch.py | 131 +++++++++++++++++++++------------ 1 file changed, 84 insertions(+), 47 deletions(-) diff --git a/tools/autopatches/autopatch.py b/tools/autopatches/autopatch.py index cfc3742b56..732a59a221 100755 --- a/tools/autopatches/autopatch.py +++ b/tools/autopatches/autopatch.py @@ -6,28 +6,34 @@ import subprocess import sys import time +from typing import Optional from pathlib import Path import requests import toml +from git import Repo + # add apps/tools to sys.path sys.path.insert(0, str(Path(__file__).parent.parent)) +from app_caches import AppDir from appslib.utils import ( # noqa: E402 pylint: disable=import-error,wrong-import-position get_catalog, ) +import appslib.get_apps_repo as get_apps_repo TOOLS_DIR = Path(__file__).resolve().parent.parent my_env = os.environ.copy() my_env["GIT_TERMINAL_PROMPT"] = "0" -os.makedirs(".apps_cache", exist_ok=True) login = (TOOLS_DIR / ".github_login").open("r", encoding="utf-8").read().strip() token = (TOOLS_DIR / ".github_token").open("r", encoding="utf-8").read().strip() github_api = "https://api.github.com" +PATCHES_PATH = Path(__file__).resolve().parent / "patches" + def apps(min_level=4): for app, infos in get_catalog().items(): @@ -36,10 +42,6 @@ def apps(min_level=4): yield infos -def app_cache_folder(app): - return os.path.join(".apps_cache", app) - - def git(cmd, in_folder=None): if not isinstance(cmd, list): cmd = cmd.split() @@ -70,56 +72,85 @@ def show(j, name=""): file.flush() -def build_cache(): +class AppToPatch: + def __init__(self, id: str, path: Path, info: dict) -> None: + self.id = id + self.path = path + self.info = info + self.repo: Optional[Repo] = None + + def init(self) -> None: + appdir = AppDir(self.id, self.path) + appdir.ensure(self.info["url"], False, self.info.get("branch", "master")) + + self.repo = Repo(self.path) + if "fork" not in self.repo.remotes: + reponame = self.info["url"].rsplit("/", 1)[-1] + self.repo.create_remote("fork", url=f"https://{LOGIN}:{token}@github.com/{LOGIN}/{reponame}") + + def apply(self, patch: Path) -> None: + assert self.repo is not None + current_branch = self.repo.active_branch + self.repo.head.reset(f"origin/{current_branch}", index=True, working_tree=True) + subprocess.call([PATCHES_PATH / patch / "patch"], cwd=self.path) + + def diff(self) -> None: + assert self.repo is not None + diff = self.repo.head.commit.diff() + if not diff: + return + print("\n\n\n") + print("=================================") + print(f"Changes in : {self.id}") + print("=================================") + print("\n") + print(diff) + + + def push(self) -> None: + pass + + +def build_cache(cache_path: Path) -> None: for app in progressbar(apps(), "Git cloning: ", 40): - folder = os.path.join(".apps_cache", app["id"]) + app_cache = cache_path / app["id"] reponame = app["url"].rsplit("/", 1)[-1] - git(f"clone --quiet --depth 1 --single-branch {app['url']} {folder}") + git(f"clone --quiet --depth 1 --single-branch {app['url']} {app_cache}") git( - f"remote add fork https://{login}:{token}@github.com/{login}/{reponame}", - in_folder=folder, + f"remote add fork https://{LOGIN}:{token}@github.com/{LOGIN}/{reponame}", + in_folder=app_cache, ) -def apply(patch): - patch_path = os.path.abspath(os.path.join("patches", patch, "patch.sh")) - +def apply(cache_path: Path, patch: str) -> None: for app in progressbar(apps(), "Apply to: ", 40): - folder = os.path.join(".apps_cache", app["id"]) - current_branch = git(f"symbolic-ref --short HEAD", in_folder=folder) + folder = cache_path / app["id"] + current_branch = git("symbolic-ref --short HEAD", in_folder=folder) git(f"reset --hard origin/{current_branch}", in_folder=folder) - os.system(f"cd {folder} && bash {patch_path}") + subprocess.call([PATCHES_PATH / patch / "patch"], cwd=folder) -def diff(): +def diff(cache_path: Path) -> None: for app in apps(): - folder = os.path.join(".apps_cache", app["id"]) - if bool( - subprocess.check_output(f"cd {folder} && git diff", shell=True) - .strip() - .decode("utf-8") - ): + folder = cache_path / app["id"] + diff = subprocess.check_output(["git", "--no-pager", "diff"], cwd=folder).decode("utf-8") + if diff.strip(): print("\n\n\n") print("=================================") print("Changes in : " + app["id"]) print("=================================") print("\n") - os.system(f"cd {folder} && git --no-pager diff") + print(diff) -def push(patch): - title = ( - "[autopatch] " - + open(os.path.join("patches", patch, "pr_title.md")).read().strip() - ) +def push(cache_path: Path, patch: str) -> None: + pr_title = (PATCHES_PATH / patch / "pr_title.md").open().read().strip() + title = f"[autopatch] {pr_title}" def diff_not_empty(app): - folder = os.path.join(".apps_cache", app["id"]) - return bool( - subprocess.check_output(f"cd {folder} && git diff", shell=True) - .strip() - .decode("utf-8") - ) + folder = cache_path / app["id"] + diff = subprocess.check_output(["git", "--no-pager", "diff"], cwd=folder).decode("utf-8") + return bool(diff.strip()) def app_is_on_github(app): return "github.com" in app["url"] @@ -138,7 +169,7 @@ def app_is_on_github(app): for app in progressbar(apps_to_push, "Pushing: ", 40): app["repo"] = app["url"][len("https://github.com/") :].strip("/") app_repo_name = app["url"].rsplit("/", 1)[-1] - folder = os.path.join(".apps_cache", app["id"]) + folder = cache_path / app["id"] current_branch = git(f"symbolic-ref --short HEAD", in_folder=folder) git(f"reset origin/{current_branch}", in_folder=folder) git( @@ -150,7 +181,7 @@ def app_is_on_github(app): except Exception: pass git( - f"remote add fork https://{login}:{token}@github.com/{login}/{app_repo_name}", + f"remote add fork https://{LOGIN}:{token}@github.com/{LOGIN}/{app_repo_name}", in_folder=folder, ) git(f"push fork {current_branch}:{patch} --quiet --force", in_folder=folder) @@ -158,26 +189,26 @@ def app_is_on_github(app): time.sleep(4) # to avoid rate limiting lol -def fork_if_needed(repo, s): +def fork_if_needed(repo: str, session: requests.Session) -> None: repo_name = repo.split("/")[-1] - r = s.get(github_api + f"/repos/{login}/{repo_name}") + r = session.get(github_api + f"/repos/{LOGIN}/{repo_name}") if r.status_code == 200: return - r = s.post(github_api + f"/repos/{repo}/forks") + r = session.post(github_api + f"/repos/{repo}/forks") if r.status_code != 200: print(r.text) def create_pull_request(repo, patch, base_branch, s): + pr_title = (PATCHES_PATH / patch / "pr_title.md").open().read().strip() + pr_body = (PATCHES_PATH / patch / "pr_body.md").open().read().strip() PR = { - "title": "[autopatch] " - + open(os.path.join("patches", patch, "pr_title.md")).read().strip(), - "body": "This is an automatic PR\n\n" - + open(os.path.join("patches", patch, "pr_body.md")).read().strip(), - "head": login + ":" + patch, + "title": f"[autopatch] {pr_title}", + "body": f"This is an automatic PR\n\n{pr_body}", + "head": f"{LOGIN}:{patch}", "base": base_branch, "maintainer_can_modify": True, } @@ -192,6 +223,7 @@ def create_pull_request(repo, patch, base_branch, s): def main(): parser = argparse.ArgumentParser() + get_apps_repo.add_args(parser) parser.add_argument( "the_patch", type=str, nargs="?", help="The name of the patch to apply" ) @@ -212,11 +244,15 @@ def main(): ) args = parser.parse_args() + get_apps_repo.from_args(args) + cache_path = get_apps_repo.cache_path(args) + cache_path.mkdir(exist_ok=True, parents=True) + if not (args.cache or args.apply or args.diff or args.push): parser.error("We required --cache, --apply, --diff or --push.") if args.cache: - build_cache() + build_cache(cache_path) if args.apply: if not args.the_patch: @@ -232,4 +268,5 @@ def main(): push(args.the_patch) -main() +if __name__ == "__main__": + main()