path: root/updateGit.py

#!/usr/bin/env python3

import os
import subprocess
import sys
import contextvars
from concurrent.futures import ThreadPoolExecutor
from threading import Lock

SAFE_BRANCHES = ["develop", "master", "main", "rc", "release"]
DEBUG = False
original_print = print
p_print_lock = Lock()  # Print Lock
c_print_lock = Lock()  # Color Lock

repository_path = contextvars.ContextVar("repository", default="")


def print(*args, **kwargs):
    with p_print_lock:
        original_print(" ".join(map(str, args)), **kwargs)


def goodPrint(*args, **kwargs):
    with c_print_lock:
        print("\033[32m", end="")
        if path := repository_path.get():
            print(f"[{path}]", end=" ")
        print(*args, **kwargs, end="\033[00m\n")


def warnPrint(*args, **kwargs):
    with c_print_lock:
        print("\033[33m", end="")
        if path := repository_path.get():
            print(f"[{path}]", end=" ")
        print(*args, **kwargs, end="\033[00m\n")


def errorPrint(*args, **kwargs):
    with c_print_lock:
        print("\033[31m", end="")
        if path := repository_path.get():
            print(f"[{path}]", end=" ")
        print(*args, **kwargs, end="\033[00m\n")


def debugPrint(*args, **kwargs):
    if DEBUG:
        if path := repository_path.get():
            print(f"[{path}]", end=" ")
        print(" ".join(map(str, args)), **kwargs)


def getStdout(command):
    process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    if process.returncode != 0:
        errorPrint("Failed to execute command:", command)
        if process.stdout:
            for line in process.stdout.decode("UTF-8").strip().split("\n"):
                errorPrint("  Stdout: ", line)
        if process.stderr:
            for line in process.stderr.decode("UTF-8").strip().split("\n"):
                errorPrint("  Stderr: ", line)
        return ""
    if process.stdout:
        stdout = process.stdout.decode("UTF-8").strip()
        if stdout:
            for line in stdout.split("\n"):
                debugPrint("  Stdout: ", line)
        return stdout
    return ""


def getGitStdout(path, command):
    debugPrint(f"git({path}): run", command)
    if isinstance(command, str):
        cmd = f"git -C {path} {command}"
        return getStdout(cmd.split())
    if isinstance(command, list):
        cmd = ["git", "-C", path]
        cmd.extend(command)
        return getStdout(cmd)


def contains_changes(git_status_output):
    debugPrint("Parsing for changes:")
    for line in git_status_output.split("\n"):
        debugPrint("      Status Line:", line)
        if line.startswith("A ") or line.startswith("D ") or line.startswith("M "):
            return True
    return False


def update_repository(path):
    friendly_path = os.path.basename(path)
    repository_path.set(friendly_path)
    debugPrint("Starting git update...")
    changed = contains_changes(getGitStdout(path, "status --porcelain"))
    cur_branch = getGitStdout(path, "rev-parse --abbrev-ref HEAD")

    # Fetch changes, delete removed repositories
    getGitStdout(path, "fetch --all --prune")

    debugPrint("Go through branches and merge with upstream")
    # Switched format 'upstream' to 'push'
    branches = getGitStdout(
        path,
        [
            "for-each-ref",
            "--format=%(refname:short)#%(push)#%(upstream:track,nobracket)",
            "refs/heads",
        ],
    ).split("\n")
    remote_branches = getGitStdout(path, ['for-each-ref', '--format=%(refname)', 'refs/remotes/origin']).split('\n')
    for branch in branches:
        local_branch, push_branch, tracking = branch.split("#")
        debugPrint(f"Trying to update {local_branch=} {push_branch=} {tracking=}")
        if tracking == "gone":
            # Upstream was merged, branch is removed later
            warnPrint("Upsteam branch for", local_branch, "is gone.")
            continue
        if push_branch not in remote_branches:
            # warnPrint(local_branch, "does not exist remotely. Either not yet created or already merged/removed.")
            continue
        local = getGitStdout(path, ["rev-parse", "--quiet", "--verify", local_branch])
        remote = getGitStdout(
            path, ["rev-parse", "--quiet", "--verify", push_branch]
        )
        if not remote:
            raise Exception("Why was rev-parse called, if remote_branch does not exist?")
        merge_base = getGitStdout(path, ["merge-base", local_branch, push_branch])
        debugPrint("    Branch:", local_branch)
        debugPrint("  Upstream:", push_branch)
        debugPrint("      Local", local)
        debugPrint("     Remote", remote)
        debugPrint("       Base", merge_base)
        if local == remote:
            continue
        if local_branch == cur_branch and changed:
            warnPrint("Repository has a working branch with changes. Stashing.")
            getGitStdout(
                path,
                [
                    "stash",
                    "push",
                    "--all", # "--include-untracked",
                    "--message",
                    "Automatic stash by updateGit.py",
                ],
            )
        if local == merge_base:  # Remote updates on branch, pull them in
            if local_branch == cur_branch:
                getGitStdout(path, ["merge", remote])
                local_short = getGitStdout(path, ["rev-parse", "--short", local])
                remote_short = getGitStdout(path, ["rev-parse", "--short", remote])
                shortstat = getGitStdout(path, ["diff", "--shortstat", local, remote])
                shortlog = getGitStdout(
                    path,
                    [
                        "log",
                        "--pretty=format:…… %s",
                        "--no-merges",
                        f"{local}..{remote}",
                    ],
                )
                goodPrint(f"Updated branch {cur_branch} {local_short}..{remote_short}")
                goodPrint(shortstat)
                debugPrint(shortlog)
            else:
                getGitStdout(path, ["branch", "-f", local_branch, remote])
                shortstat = getGitStdout(path, ["diff", "--shortstat", local, remote])
                shortlog = getGitStdout(
                    path,
                    [
                        "log",
                        "--pretty=format:…… %s",
                        "--no-merges",
                        f"{local}..{remote}",
                    ],
                )
                goodPrint(f"Updated branch {cur_branch}: {local}..{remote}")
                goodPrint(shortstat)
                debugPrint(shortlog)
        if remote == merge_base:  # Local updates on branch
            warnPrint(f"{local_branch} is ahead. Consider pushing your changes.")
            continue
        if local_branch == cur_branch and local != remote != merge_base:
            warnPrint(
                "Local, Remote and their base are different. Trying to rebase..."
            )
            getGitStdout(path, ["rebase", remote])
        if local_branch == cur_branch and changed:
            warnPrint("Reapplying stashed changes.")
            getGitStdout(path, ["stash", "pop"])
        if local_branch != cur_branch and local != remote != merge_base:
            rebase_check = getGitStdout(path, [
                "log",
                "--oneline",
                "--cherry",
                f"{remote}...{local}",
            ])
            was_just_rebased = all(line.startswith('= ') for line in rebase_check.split('\n'))
            if was_just_rebased:
                goodPrint("Resetting", local_branch, "to origin version after remote rebase.")
                getGitStdout(path, [
                    "branch",
                    "--force",
                    local_branch,
                    f"origin/{local_branch}",
                ])
            else:
                warnPrint(
                    local_branch,
                    "was modified remotely, but not just rebased.",
                )

    debugPrint("Find safe branch from", SAFE_BRANCHES)
    safe_branch = None
    branches = getGitStdout(
        path, ["for-each-ref", "--format=%(refname:short)", "refs/heads"]
    ).split("\n")
    debugPrint("Branches:", branches)
    for branch in SAFE_BRANCHES:
        if branch in branches:
            safe_branch = branch
            debugPrint("Local safe branch:", safe_branch)
            break

    if not safe_branch:
        debugPrint("No local safe branch, trying to find it remotely...")
        branches = getGitStdout(
            path, ["for-each-ref", "--format=%(refname:short)"]
        ).split("\n")
        debugPrint("Branches:", branches)
        for branch in SAFE_BRANCHES:
            for branchname in branches:
                if branchname.endswith(branch):
                    safe_branch = branch
                    debugPrint("Remote safe branch:", safe_branch)
                    safe_branch_switch = getGitStdout(
                        path, ["switch", safe_branch]
                    ).split("\n")[-1]
                    # warnPrint(f"Safe Branch Switch: {safe_branch_switch}.")
                    break
    if not safe_branch:
        warnPrint("The repository seems to not contain a safe branch! Abort!")
        return

    # Checkout safe branch, if detached HEAD and not proceeded
    if cur_branch == "HEAD":
        debugPrint("Should the safe branch be checked out?", cur_branch)
        if changed:
            errorPrint(
                "Head detached and changes found."
                f"Stash them to checkout {safe_branch}."
            )
            return
        revision_head = getGitStdout(path, "rev-parse --quiet --verify HEAD")
        revision_merge = getGitStdout(path, ["merge-base", "HEAD", safe_branch])
        if revision_head == revision_merge:
            warnPrint(
                f"Head detached, checking {safe_branch} out.",
                getGitStdout(path, ["checkout", safe_branch]),
            )
        else:
            warnPrint("Head detached, but not part of safe branch!")
    # Bad, if diverged from origin/master
    # else:
    #     switch = getGitStdout(path, ['switch', safe_branch])
    #     debugPrint(switch)

    # Delete branches, that were already merged to our safe_branch
    # Experiment: Merged into origin/safe_branch, because I sometimes push to local
    merged_branches = (
        getGitStdout(
            path,
            [
                "branch",
                "--format=%(refname:short)",
                f"--merged=origin/{safe_branch}",
            ],
        )
        .strip()
        .split("\n")
    )
    # debugPrint('Deletion of branches', merged_branches, cur_branch)
    for merged_branch in merged_branches:
        if not merged_branch:  # List might contain empty string
            continue
        if merged_branch in SAFE_BRANCHES:
            continue
        get_tracking_status = getGitStdout(
            path,
            [
                "for-each-ref",
                "--format=%(upstream:track,nobracket)",
                "refs/heads/%s" % merged_branch,
            ],
        )
        if merged_branch == cur_branch:
            # TODO: This does not work it seems?
            warnPrint("Current branch is merged. Checking out safe branch.")
            if changed:
                warnPrint("Local changes detected, creating temporary stash")
                getGitStdout(
                    path,
                    [
                        "stash",
                        "push",
                        "--include-untracked",
                        "--message",
                        "Automatic stash by updateGit.py",
                    ],
                )
            getGitStdout(path, ["checkout", safe_branch])
            if changed:
                warnPrint("Unstashing local changes")
                getGitStdout(path, ["stash", "pop"])
        if get_tracking_status.strip() == "gone":
            warnPrint("Removing branch by force, upstream is gone.")
            getGitStdout(path, ["branch", "-D", merged_branch]),
        else:
            # Might fail / not properly merged?
            warnPrint(
                "Not sure, why this condition was here...",
                getGitStdout(path, ["branch", "-d", merged_branch]),
            )

    cur_branch = getGitStdout(path, "rev-parse --abbrev-ref HEAD")
    if cur_branch not in SAFE_BRANCHES:
        warnPrint(f"Currently on {cur_branch}, which is not considered a safe branch.")
    # getGitStdout(path, ["gc", "--quiet", "--no-prune"])


class Version:
    def __init__(self, version):
        self.parts = version.split(".")

    def __lt__(self, other):
        if len(self.parts) != len(other.parts):
            return 0
        for i, part in enumerate(self.parts):
            if part != other.parts[i]:
                return part < other.parts[i]
        return 0

    def __repr__(self):
        return "Version(%s)" % ".".join(self.parts)

    def __str__(self):
        return ".".join(self.parts)


def main():
    paths = []
    if len(sys.argv) == 1:
        paths.append(".")
    else:
        for path in sys.argv[1:]:
            paths.append(path)

    repos = []
    for path in paths:
        for current_dir, dirs, _files in os.walk(path):
            if '/.' in current_dir:
                continue
            if ".git" in dirs and os.path.exists(
                os.path.join(current_dir, ".git", "HEAD")
            ):
                debugPrint("Found repository", current_dir)
                repos.append(current_dir)

    if len(repos) == 0:
        errorPrint("No repositories found.")
        return

    goodPrint("Updating", len(repos), "repositories.")
    # executor = ThreadPoolExecutor(max_workers=len(repos))
    executor = ThreadPoolExecutor(max_workers=8)
    for repo in repos:
        # update_repository(subdir)
        executor.submit(update_repository, repo)
    executor.shutdown()


if __name__ == "__main__":
    main()