oraios--serena/repo_dir_sync.py

# -*- coding: utf-8 -*-
import glob
import os
import shutil
from subprocess import Popen, PIPE
import re
import sys
from typing import List, Optional, Sequence
import platform


def popen(cmd):
    shell = platform.system() != "Windows"
    p = Popen(cmd, shell=shell, stdin=PIPE, stdout=PIPE)
    return p


def call(cmd):
    p = popen(cmd)
    return p.stdout.read().decode("utf-8")


def execute(cmd, exceptionOnError=True):
    """
    :param cmd: the command to execute
    :param exceptionOnError: if True, raise on exception on error (return code not 0); if False return
        whether the call was successful
    :return: True if the call was successful, False otherwise (if exceptionOnError==False)
    """
    p = popen(cmd)
    p.wait()
    success = p.returncode == 0
    if exceptionOnError:
        if not success:
            raise Exception("Command failed: %s" % cmd)
    else:
        return success


def gitLog(path, arg):
    oldPath = os.getcwd()
    os.chdir(path)
    lg = call("git log --no-merges " + arg)
    os.chdir(oldPath)
    return lg


def gitCommit(msg):
    with open(COMMIT_MSG_FILENAME, "wb") as f:
        f.write(msg.encode("utf-8"))
    gitCommitWithMessageFromFile(COMMIT_MSG_FILENAME)


def gitCommitWithMessageFromFile(commitMsgFilename):
    if not os.path.exists(commitMsgFilename):
        raise FileNotFoundError(f"{commitMsgFilename} not found in {os.path.abspath(os.getcwd())}")
    os.system(f"git commit --file={commitMsgFilename}")
    os.unlink(commitMsgFilename)


COMMIT_MSG_FILENAME = "commitmsg.txt"


class OtherRepo:
    SYNC_COMMIT_ID_FILE_LIB_REPO = ".syncCommitId.remote"
    SYNC_COMMIT_ID_FILE_THIS_REPO = ".syncCommitId.this"
    SYNC_COMMIT_MESSAGE = f"Updated %s sync commit identifiers"
    SYNC_BACKUP_DIR = ".syncBackup"
    
    def __init__(self, name, branch, pathToLib):
        self.pathToLibInThisRepo = os.path.abspath(pathToLib)
        if not os.path.exists(self.pathToLibInThisRepo):
            raise ValueError(f"Repository directory '{self.pathToLibInThisRepo}' does not exist")
        self.name = name
        self.branch = branch
        self.libRepo: Optional[LibRepo] = None

    def isSyncEstablished(self):
        return os.path.exists(os.path.join(self.pathToLibInThisRepo, self.SYNC_COMMIT_ID_FILE_LIB_REPO))
    
    def lastSyncIdThisRepo(self):
        with open(os.path.join(self.pathToLibInThisRepo, self.SYNC_COMMIT_ID_FILE_THIS_REPO), "r") as f:
            commitId = f.read().strip()
        return commitId

    def lastSyncIdLibRepo(self):
        with open(os.path.join(self.pathToLibInThisRepo, self.SYNC_COMMIT_ID_FILE_LIB_REPO), "r") as f:
            commitId = f.read().strip()
        return commitId

    def gitLogThisRepoSinceLastSync(self):
        lg = gitLog(self.pathToLibInThisRepo, '--name-only HEAD "^%s" .' % self.lastSyncIdThisRepo())
        lg = re.sub(r'commit [0-9a-z]{8,40}\n.*\n.*\n\s*\n.*\n\s*(\n.*\.syncCommitId\.(this|remote))+', r"", lg, flags=re.MULTILINE)  # remove commits with sync commit id update
        indent = "  "
        lg = indent + lg.replace("\n", "\n" + indent)
        return lg

    def gitLogLibRepoSinceLastSync(self, libRepo: "LibRepo"):
        syncIdFile = os.path.join(self.pathToLibInThisRepo, self.SYNC_COMMIT_ID_FILE_LIB_REPO)
        if not os.path.exists(syncIdFile):
            return ""
        with open(syncIdFile, "r") as f:
            syncId = f.read().strip()
        lg = gitLog(libRepo.libPath, '--name-only HEAD "^%s" .'  % syncId)
        lg = re.sub(r"Sync (\w+)\n\s*\n", r"Sync\n\n", lg, flags=re.MULTILINE)
        indent = "  "
        lg = indent + lg.replace("\n", "\n" + indent)
        return "\n\n" + lg

    def _userInputYesNo(self, question) -> bool:
        result = None
        while result not in ("y", "n"):
            result = input(question + " [y|n]: ").strip()
        return result == "y"

    def pull(self, libRepo: "LibRepo"):
        """
        Pulls in changes from this repository into the lib repo
        """
        # switch to branch in lib repo
        os.chdir(libRepo.rootPath)
        execute("git checkout %s" % self.branch)

        # check if the branch contains the commit that is referenced as the remote commit
        remoteCommitId = self.lastSyncIdLibRepo()
        remoteCommitExists = execute("git rev-list HEAD..%s" % remoteCommitId, exceptionOnError=False)
        if not remoteCommitExists:
            if not self._userInputYesNo(f"\nWARNING: The referenced remote commit {remoteCommitId} does not exist "
                                        f"in your {self.libRepo.name} branch '{self.branch}'!\nSomeone else may have "
                                        f"pulled/pushed in the meantime.\nIt is recommended that you do not continue. "
                                        f"Continue?"):
                return

        # check if this branch is clean
        lgLib = self.gitLogLibRepoSinceLastSync(libRepo).strip()
        if lgLib != "":
            print(f"The following changes have been added to this branch in the library:\n\n{lgLib}\n\n")
            print(f"ERROR: You must push these changes before you can pull or reset this branch to {remoteCommitId}")
            sys.exit(1)

        # get log with relevant commits in this repo that are to be pulled
        lg = self.gitLogThisRepoSinceLastSync()

        os.chdir(libRepo.rootPath)

        # create commit message file
        commitMsg = f"Sync {self.name}\n\n" + lg
        with open(COMMIT_MSG_FILENAME, "w") as f:
            f.write(commitMsg)

        # ask whether to commit these changes
        print("Relevant commits:\n\n" + lg + "\n\n")
        if not self._userInputYesNo(f"The above changes will be pulled from {self.name}.\n"
                f"You may change the commit message by editing {os.path.abspath(COMMIT_MSG_FILENAME)}.\n"
                "Continue?"):
            os.unlink(COMMIT_MSG_FILENAME)
            return

        # prepare restoration of ignored files
        self.prepare_restoration_of_ignored_files(libRepo.rootPath)

        # remove library tree in lib repo
        shutil.rmtree(self.libRepo.libDirectory)

        # copy tree from this repo to lib repo (but drop the sync commit id files)
        shutil.copytree(self.pathToLibInThisRepo, self.libRepo.libDirectory)
        for fn in (self.SYNC_COMMIT_ID_FILE_LIB_REPO, self.SYNC_COMMIT_ID_FILE_THIS_REPO):
            p = os.path.join(self.libRepo.libDirectory, fn)
            if os.path.exists(p):
                os.unlink(p)

        # restore ignored directories/files
        self.restore_ignored_files(libRepo.rootPath)

        # make commit in lib repo
        os.system("git add %s" % self.libRepo.libDirectory)
        gitCommitWithMessageFromFile(COMMIT_MSG_FILENAME)
        newSyncCommitIdLibRepo = call("git rev-parse HEAD").strip()

        # update commit ids in this repo
        os.chdir(self.pathToLibInThisRepo)
        newSyncCommitIdThisRepo = call("git rev-parse HEAD").strip()
        with open(self.SYNC_COMMIT_ID_FILE_LIB_REPO, "w") as f:
            f.write(newSyncCommitIdLibRepo)
        with open(self.SYNC_COMMIT_ID_FILE_THIS_REPO, "w") as f:
            f.write(newSyncCommitIdThisRepo)
        execute('git add %s %s' % (self.SYNC_COMMIT_ID_FILE_LIB_REPO, self.SYNC_COMMIT_ID_FILE_THIS_REPO))
        execute(f'git commit -m "{self.SYNC_COMMIT_MESSAGE % self.libRepo.name} (pull)"')

        print(f"\n\nIf everything was successful, you should now push your changes to branch "
              f"'{self.branch}'\nand get your branch merged into develop (issuing a pull request where appropriate)")
        
    def push(self, libRepo: "LibRepo"):
        """
        Pushes changes from the lib repo to this repo
        """
        os.chdir(libRepo.rootPath)

        # switch to the source repo branch
        execute(f"git checkout {self.branch}")

        if self.isSyncEstablished():

            # check if there are any commits that have not yet been pulled
            unpulledCommits = self.gitLogThisRepoSinceLastSync().strip()
            if unpulledCommits != "":
                print(f"\n{unpulledCommits}\n\n")
                if not self._userInputYesNo(f"WARNING: The above changes in repository '{self.name}' have not"
                                            f" yet been pulled.\nYou might want to pull them.\n"
                                            f"If you continue with the push, they will be lost. Continue?"):
                    return

            # get change log in lib repo since last sync
            libLogSinceLastSync = self.gitLogLibRepoSinceLastSync(libRepo)

            print("Relevant commits:\n\n" + libLogSinceLastSync + "\n\n")
            if not self._userInputYesNo("The above changes will be pushed. Continue?"):
                return
            print()
        else:
            libLogSinceLastSync = ""

        # prepare restoration of ignored files in target repo
        base_dir_this_repo = os.path.join(self.pathToLibInThisRepo, "..")
        self.prepare_restoration_of_ignored_files(base_dir_this_repo)

        # remove the target repo tree and update it with the tree from the source repo
        shutil.rmtree(self.pathToLibInThisRepo)
        shutil.copytree(libRepo.libPath, self.pathToLibInThisRepo)

        # get the commit id of the source repo we just copied
        commitId = call("git rev-parse HEAD").strip()

        # restore ignored directories and files
        self.restore_ignored_files(base_dir_this_repo)

        # go to the target repo
        os.chdir(self.pathToLibInThisRepo)

        # commit new version in this repo
        execute("git add .")
        with open(self.SYNC_COMMIT_ID_FILE_LIB_REPO, "w") as f:
            f.write(commitId)
        execute("git add %s" % self.SYNC_COMMIT_ID_FILE_LIB_REPO)
        gitCommit(f"{self.libRepo.name} {commitId}" + libLogSinceLastSync)
        commitId = call("git rev-parse HEAD").strip()

        # update information on the commit id we just added
        with open(self.SYNC_COMMIT_ID_FILE_THIS_REPO, "w") as f:
            f.write(commitId)
        execute("git add %s" % self.SYNC_COMMIT_ID_FILE_THIS_REPO)
        execute(f'git commit -m "{self.SYNC_COMMIT_MESSAGE % self.libRepo.name} (push)"')

        os.chdir(libRepo.rootPath)
        
        print(f"\n\nIf everything was successful, you should now update the remote branch:\ngit push")

    def prepare_restoration_of_ignored_files(self, base_dir: str):
        """
        :param base_dir: the directory containing the lib directory, to which ignored paths are relative
        """
        cwd = os.getcwd()
        os.chdir(base_dir)

        # ensure backup dir exists and is empty
        if os.path.exists(self.SYNC_BACKUP_DIR):
            shutil.rmtree(self.SYNC_BACKUP_DIR)
        os.mkdir(self.SYNC_BACKUP_DIR)

        # backup ignored, unversioned directories
        for d in self.libRepo.fullyIgnoredUnversionedDirectories:
            if os.path.exists(d):
                shutil.copytree(d, os.path.join(self.SYNC_BACKUP_DIR, d))

        os.chdir(cwd)

    def restore_ignored_files(self, base_dir: str):
        """
        :param base_dir: the directory containing the lib directory, to which ignored paths are relative
        """
        cwd = os.getcwd()
        os.chdir(base_dir)

        # remove fully ignored directories that were overwritten by the sync
        for d in self.libRepo.fullyIgnoredVersionedDirectories + self.libRepo.fullyIgnoredUnversionedDirectories:
            if os.path.exists(d):
                print("Removing overwritten content: %s" % d)
                shutil.rmtree(d)

        # restore directories and files that can be restored via git
        for d in self.libRepo.ignoredDirectories + self.libRepo.fullyIgnoredVersionedDirectories:
            restoration_cmd = "git checkout %s" % d
            print("Restoring: %s" % restoration_cmd)
            os.system(restoration_cmd)
        for pattern in self.libRepo.ignoredFileGlobPatterns:
            for path in glob.glob(pattern, recursive=True):
                print("Restoring via git: %s" % path)
                os.system("git checkout %s" % path)

        # restore directories that were backed up
        for d in self.libRepo.fullyIgnoredUnversionedDirectories:
            if os.path.exists(os.path.join(self.SYNC_BACKUP_DIR, d)):
                print("Restoring from backup: %s" % d)
                shutil.copytree(os.path.join(self.SYNC_BACKUP_DIR, d), d)

        # remove backup dir
        shutil.rmtree(self.SYNC_BACKUP_DIR)

        os.chdir(cwd)


class LibRepo:
    def __init__(self, name: str, libDirectory: str,
            ignoredDirectories: Sequence[str] = (),
            fullyIgnoredVersionedDirectories: Sequence[str] = (),
            fullyIgnoredUnversionedDirectories: Sequence[str] = (),
            ignoredFileGlobPatterns: Sequence[str] = ()
    ):
        """
        :param name: name of the library being synced
        :param libDirectory: relative path to the library directory within this repo
        :param ignoredDirectories: ignored directories; existing files in ignored directories will be restored
            via 'git checkout' on pull/push, but new files will be added.
            This is useful for configuration-like files, where users may have local changes that should not
            be overwritten, but new files should still be added.
        :param fullyIgnoredVersionedDirectories:
            fully ignored versioned directories will be restored to original state after push/pull via git checkout
        :param fullyIgnoredUnversionedDirectories:
            fully ignored unversioned directories will be backed up and restored to original state after push/pull
        :param ignoredFileGlobPatterns: files matching ignored glob patterns will be restored via 'git checkout'
            on pull/push
        """
        self.name = name
        self.rootPath = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
        self.libDirectory = libDirectory
        self.libPath = os.path.join(self.rootPath, self.libDirectory)
        self.ignoredDirectories: List[str] = list(ignoredDirectories)
        self.fullyIgnoredVersionedDirectories: List[str] = list(fullyIgnoredVersionedDirectories)
        self.fullyIgnoredUnversionedDirectories: List[str] = list(fullyIgnoredUnversionedDirectories)
        self.ignoredFileGlobPatterns: List[str] = list(ignoredFileGlobPatterns)
        self.otherRepos: List[OtherRepo] = []

    def add(self, repo: OtherRepo):
        repo.libRepo = self
        self.otherRepos.append(repo)

    def runMain(self):
        repos = self.otherRepos
        args = sys.argv[1:]
        if len(args) != 2:
            print(f"usage: sync.py <{'|'.join([repo.name for repo in repos])}> <push|pull>")
        else:
            repo = [r for r in repos if r.name == args[0]]
            if len(repo) != 1:
                raise ValueError(f"Unknown repo '{args[0]}'")
            repo = repo[0]

            if args[1] == "push":
                repo.push(self)
            elif args[1] == "pull":
                repo.pull(self)
            else:
                raise ValueError(f"Unknown command '{args[1]}'")