GiteaPC/giteapc/repos.py

from giteapc.config import REPO_FOLDER
from giteapc.util import *
import subprocess
from subprocess import PIPE
import os.path
import shutil
import glob
import re

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"], stdout=PIPE)
            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 describe(self, tags=True, all=False, always=True):
        args = ["--tags"]*tags + ["--all"]*all + ["--always"]*always
        p = self._git(["describe"] + args, stdout=PIPE)
        return p.stdout.decode("utf-8").strip()

    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)

    # Metadata

    def metadata(self):
        RE_METADATA = re.compile(r'^\s*#\s*giteapc\s*:\s*(.*)$')
        metadata = dict()

        with open(self.folder + "/giteapc.make", "r") as fp:
            makefile = fp.read()

        for line in makefile.split("\n"):
            m = re.match(RE_METADATA, line)
            if m is None:
                break

            for assignment in m[1].split():
                name, value = assignment.split("=", 1)
                if name not in metadata:
                    metadata[name] = value
                else:
                    metadata[name] += "," + value

        return metadata

    def dependencies(self):
        metadata = self.metadata()
        if "depends" in metadata:
            return metadata["depends"].split(",")
        else:
            return []