repositories.py

import contextlib
import glob
import json
import os
import shutil
import sys
import traceback
from datetime import datetime
from pathlib import Path
from stat import S_IREAD, S_IWRITE
from urllib.request import urlretrieve

import cadetrdm
from cadetrdm.io_utils import recursive_chmod, write_lines_to_file, wait_for_user, init_lfs
from cadetrdm.jupyter_functionality import Notebook
from cadetrdm.remote_integration import GitHubRemote, GitLabRemote
from cadetrdm.logging import OutputLog

try:
    import git
except ImportError:
    # Adding this hint to save users the confusion of trying $pip install git
    raise ImportError("No module named git, please install the gitpython package")

from cadetrdm.web_utils import ssh_url_to_http_url
from cadetrdm.io_utils import delete_path


def validate_is_output_repo(path_to_repo):
    with open(os.path.join(path_to_repo, ".cadet-rdm-data.json"), "r") as file_handle:
        rdm_data = json.load(file_handle)
        if rdm_data["is_project_repo"]:
            raise ValueError("Please use the URL to the output repository.")


class BaseRepo:
    def __init__(self, repository_path=None, search_parent_directories=True, *args, **kwargs):
        """
        Base class handling most git workflows.

        :param repository_path:
            Path to the root directory of the repository.
        :param search_parent_directories:
            if True, all parent directories will be searched for a valid repo as well.

            Please note that this was the default behaviour in older versions of GitPython,
            which is considered a bug though.
        :param args:
            Args handed to git.Repo()
        :param kwargs:
            Kwargs handed to git.Repo()
        """
        if repository_path is None or repository_path == ".":
            repository_path = os.getcwd()

        if type(repository_path) is str:
            repository_path = Path(repository_path)

        self._git_repo = git.Repo(repository_path, search_parent_directories=search_parent_directories, *args, **kwargs)
        self._git = self._git_repo.git

        self._most_recent_branch = self.active_branch.name
        self._earliest_commit = None

        self.add = self._git.add

    @property
    def active_branch(self):
        return self._git_repo.active_branch

    @property
    def untracked_files(self):
        return self._git_repo.untracked_files

    @property
    def current_commit_hash(self):
        return str(self.head.commit)

    @property
    def path(self):
        return Path(self._git_repo.working_dir)

    @property
    def bare(self):
        return self._git_repo.bare

    @property
    def working_dir(self):
        print("Deprecation Warning. .working_dir is getting replaced with .path")
        return Path(self._git_repo.working_dir)

    @property
    def head(self):
        return self._git_repo.head

    @property
    def remotes(self):
        return self._git_repo.remotes

    @property
    def remote_urls(self):
        if len(self.remotes) == 0:
            print(RuntimeWarning(f"No remote for repo at {self.path} set yet. Please add remote ASAP."))
        return [str(remote.url) for remote in self.remotes]

    @property
    def earliest_commit(self):
        if self._earliest_commit is None:
            *_, earliest_commit = self._git_repo.iter_commits()
            self._earliest_commit = earliest_commit
        return self._earliest_commit

    @property
    def tags(self):
        return list()

    @property
    def data_json_path(self):
        return self.path / ".cadet-rdm-data.json"

    @property
    def cache_json_path(self):
        return self.path / ".cadet-rdm-cache.json"

    @property
    def has_changes_upstream(self):
        try:
            remote_hash = str(self.remotes[0].fetch()[0].commit)

            if self.current_commit_hash != remote_hash:
                return True
            else:
                return False

        except git.GitCommandError as e:
            traceback.print_exc()
            print(f"Git command error in {self.path}: {e}")

    def fetch(self):
        self._git.fetch()

    def update(self):
        try:
            self.fetch()

            if self.has_changes_upstream:
                print(f"New changes detected in {self.remotes[0]}, pulling updates...")
                self.remotes[0].pull()
                self._git.reset(["--hard", f"{self.remotes[0]}/{self.active_branch}"])

        except git.GitCommandError as e:
            traceback.print_exc()
            print(f"Git command error in {self.path}: {e}")

    def add_remote(self, remote_url, remote_name=None):
        """
        Add a remote to the repository.

        :param remote_url:
        :param remote_name:
        :return:
        """
        if remote_name is None:
            remote_name = "origin"
        self._git_repo.create_remote(remote_name, url=remote_url)
        with open(self.data_json_path, "r") as handle:
            rdm_data = json.load(handle)
        if rdm_data["is_project_repo"]:
            # This folder is a project repo. Use a project repo class to easily access the output repo.
            output_repo = ProjectRepo(self.path).output_repo

            if output_repo.active_branch != "main":
                if output_repo.exist_uncomitted_changes:
                    output_repo.stash_all_changes()
                output_repo.checkout("main")

            output_repo.add_list_of_remotes_in_readme_file("project_repo", self.remote_urls)
            output_repo.commit("Add remote for project repo", verbosity=0)
        if rdm_data["is_output_repo"]:
            # This folder is an output repo
            project_repo = ProjectRepo(self.path.parent)
            project_repo.update_output_remotes_json()
            project_repo.add_list_of_remotes_in_readme_file("output_repo", self.remote_urls)
            project_repo.commit("Add remote for output repo", verbosity=0)

    def add_filetype_to_lfs(self, file_type):
        """
        Add the filetype given in file_type to the GIT-LFS tracking

        :param file_type:
        Wildcard formatted string. Examples: "*.png" or "*.xlsx"
        :return:
        """
        init_lfs(lfs_filetypes=[file_type], path=self.path)
        self.add_all_files()
        self.commit(f"Add {file_type} to lfs")

    def import_remote_repo(self, source_repo_location, source_repo_branch, target_repo_location=None):
        """
        Import a remote repo and update the cadet-rdm-cache