feat: simulate git clone
This commit is contained in:
parent
6ef9911461
commit
8ca7cd3504
@ -22,3 +22,5 @@ tools_cache_path = base_dirs.site_cache_path.joinpath("tools")
|
|||||||
tools_cache_path.mkdir(parents=True, exist_ok=True)
|
tools_cache_path.mkdir(parents=True, exist_ok=True)
|
||||||
launcher_cache_path = base_dirs.site_cache_path.joinpath("launcher")
|
launcher_cache_path = base_dirs.site_cache_path.joinpath("launcher")
|
||||||
launcher_cache_path.mkdir(parents=True, exist_ok=True)
|
launcher_cache_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
utils_cache_path = base_dirs.site_cache_path.joinpath("utils")
|
||||||
|
utils_cache_path.mkdir(parents=True, exist_ok=True)
|
@ -1,35 +1,99 @@
|
|||||||
import subprocess
|
import subprocess
|
||||||
|
import requests
|
||||||
|
import json
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from shutil import which
|
from zipfile import ZipFile
|
||||||
from vollerei.utils.git.exceptions import GitCloneError, GitNotInstalled
|
from io import BytesIO
|
||||||
|
from shutil import which, rmtree
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
from vollerei.constants import utils_cache_path
|
||||||
|
from vollerei.utils.git.exceptions import GitCloneError
|
||||||
|
|
||||||
|
|
||||||
class Git:
|
class Git:
|
||||||
"""
|
"""
|
||||||
Quick wrapper around git binary
|
Quick wrapper around git binary (or simulate git if git is not installed)
|
||||||
|
|
||||||
|
Simulate git because Windows users may not have git installed.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
pass
|
self._cache = utils_cache_path.joinpath("git")
|
||||||
|
self._cache.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def check_git():
|
def is_installed():
|
||||||
"""
|
"""
|
||||||
Check for git installation, if not found raise GitNotInstalled
|
Check for git installation, if not found raise GitNotInstalled
|
||||||
"""
|
"""
|
||||||
if not which("git"):
|
if not which("git"):
|
||||||
raise GitNotInstalled("git is not installed")
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _is_gitea(self, netloc: str) -> bool:
|
||||||
|
"""
|
||||||
|
Check if the url is a Gitea server
|
||||||
|
"""
|
||||||
|
rsp = requests.get(f"https://{netloc}/api/v1/meta")
|
||||||
|
try:
|
||||||
|
data: dict = rsp.json()
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
return False
|
||||||
|
if not data.get("version"):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
def _clone(self, url: str, path: str = None) -> None:
|
||||||
|
"""
|
||||||
|
"Clone" a git repository without git
|
||||||
|
"""
|
||||||
|
if Path(url).suffix == ".git":
|
||||||
|
url = url[:-4]
|
||||||
|
url_info = urlparse(url)
|
||||||
|
netloc = url_info.netloc
|
||||||
|
if path is None:
|
||||||
|
path = Path.cwd().joinpath(Path(url).stem)
|
||||||
|
if self._is_gitea(netloc):
|
||||||
|
# Hardcoding the branch to master, because well :D
|
||||||
|
file = BytesIO()
|
||||||
|
rsp = requests.get(
|
||||||
|
f"https://{netloc}/api/v1/repos/{url_info.path}/archive/master.zip",
|
||||||
|
stream=True,
|
||||||
|
)
|
||||||
|
rsp.raise_for_status()
|
||||||
|
with open(file, "wb") as f:
|
||||||
|
for chunk in rsp.iter_content(chunk_size=32768):
|
||||||
|
f.write(chunk)
|
||||||
|
zip_file = ZipFile(file)
|
||||||
|
zip_file.extractall(path)
|
||||||
|
with Path(path).joinpath(".git/PLEASE_INSTALL_GIT").open("w") as f:
|
||||||
|
f.write(json.dumps({"type": "gitea", "netloc": netloc, "path": url_info.path}))
|
||||||
|
else:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
def pull_or_clone(self, url: str, path: str = None) -> None:
|
def pull_or_clone(self, url: str, path: str = None) -> None:
|
||||||
self.check_git()
|
|
||||||
"""
|
"""
|
||||||
Pulls or clones a git repository
|
Pulls or clones a git repository
|
||||||
|
|
||||||
If the repository already exists and the url matches, it'll be pulled.
|
If the repository already exists and the url matches, it'll be pulled.
|
||||||
"""
|
"""
|
||||||
|
if not self.is_installed():
|
||||||
|
# Git is not installed, we need to simulate it
|
||||||
|
self._clone(url, path)
|
||||||
|
return
|
||||||
if path is None:
|
if path is None:
|
||||||
|
if Path(url).suffix == ".git":
|
||||||
path = Path.cwd().joinpath(Path(url).stem)
|
path = Path.cwd().joinpath(Path(url).stem)
|
||||||
|
else:
|
||||||
|
path = Path.cwd().joinpath(Path(url).name)
|
||||||
|
path_as_path = Path(path)
|
||||||
|
if path_as_path.joinpath(".git/PLEASE_INSTALL_GIT").exists():
|
||||||
|
# This is a fake .git directory we created.
|
||||||
|
# We need to clone the repository
|
||||||
|
rmtree(path)
|
||||||
try:
|
try:
|
||||||
|
if not path_as_path.exists():
|
||||||
|
raise subprocess.CalledProcessError
|
||||||
origin_url = subprocess.check_output(
|
origin_url = subprocess.check_output(
|
||||||
["git", "config", "--get", "remote.origin.url"], cwd=path
|
["git", "config", "--get", "remote.origin.url"], cwd=path
|
||||||
).decode()
|
).decode()
|
||||||
@ -37,6 +101,8 @@ class Git:
|
|||||||
raise subprocess.CalledProcessError
|
raise subprocess.CalledProcessError
|
||||||
subprocess.check_call(["git", "pull"], cwd=path)
|
subprocess.check_call(["git", "pull"], cwd=path)
|
||||||
except subprocess.CalledProcessError:
|
except subprocess.CalledProcessError:
|
||||||
|
if path_as_path.exists():
|
||||||
|
rmtree(path)
|
||||||
try:
|
try:
|
||||||
subprocess.check_call(["git", "clone", url, path])
|
subprocess.check_call(["git", "clone", url, path])
|
||||||
except subprocess.CalledProcessError as e:
|
except subprocess.CalledProcessError as e:
|
||||||
|
Loading…
Reference in New Issue
Block a user