git-flow/lib/git.py

import os
import subprocess
import sys

from lib.io import *

FLOWCONFIG_FILE = ".flowconfig"
CHANGELOG_FILE = "CHANGELOG.md"
UPDATED_CHANGELOG_MSG = "Updated " + CHANGELOG_FILE


def branch_exists(branch):
    return os.path.isfile(os.path.join(".git", "refs", "heads", branch))


def get_current_branch():
    output = os.popen("git rev-parse --abbrev-ref HEAD").readline()
    output = output.strip()

    return None if output == "HEAD" else output


def is_repository():
    return os.path.isdir(".git")


def flow_config(name: str, value: str | None = None) -> str | None:
    if value is None:
        value = os.popen(f"git config --file {FLOWCONFIG_FILE} '{name}'").readline()
        return value.strip()
    else:
        os.popen(f"git config --file {FLOWCONFIG_FILE} '{name}' '{value}'").close()


def config(name: str, value: str | None = None, **kwargs) -> str | None:
    options = []

    for k in kwargs:
        value = kwargs[k]
        k = k.replace("_", "-")

        if isinstance(value, bool):
            options.append("--" + k if value else "--no-" + k)
        else:
            options.append(f"--{k}={value}")

    options = " ".join(options)

    if value is None:
        return os.popen(f"git config {options} '{name}'").readline().strip()
    else:
        os.popen(f"git config {options} '{name}' '{value}'").close()


def create_branch(name, base: str | None = None, switch: bool = False):
    command = "switch -c" if switch else "branch"
    os.popen(f"git {command} '{name}'" + (f" '{base}'" if base is not None else ""))


def get_remotes():
    return os.popen("git remote").readlines()


class Status:
    worktree: str
    index: str
    file: str

    def __init__(self, line: str) -> None:
        self.worktree = line[0]
        self.index = line[1]
        self.file = line[3:].strip("\n")


def status():
    status: list[Status] = []
    lines = os.popen("git status --porcelain").readlines()

    for line in lines:
        status.append(Status(line))

    return status


def switch(branch: str = "-"):
    os.popen(f"git switch '{branch}'").close()


def rename_branch(oldbranch: str, newbranch):
    os.popen(f"git branch -m '{oldbranch}' '{newbranch}'").close()


def _command_list_as_str(cmd: list[str]) -> str:
    s = []

    for arg in cmd:
        arg = ('"' + arg + '"') if " " in arg else arg
        s.append(arg)

    return " ".join(s)


def _run_command(git_subcommand: str, *args, **kwargs) -> subprocess.CompletedProcess:
    cmd = ["git", git_subcommand]

    for k in kwargs:
        value = kwargs[k]
        k = k.replace("_", "-")

        if isinstance(value, bool):
            cmd.append("--" + k if value else "--no-" + k)
        else:
            if len(k) == 1:
                cmd.append("-" + k)
            else:
                cmd.append("--" + k)

            cmd.append(value)

    cmd += args

    print(f"> Running: " + _command_list_as_str(cmd))

    return subprocess.run(cmd, capture_output=True, text=True)


def output(git_subcommand: str, *args, **kwargs) -> list[str] | None:
    completed_process = _run_command(git_subcommand, *args, **kwargs)

    if completed_process.returncode == 0:
        return list(filter(lambda l: l, str(completed_process.stdout).splitlines()))
    else:
        for line in str(completed_process.stderr).splitlines():
            print("! " + line, file=sys.stderr)


def exec(git_subcommand: str, *args, **kwargs) -> bool:
    return _run_command(git_subcommand, *args, **kwargs).returncode == 0

class Git:
    FLOWCONFIG_FILE = ".flowconfig"

    command: list[str]
    check_returncode: bool = True

    @staticmethod
    def flow_config(*args, **kwargs):
        return Git("config", *args, file=FLOWCONFIG_FILE, **kwargs)

    @staticmethod
    def get_current_branch():
        return Git("rev-parse", "HEAD", abbrev_ref=True).firstline()

    @staticmethod
    def get_branches():
        return Git._get_references("heads")

    @staticmethod
    def status():
        return list(map(Status, Git("status", porcelain=True).lines(False)))

    @staticmethod
    def _get_references(kind: str):
        prefix = "refs/" + kind + "/"
        return map(
            lambda r: r.removeprefix(prefix),
            filter(
                lambda l: l.startswith(prefix),
                Git("for-each-ref", format="%(refname)").lines()
            )
        )

    def __init__(self, subcommand: str, *args: str, **kwargs: str | int | bool) -> None:
        self.command = ["git", subcommand]

        for option, value in kwargs.items():
            option = option.replace("_", "-")

            if isinstance(value, bool):
                if len(option) == 1:
                    self.command.append("-" + option)
                else:
                    prefix = "--" if value else "--no-"
                    self.command.append(prefix + option)
            else:
                value = value if isinstance(value, str) else str(value)

                if len(option) == 1:
                    self.command.append("-" + option)
                    self.command.append(value)
                else:
                    self.command.append("--" + option + '=' + value)

        self.command += args

    def lines(self, strip: bool = True):
        process = subprocess.run(self.command, capture_output=True, text=True)

        self.__print_process(process.stdout, process.stderr)

        if self.check_returncode:
            process.check_returncode()

        lines = process.stdout.splitlines()

        return lines if not strip else list(map(lambda l: l.strip(), lines))

    def firstline(self):
        lines = self.lines()

        return lines[0] if lines else ""

    def exec(self):
        process = subprocess.run(self.command, capture_output=True, text=True)

        self.__print_process(process.stdout, process.stderr)

        if self.check_returncode:
            process.check_returncode()

    def code(self):
        process = subprocess.run(self.command, capture_output=True, text=True)

        self.__print_process(process.stdout, process.stderr)

        return process.returncode

    def without_checking(self):
        self.check_returncode = False

        return self

    def __print_process(self, stdout: str, stderr: str):
        command = []

        for arg in self.command:
            if arg.startswith("--") and "=" in arg:
                pos = arg.index("=")
                arg = arg[:pos+1] + '"' + arg[pos+1:] + '"'
            elif " " in arg:
                arg = '"' + arg + '"'

            command.append(arg)

        print(COLOR_YELLOW + "> " + " ".join(command) + COLOR_RESET)

        for line in stdout.splitlines():
            print(COLOR_BLUE + "< " + line + COLOR_RESET)

        for line in stderr.splitlines():
            print(COLOR_RED + "! " + line + COLOR_RESET)