repositories.py

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

from tabulate import tabulate

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 create_gitlab_remote, create_github_remote
from cadetrdm.version import version as cadetrdm_version

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=False, *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 working_dir(self):
        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.working_dir} 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.working_dir / ".cadet-rdm-data.json"

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

    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.working_dir).output_repo

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

            output_repo.add_list_of_remotes_in_readme_file("project_repo", self.remote_urls)
            output_repo.commit("Add remote for project repo")
        if rdm_data["is_output_repo"]:
            # This folder is an output repo
            project_repo = ProjectRepo(self.working_dir.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")

    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.working_dir)
        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

        :param source_repo_location:
        Path or URL to the source repo.
        Example https://jugit.fz-juelich.de/IBG-1/ModSim/cadet/agile_cadet_rdm_presentation_output.git
        or git@jugit.fz-juelich.de:IBG-1/ModSim/cadet/agile_cadet_rdm_presentation_output.git

        :param source_repo_branch:
        Branch of the source repo to check out.

        :param target_repo_location:
        Place to store the repo. If None, the external_cache folder is used.

        :return:
        Path to the cloned repository
        """
        if target_repo_location is None:
            target_repo_location = self.working_dir / "external_cache" / source_repo_location.split("/")[-1]
        else:
            target_repo_location = self.working_dir / target_repo_location

        self.add_path_to_gitignore(target_repo_location)

        print(f"Cloning from {source_repo_location} into {target_repo_location}")
        multi_options = ["--filter=blob:none", "--branch", source_repo_branch, "--single-branch"]
        repo = git.Repo.clone_from(source_repo_location, target_repo_location, multi_options=multi_options)
        repo.git.clear_cache()
        repo.close()

        self.update_cadet_rdm_cache_json(source_repo_branch=source_repo_branch,
                                         target_repo_location=target_repo_location,
                                         source_repo_location=source_repo_location)
        return target_repo_location

    def add_path_to_gitignore(self, path_to_be_ignored):
        """
        Add the path to the .gitignore file