mirror of
https://github.com/Cog-Creators/Red-DiscordBot.git
synced 2025-11-06 03:08:55 -05:00
284 lines
9.1 KiB
Python
284 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import functools
|
|
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.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:
|
|
copy_func = functools.partial(shutil.copytree, dirs_exist_ok=True)
|
|
|
|
# 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: %s", 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:
|
|
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
|
|
)
|