Red-DiscordBot/redbot/cogs/downloader/installable.py

from __future__ import annotations

import json
import distutils.dir_util
import shutil
from enum import IntEnum
from pathlib import Path
from typing import MutableMapping, Any, TYPE_CHECKING, Optional, Dict, Union, Callable, Tuple, cast

from .log import log
from .json_mixins import RepoJSONMixin

from redbot.core import __version__, version_info as red_version_info, VersionInfo

if TYPE_CHECKING:
    from .repo_manager import RepoManager, Repo


class InstallableType(IntEnum):
    # using IntEnum, because hot-reload breaks its identity
    UNKNOWN = 0
    COG = 1
    SHARED_LIBRARY = 2


class Installable(RepoJSONMixin):
    """Base class for anything the Downloader cog can install.

     - Modules
     - Repo Libraries
     - Other stuff?

    The attributes of this class will mostly come from the installation's
    info.json.

    Attributes
    ----------
    repo_name : `str`
        Name of the repository which this package belongs to.
    repo : Repo, optional
        Repo object of the Installable, if repo is missing this will be `None`
    commit : `str`, optional
        Installable's commit. This is not the same as ``repo.commit``
    author : `tuple` of `str`, optional
        Name(s) of the author(s).
    bot_version : `tuple` of `int`
        The minimum bot version required for this installation. Right now
        this is always :code:`3.0.0`.
    min_python_version : `tuple` of `int`
        The minimum python version required for this cog. This field will not
        apply to repo info.json's.
    hidden : `bool`
        Whether or not this cog will be hidden from the user when they use
        `Downloader`'s commands.
    required_cogs : `dict`
        In the form :code:`{cog_name : repo_url}`, these are cogs which are
        required for this installation.
    requirements : `tuple` of `str`
        Required libraries for this installation.
    tags : `tuple` of `str`
        List of tags to assist in searching.
    type : `int`
        The type of this installation, as specified by
        :class:`InstallationType`.

    """

    def __init__(self, location: Path, repo: Optional[Repo] = None, commit: str = ""):
        """Base installable initializer.

        Parameters
        ----------
        location : pathlib.Path
            Location (file or folder) to the installable.
        repo : Repo, optional
            Repo object of the Installable, if repo is missing this will be `None`
        commit : str
            Installable's commit. This is not the same as ``repo.commit``

        """
        super().__init__(location)

        self._location = location

        self.repo = repo
        self.repo_name = self._location.parent.stem
        self.commit = commit

        self.author: Tuple[str, ...] = ()
        self.min_bot_version = red_version_info
        self.max_bot_version = red_version_info
        self.min_python_version = (3, 5, 1)
        self.hidden = False
        self.disabled = False
        self.required_cogs: Dict[str, str] = {}  # Cog name -> repo URL
        self.requirements: Tuple[str, ...] = ()
        self.tags: Tuple[str, ...] = ()
        self.type = InstallableType.UNKNOWN

        if self._info_file.exists():
            self._process_info_file(self._info_file)

        if self._info == {}:
            self.type = InstallableType.COG

    def __eq__(self, other: Any) -> bool:
        # noinspection PyProtectedMember
        return self._location == other._location

    def __hash__(self) -> int:
        return hash(self._location)

    @property
    def name(self) -> str:
        """`str` : The name of this package."""
        return self._location.stem

    async def copy_to(self, target_dir: Path) -> bool:
        """
        Copies this cog/shared_lib to the given directory. This
        will overwrite any files in the target directory.

        :param pathlib.Path target_dir: The installation directory to install to.
        :return: Status of installation
        :rtype: bool
        """
        copy_func: Callable[..., Any]
        if self._location.is_file():
            copy_func = shutil.copy2
        else:
            # clear copy_tree's cache to make sure missing directories are created (GH-2690)
            distutils.dir_util._path_created = {}
            copy_func = distutils.dir_util.copy_tree

        # noinspection PyBroadException
        try:
            copy_func(src=str(self._location), dst=str(target_dir / self._location.stem))
        except:  # noqa: E722
            log.exception("Error occurred when copying path: {}".format(self._location))
            return False
        return True

    def _read_info_file(self) -> None:
        super()._read_info_file()

        if self._info_file.exists():
            self._process_info_file()

    def _process_info_file(
        self, info_file_path: Optional[Path] = None
    ) -> MutableMapping[str, Any]:
        """
        Processes an information file. Loads dependencies among other
        information into this object.

        :type info_file_path:
        :param info_file_path: Optional path to information file, defaults to `self.__info_file`
        :return: Raw information dictionary
        """
        info_file_path = info_file_path or self._info_file
        if info_file_path is None or not info_file_path.is_file():
            raise ValueError("No valid information file path was found.")

        info: Dict[str, Any] = {}
        with info_file_path.open(encoding="utf-8") as f:
            try:
                info = json.load(f)
            except json.JSONDecodeError:
                info = {}
                log.exception("Invalid JSON information file at path: {}".format(info_file_path))
            else:
                self._info = info

        try:
            author = tuple(info.get("author", []))
        except ValueError:
            author = ()
        self.author = author

        try:
            min_bot_version = VersionInfo.from_str(str(info.get("min_bot_version", __version__)))
        except ValueError:
            min_bot_version = self.min_bot_version
        self.min_bot_version = min_bot_version

        try:
            max_bot_version = VersionInfo.from_str(str(info.get("max_bot_version", __version__)))
        except ValueError:
            max_bot_version = self.max_bot_version
        self.max_bot_version = max_bot_version

        try:
            min_python_version = tuple(info.get("min_python_version", (3, 5, 1)))
        except ValueError:
            min_python_version = self.min_python_version
        self.min_python_version = min_python_version

        try:
            hidden = bool(info.get("hidden", False))
        except ValueError:
            hidden = False
        self.hidden = hidden

        try:
            disabled = bool(info.get("disabled", False))
        except ValueError:
            disabled = False
        self.disabled = disabled

        self.required_cogs = info.get("required_cogs", {})

        self.requirements = info.get("requirements", ())

        try:
            tags = tuple(info.get("tags", ()))
        except ValueError:
            tags = ()
        self.tags = tags

        installable_type = info.get("type", "")
        if installable_type in ("", "COG"):
            self.type = InstallableType.COG
        elif installable_type == "SHARED_LIBRARY":
            self.type = InstallableType.SHARED_LIBRARY
            self.hidden = True
        else:
            self.type = InstallableType.UNKNOWN

        return info


class InstalledModule(Installable):
    """Base class for installed modules,
    this is basically instance of installed `Installable`
    used by Downloader.

    Attributes
    ----------
    pinned : `bool`
        Whether or not this cog is pinned, always `False` if module is not a cog.
    """

    def __init__(
        self,
        location: Path,
        repo: Optional[Repo] = None,
        commit: str = "",
        pinned: bool = False,
        json_repo_name: str = "",
    ):
        super().__init__(location=location, repo=repo, commit=commit)
        self.pinned: bool = pinned if self.type == InstallableType.COG else False
        # this is here so that Downloader could use real repo name instead of "MISSING_REPO"
        self._json_repo_name = json_repo_name

    def to_json(self) -> Dict[str, Union[str, bool]]:
        module_json: Dict[str, Union[str, bool]] = {
            "repo_name": self.repo_name,
            "module_name": self.name,
            "commit": self.commit,
        }
        if self.type == InstallableType.COG:
            module_json["pinned"] = self.pinned
        return module_json

    @classmethod
    def from_json(
        cls, data: Dict[str, Union[str, bool]], repo_mgr: RepoManager
    ) -> InstalledModule:
        repo_name = cast(str, data["repo_name"])
        cog_name = cast(str, data["module_name"])
        commit = cast(str, data.get("commit", ""))
        pinned = cast(bool, data.get("pinned", False))

        # TypedDict, where are you :/
        repo = repo_mgr.get_repo(repo_name)
        if repo is not None:
            repo_folder = repo.folder_path
        else:
            repo_folder = repo_mgr.repos_folder / "MISSING_REPO"

        location = repo_folder / cog_name

        return cls(
            location=location, repo=repo, commit=commit, pinned=pinned, json_repo_name=repo_name
        )

    @classmethod
    def from_installable(cls, module: Installable, *, pinned: bool = False) -> InstalledModule:
        return cls(
            location=module._location, repo=module.repo, commit=module.commit, pinned=pinned
        )