Use gitpython

Author: SanmerDev

requirements.txt

@@ -1,4 +1,5 @@
 pygithub>=1.59.0
 python-dateutil>=2.8.2
 requests>=2.31.0
-tabulate>=0.9.0
+tabulate>=0.9.0
+gitpython>=3.1.37

sync/cli/Parameters.py

@@ -281,8 +281,7 @@ def configure_parser_github(cls, sub_parsers):
             metavar="GITHUB_TOKEN",
             type=str,
             default=cls._github_token,
-            help="GitHub REST API Token for PyGitHub. "
-                 "This can be defined in env as 'GITHUB_TOKEN', default: {0}.".format("%(default)s")
+            help="GitHub REST API Token for PyGitHub, same as 'export GITHUB_TOKEN=...'."
         )
 
     @classmethod
@@ -412,7 +411,7 @@ def add_parser_env(cls, p, add_quiet=False):
                 "-q",
                 "--quiet",
                 action="store_true",
-                help="Disable all logging piped through stdout."
+                help="Show only error logs (piped through stderr)."
             )
 
         env.add_argument(
@@ -436,7 +435,7 @@ def add_parser_git(cls, p, add_set_size=True):
             dest="git_branch",
             metavar="GIT_BRANCH",
             type=str,
-            default=get_current_branch(cls._root_folder),
+            default=GitUtils.current_branch(cls._root_folder),
             help="Define branch to push, current: {0}.".format("%(default)s")
         )
 
@@ -451,26 +450,3 @@ def add_parser_git(cls, p, add_set_size=True):
             )
 
         return git
-
-    @classmethod
-    def add_parser_quiet(cls, p):
-        p.add_argument(
-            "-q",
-            "--quiet",
-            action="store_true",
-            help="Show only error logs (piped through stderr)."
-        )
-
-    @classmethod
-    def add_parser_help(cls, p):
-        p.add_argument(
-            "-h",
-            "--help",
-            action=_HelpAction,
-            help="Show this help message and exit.",
-        )
-
-
-def get_current_branch(root_folder: Path):
-    GitUtils.set_cwd_folder(root_folder)
-    return GitUtils.current_branch()

sync/core/Sync.py

@@ -1,8 +1,8 @@
 import concurrent.futures
-import subprocess
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 
+from git import Repo
 from tabulate import tabulate
 
 from .Config import Config
@@ -13,7 +13,7 @@
     TrackJson
 )
 from ..track import BaseTracks, LocalTracks, GithubTracks
-from ..utils import Log, GitUtils
+from ..utils import Log
 
 
 class Sync:
@@ -156,16 +156,13 @@ def update(self, module_ids=None, force=False, single=False, **kwargs):
 
     def push_by_git(self, branch):
         json_file = self._json_folder.joinpath(ModulesJson.filename())
-        if not GitUtils.is_enable():
-            self._log.e("push_by_git: git command not found")
-            return
-
         timestamp = ModulesJson.load(json_file).get_timestamp()
         msg = f"Update by CLI ({datetime.fromtimestamp(timestamp)})"
 
-        subprocess.run(["git", "add", "."], cwd=self._root_folder.as_posix())
-        subprocess.run(["git", "commit", "-m", msg], cwd=self._root_folder.as_posix())
-        subprocess.run(["git", "push", "-u", "origin", branch], cwd=self._root_folder.as_posix())
+        repo = Repo(self._root_folder)
+        repo.git.add(all=True)
+        repo.index.commit(msg)
+        repo.remote().push(branch)
 
     def get_versions_diff(self):
         headers = ["id", "name", "version"]

sync/utils/GitUtils.py

@@ -1,98 +1,35 @@
 import functools
 import os
 import shutil
-import subprocess
-from datetime import datetime
 from pathlib import Path
-from subprocess import CalledProcessError
 from typing import Optional
 
-from dateutil.parser import parse
-from requests import HTTPError
+from git import Repo, InvalidGitRepositoryError, GitCommandError
 
 
 class GitUtils:
-    _cwd_folder = None
-
-    @classmethod
-    def set_cwd_folder(cls, cwd: Optional[Path] = None):
-        cls._cwd_folder = cwd
-
     @classmethod
     @functools.lru_cache()
-    def exec(cls, cmd: str) -> str:
-        return subprocess.run(
-            args=cmd.split(" "),
-            stdout=subprocess.PIPE,
-            stderr=subprocess.DEVNULL,
-            cwd=cls._cwd_folder
-        ).stdout.decode("utf-8").strip()
-
-    @classmethod
-    def version(cls) -> str:
-        return cls.exec("git --version")
-
-    @classmethod
-    def is_enable(cls) -> bool:
-        return cls.version() is not None
-
-    @classmethod
-    def branch_all(cls) -> str:
-        return cls.exec("git branch --all")
-
-    @classmethod
-    def current_branch(cls) -> str:
-        for out in cls.branch_all().splitlines():
-            if not out.startswith("*"):
-                continue
-
-            out = out.strip().split(maxsplit=1)
-            return out[-1]
-
-    @classmethod
-    def commit_id(cls) -> str:
-        return cls.exec("git rev-parse --short HEAD")
-
-    @classmethod
-    def commit_count(cls) -> int:
-        return int(cls.exec("git rev-list --count HEAD"))
-
-    @classmethod
-    def has_tag(cls, name: str) -> bool:
-        return cls.exec(f"git tag -l {name}") != ""
+    def current_branch(cls, repo_dir: Path) -> Optional[str]:
+        try:
+            return Repo(repo_dir).active_branch.name
+        except InvalidGitRepositoryError:
+            return None
 
     @classmethod
     def clone_and_zip(cls, url: str, out: Path) -> float:
-        out.parent.mkdir(parents=True, exist_ok=True)
         repo_dir = out.with_suffix("")
-
-        if not cls.is_enable():
-            raise RuntimeError("git command not found")
+        if repo_dir.exists():
+            shutil.rmtree(repo_dir)
 
         try:
-            subprocess.run(
-                args=["git", "clone", url, repo_dir.as_posix(), "--depth=1"],
-                cwd=out.parent.as_posix(),
-                stderr=subprocess.DEVNULL,
-                check=True
-            )
-        except CalledProcessError:
+            repo = Repo.clone_from(url, repo_dir)
+            last_committed = float(repo.commit().committed_date)
+        except GitCommandError:
             shutil.rmtree(repo_dir, ignore_errors=True)
-            raise HTTPError(f"clone failed: {url}")
-
-        try:
-            result = subprocess.run(
-                ["git", "log", "--format=%cd"],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.DEVNULL,
-                cwd=repo_dir.as_posix()
-            ).stdout.decode("utf-8")
-
-            last_committed = parse(result).timestamp()
-        except CalledProcessError:
-            last_committed = datetime.now().timestamp()
+            raise GitCommandError(f"clone failed: {url}")
 
-        for path in repo_dir.rglob("*"):
+        for path in repo_dir.iterdir():
             if path.name.startswith(".git"):
                 if path.is_dir():
                     shutil.rmtree(path, ignore_errors=True)
@@ -103,7 +40,10 @@ def clone_and_zip(cls, url: str, out: Path) -> float:
 
             os.utime(path, (last_committed, last_committed))
 
-        shutil.make_archive(repo_dir.as_posix(), format="zip", root_dir=repo_dir)
-        shutil.rmtree(repo_dir)
+        try:
+            shutil.make_archive(repo_dir.as_posix(), format="zip", root_dir=repo_dir)
+            shutil.rmtree(repo_dir)
+        except FileNotFoundError:
+            raise FileNotFoundError(f"archive failed: {repo_dir.as_posix()}")
 
         return last_committed