forked from Lephenixnoir/GiteaPC
137 lines
3.7 KiB
Python
137 lines
3.7 KiB
Python
from giteapc.config import REPO_FOLDER
|
|
from giteapc.util import *
|
|
import subprocess
|
|
from subprocess import PIPE
|
|
import os.path
|
|
import shutil
|
|
import glob
|
|
|
|
class RemoteRepo:
|
|
# Create a remote repo from the JSON object returned by Gitea
|
|
def __init__(self, j):
|
|
self.j = j
|
|
self.remote = True
|
|
|
|
@property
|
|
def name(self):
|
|
return self.j["name"]
|
|
|
|
@property
|
|
def owner(self):
|
|
return self.j["owner"]["username"]
|
|
|
|
@property
|
|
def fullname(self):
|
|
return self.j["full_name"]
|
|
|
|
@property
|
|
def description(self):
|
|
return self.j["description"]
|
|
|
|
@property
|
|
def parent(self):
|
|
p = self.j["parent"]
|
|
return RemoteRepo(p) if p is not None else None
|
|
|
|
@property
|
|
def url(self):
|
|
return f"/repos/{self.owner}/{self.name}"
|
|
|
|
def clone_url(self, protocol):
|
|
if protocol == "ssh":
|
|
return self.j["ssh_url"]
|
|
else:
|
|
return self.j["clone_url"]
|
|
|
|
|
|
class LocalRepo:
|
|
# Create a remote repo from the full name or path
|
|
def __init__(self, fullname):
|
|
if fullname.startswith(REPO_FOLDER + "/"):
|
|
fullname = fullname[len(REPO_FOLDER)+1:]
|
|
|
|
assert fullname.count("/") == 1
|
|
|
|
self.fullname = fullname
|
|
self.owner, self.name = fullname.split("/")
|
|
self.folder = REPO_FOLDER + "/" + fullname
|
|
self.remote = False
|
|
self.makefile = self.folder + "/giteapc.make"
|
|
|
|
@staticmethod
|
|
def path(fullname):
|
|
return REPO_FOLDER + "/" + fullname
|
|
|
|
@staticmethod
|
|
def exists(fullname):
|
|
return os.path.exists(LocalRepo.path(fullname))
|
|
|
|
@staticmethod
|
|
def all():
|
|
return [ LocalRepo(path) for path in glob.glob(REPO_FOLDER + f"/*/*") ]
|
|
|
|
@staticmethod
|
|
def clone(r, method="https"):
|
|
src = r.clone_url(method)
|
|
dst = LocalRepo.path(r.fullname)
|
|
mkdir_p(dst)
|
|
cmd = ["git", "clone", src, dst]
|
|
|
|
try:
|
|
run(cmd, stdout=PIPE, stderr=PIPE)
|
|
return LocalRepo(r.fullname)
|
|
except ProcessError as e:
|
|
# On error, delete the failed clone
|
|
shutil.rmtree(dst)
|
|
raise e
|
|
|
|
# Git commands
|
|
|
|
def _git(self, command, *args, **kwargs):
|
|
return run(["git", "-C", self.folder] + command,
|
|
*args, **kwargs)
|
|
|
|
def is_on_branch(self):
|
|
try:
|
|
self._git(["symbolic-ref", "-q", "HEAD"])
|
|
return True
|
|
except ProcessError as e:
|
|
if e.returncode == 1:
|
|
return False
|
|
raise e
|
|
|
|
def fetch(self):
|
|
self._git(["fetch"])
|
|
|
|
def pull(self):
|
|
if self.is_on_branch():
|
|
self._git(["pull"])
|
|
|
|
def checkout(self, version):
|
|
self._git(["checkout", version], stdout=PIPE, stderr=PIPE)
|
|
|
|
def branches(self):
|
|
proc = self._git(["branch"], stdout=subprocess.PIPE)
|
|
local = proc.stdout.decode("utf-8").split("\n")
|
|
local = [ b[2:] for b in local if b ]
|
|
|
|
proc = self._git(["branch", "-r"], stdout=subprocess.PIPE)
|
|
remote = proc.stdout.decode("utf-8").split("\n")
|
|
remote = [ b[9:] for b in remote
|
|
if b and b.startswith(" origin/") and "/HEAD " not in b ]
|
|
remote = [ b for b in remote if b not in local ]
|
|
|
|
return [ {"name": b, "local": True } for b in local ] + \
|
|
[ {"name": b, "local": False} for b in remote ]
|
|
|
|
def tags(self):
|
|
proc = self._git(["tag", "--list"], stdout=subprocess.PIPE)
|
|
tags = proc.stdout.decode("utf-8").split("\n")
|
|
return [ {"name": t} for t in tags if t ]
|
|
|
|
# Make commands
|
|
|
|
def make(self, target, env=None):
|
|
with ChangeDirectory(self.folder):
|
|
return run(["make", "-f", "giteapc.make", target], env=env)
|