autopatch refactor 1

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,34 +181,34 @@ 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)
             create_pull_request(app["repo"], patch, current_branch, s)
             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()