diff --git a/boticordpy/__init__.py b/boticordpy/__init__.py index 5fe812c..9e868ea 100644 --- a/boticordpy/__init__.py +++ b/boticordpy/__init__.py @@ -13,6 +13,5 @@ __copyright__ = "Copyright 2022 Marakarka" __version__ = "2.2.2" from .client import BoticordClient -from .webhook import Webhook from .types import * diff --git a/boticordpy/autopost.py b/boticordpy/autopost.py index 6c09de9..c463f09 100644 --- a/boticordpy/autopost.py +++ b/boticordpy/autopost.py @@ -28,12 +28,16 @@ class AutoPost: _stats: typing.Any _task: typing.Optional["asyncio.Task[None]"] + bot_id: str + def __init__(self, client): self.client = client self._stopped: bool = False self._interval: int = 900 self._task: typing.Optional["asyncio.Task[None]"] = None + self.bot_id = None + @property def is_running(self) -> bool: """ @@ -145,7 +149,7 @@ class AutoPost: while True: stats = await self._stats() try: - await self.client.http.post_bot_stats(stats) + await self.client.http.post_bot_stats(self.bot_id, stats) except Exception as err: on_error = getattr(self, "_error", None) if on_error: @@ -160,14 +164,21 @@ class AutoPost: await asyncio.sleep(self._interval) - def start(self): + def start(self, bot_id: typing.Union[str, int]): """ Starts the loop. + Args: + bot_id ( Union[:obj:`int`, :obj:`str`] ) + Id of the bot to send stats of. + Raises: :obj:`~.exceptions.InternalException` If there's no callback (for getting stats) provided or the autopost is already running. """ + + self.bot_id = bot_id + if not hasattr(self, "_stats"): raise bexc.InternalException("You must provide stats") diff --git a/boticordpy/client.py b/boticordpy/client.py index 9cbdb76..18c7236 100644 --- a/boticordpy/client.py +++ b/boticordpy/client.py @@ -8,11 +8,6 @@ from .autopost import AutoPost class BoticordClient: """Represents a client that can be used to interact with the BotiCord API. - .. warning:: - - In BotiCord API v2 there are some changes with token. - `Read more here `_ - Note: Remember that every http method can return an http exception. @@ -20,52 +15,47 @@ class BoticordClient: token (:obj:`str`) Your bot's Boticord API Token. version (:obj:`int`) - BotiCord API version (Default: 2) + BotiCord API version (Default: 3) """ __slots__ = ("http", "_autopost", "_token") http: HttpClient - def __init__(self, token: str = None, version: int = 2): + def __init__(self, token: str = None, version: int = 3): self._token = token self._autopost: typing.Optional[AutoPost] = None self.http = HttpClient(token, version) - async def get_bot_info(self, bot_id: int) -> boticord_types.Bot: + async def get_bot_info( + self, bot_id: typing.Union[str, int] + ) -> boticord_types.ResourceBot: """Gets information about specified bot. Args: - bot_id (:obj:`int`) + bot_id (Union[:obj:`str`, :obj:`int`]) Id of the bot Returns: - :obj:`~.types.Bot`: - Bot object. + :obj:`~.types.ResourceBot`: + ResourceBot object. """ response = await self.http.get_bot_info(bot_id) - return boticord_types.Bot(**response) - - async def get_bot_comments(self, bot_id: int) -> list: - """Gets list of comments of specified bot. - - Args: - bot_id (:obj:`int`) - Id of the bot - - Returns: - :obj:`list` [ :obj:`~.types.SingleComment` ]: - List of comments. - """ - response = await self.http.get_bot_comments(bot_id) - return [boticord_types.SingleComment(**comment) for comment in response] + return boticord_types.ResourceBot.from_dict(response) async def post_bot_stats( - self, servers: int = 0, shards: int = 0, users: int = 0 - ) -> dict: + self, + bot_id: typing.Union[str, int], + *, + servers: int = 0, + shards: int = 0, + users: int = 0, + ) -> boticord_types.ResourceBot: """Post Bot's stats. Args: + bot_id (Union[:obj:`str`, :obj:`) + Id of the bot to post stats of. servers ( :obj:`int` ) Bot's servers count shards ( :obj:`int` ) @@ -73,15 +63,15 @@ class BoticordClient: users ( :obj:`int` ) Bot's users count Returns: - :obj:`dict`: - Boticord API Response status + :obj:`~.types.ResourceBot`: + ResourceBot object. """ response = await self.http.post_bot_stats( - {"servers": servers, "shards": shards, "users": users} + bot_id, {"servers": servers, "shards": shards, "users": users} ) - return response + return boticord_types.ResourceBot.from_dict(response) - async def get_server_info(self, server_id: int) -> boticord_types.Server: + async def get_server_info(self, server_id: int): """Gets information about specified server. Args: @@ -124,7 +114,7 @@ class BoticordClient: response = await self.http.post_server_stats(payload) return response - async def get_user_info(self, user_id: int) -> boticord_types.UserProfile: + async def get_user_info(self, user_id: int): """Gets information about specified user. Args: @@ -138,7 +128,7 @@ class BoticordClient: response = await self.http.get_user_info(user_id) return boticord_types.UserProfile(**response) - async def get_user_comments(self, user_id: int) -> boticord_types.UserComments: + async def get_user_comments(self, user_id: int): """Gets comments of specified user. Args: @@ -185,9 +175,7 @@ class BoticordClient: else boticord_types.ShortedLink(**response[0]) ) - async def create_shorted_link( - self, *, code: str, link: str, domain: boticord_types.LinkDomain = 1 - ): + async def create_shorted_link(self, *, code: str, link: str, domain=1): """Creates new shorted link Args: @@ -206,9 +194,7 @@ class BoticordClient: return boticord_types.ShortedLink(**response) - async def delete_shorted_link( - self, code: str, domain: boticord_types.LinkDomain = 1 - ): + async def delete_shorted_link(self, code: str, domain=1): """Deletes shorted link Args: diff --git a/boticordpy/exceptions.py b/boticordpy/exceptions.py index be501c9..0f43862 100644 --- a/boticordpy/exceptions.py +++ b/boticordpy/exceptions.py @@ -1,3 +1,6 @@ +from enum import IntEnum + + class BoticordException(Exception): """Base exception class for boticordpy. This could be caught to handle any exceptions thrown from this library. @@ -29,26 +32,131 @@ class HTTPException(BoticordException): def __init__(self, response): self.response = response - fmt = f"{self.response.reason} (Status code: {self.response.status})" + fmt = f"{HTTPErrors(self.response['error']).name} (Status code: {StatusCodes(self.response['status']).name})" super().__init__(fmt) -class Unauthorized(HTTPException): - """Exception that's thrown when status code 401 occurs.""" +class StatusCodes(IntEnum): + """Status codes of response""" + + SERVER_ERROR = 500 + """Server Error (>500)""" + + TOO_MANY_REQUESTS = 429 + """Too Many Requests""" + + NOT_FOUND = 404 + """Requested resource was not found""" + + FORBIDDEN = 403 + """You don't have access to this resource""" + + UNAUTHORIZED = 401 + """Authorization is required to access this resource""" + + BAD_REQUEST = 400 + """Bad Request""" -class Forbidden(HTTPException): - """Exception that's thrown when status code 403 occurs.""" +class HTTPErrors(IntEnum): + """Errors which BotiCord may return""" + UNKNOWN_ERROR = 0 + """Unknown error""" -class NotFound(HTTPException): - """Exception that's thrown when status code 404 occurs.""" + INTERNAL_SERVER_ERROR = 1 + """Server error (>500)""" + RATE_LIMITED = 2 + """Too many requests""" -class ToManyRequests(HTTPException): - """Exception that's thrown when status code 429 occurs.""" + NOT_FOUND = 3 + """Not found""" + FORBIDDEN = 4 + """Access denied""" -class ServerError(HTTPException): - """Exception that's thrown when status code 500 or 503 occurs.""" + BAD_REQUEST = 5 + """Bad request""" + + UNAUTHORIZED = 6 + """Unauthorized. Authorization required""" + + RPC_ERROR = 7 + """Server error (RPC)""" + + WS_ERROR = 8 + """Server error (WS)""" + + THIRD_PARTY_FAIL = 9 + """Third-party service error""" + + UNKNOWN_USER = 10 + """Unknown user""" + + SHORT_DOMAIN_TAKEN = 11 + """Short link already taken""" + + UNKNOWN_SHORT_DOMAIN = 12 + """Unknown short link""" + + UNKNOWN_LIBRARY = 13 + """Unknown library""" + + TOKEN_INVALID = 14 + """Invalid token""" + + UNKNOWN_RESOURCE = 15 + """Unknown resource""" + + UNKNOWN_TAG = 16 + """Unknown tag""" + + PERMISSION_DENIED = 17 + """Insufficient permissions""" + + UNKNOWN_COMMENT = 18 + """Unknown comment""" + + UNKNOWN_BOT = 19 + """Unknown bot""" + + UNKNOWN_SERVER = 20 + """Unknown server""" + + UNKNOWN_BADGE = 21 + """Unknown badge""" + + USER_ALREADY_HAS_A_BADGE = 22 + """User already has a badge""" + + INVALID_INVITE_CODE = 23 + """Invalid invite code""" + + SERVER_ALREADY_EXISTS = 24 + """Server already exists""" + + BOT_NOT_PRESENT_ON_QUEUE_SERVER = 25 + """Bot not present on queue server""" + + UNKNOWN_UP = 26 + """Unknown up""" + + TOO_MANY_UPS = 27 + """Too many ups""" + + INVALID_STATUS = 28 + """Invalid resource status""" + + UNKNOWN_REPORT = 29 + """Unknown report""" + + UNSUPPORTED_MEDIA_TYPE = 30 + """Unsupported media type. Should be one of""" + + UNKNOWN_APPLICATION = 31 + """Unknown application""" + + AUTOMATED_REQUESTS_NOT_ALLOWED = 32 + """Please confirm that you are not a robot by refreshing the page""" diff --git a/boticordpy/http.py b/boticordpy/http.py index c8ca187..77c0c63 100644 --- a/boticordpy/http.py +++ b/boticordpy/http.py @@ -1,4 +1,5 @@ import asyncio +import typing import aiohttp @@ -20,9 +21,9 @@ class HttpClient: loop: `asyncio loop` """ - def __init__(self, auth_token: str, version: int = 1, **kwargs): + def __init__(self, auth_token: str = None, version: int = 3, **kwargs): self.token = auth_token - self.API_URL = f"https://api.boticord.top/v{version}/" + self.API_URL = f"https://api.arbuz.pro/" loop = kwargs.get("loop") or asyncio.get_event_loop() @@ -31,44 +32,30 @@ class HttpClient: async def make_request(self, method: str, endpoint: str, **kwargs): """Send requests to the API""" - kwargs["headers"] = { - "Content-Type": "application/json", - "Authorization": self.token, - } + kwargs["headers"] = {"Content-Type": "application/json"} + + if self.token is not None: + kwargs["headers"]["Authorization"] = self.token url = f"{self.API_URL}{endpoint}" async with self.session.request(method, url, **kwargs) as response: data = await response.json() - if response.status == 200: - return data - elif response.status == 401: - raise exceptions.Unauthorized(response) - elif response.status == 403: - raise exceptions.Forbidden(response) - elif response.status == 404: - raise exceptions.NotFound(response) - elif response.status == 429: - raise exceptions.ToManyRequests(response) - elif response.status == 500: - raise exceptions.ServerError(response) - elif response.status == 503: - raise exceptions.ServerError(response) + if response.status == 200 or response.status == 201: + return data["result"] + else: + raise exceptions.HTTPException( + {"status": response.status, "error": data["errors"][0]["code"]} + ) - raise exceptions.HTTPException(response) - - def get_bot_info(self, bot_id: int): + def get_bot_info(self, bot_id: typing.Union[str, int]): """Get information about the specified bot""" - return self.make_request("GET", f"bot/{bot_id}") + return self.make_request("GET", f"bots/{bot_id}") - def get_bot_comments(self, bot_id: int): - """Get list of specified bot comments""" - return self.make_request("GET", f"bot/{bot_id}/comments") - - def post_bot_stats(self, stats: dict): + def post_bot_stats(self, bot_id: typing.Union[str, int], stats: dict): """Post bot's stats""" - return self.make_request("POST", "stats", json=stats) + return self.make_request("POST", f"bots/{bot_id}/stats", json=stats) def get_server_info(self, server_id: int): """Get information about specified server""" diff --git a/boticordpy/types.py b/boticordpy/types.py index 8998a36..7bd826a 100644 --- a/boticordpy/types.py +++ b/boticordpy/types.py @@ -1,375 +1,588 @@ -import typing -from enum import IntEnum +from datetime import datetime, timezone +from enum import IntEnum, Enum, EnumMeta +import copy +from dataclasses import _is_dataclass_instance, fields, dataclass +from typing import ( + Dict, + Union, + Generic, + Tuple, + TypeVar, + get_origin, + get_args, + Optional, + List, +) +from sys import modules +from itertools import chain -KT = typing.TypeVar("KT") -VT = typing.TypeVar("VT") +from typing_extensions import get_type_hints -def parse_response_dict(input_data: dict) -> dict: - data = input_data.copy() - - for key, value in data.copy().items(): - converted_key = "".join( - ["_" + x.lower() if x.isupper() else x for x in key] - ).lstrip("_") - - if key != converted_key: - del data[key] - - data[converted_key] = value - - return data +KT = TypeVar("KT") +VT = TypeVar("VT") +T = TypeVar("T") -def parse_with_information_dict(bot_data: dict) -> dict: - data = bot_data.copy() +class Singleton(type): + # Thanks to this stackoverflow answer (method 3): + # https://stackoverflow.com/q/6760685/12668716 + _instances = {} - for key, value in data.copy().items(): - if key.lower() == "links": - converted_key = "page_links" - else: - converted_key = "".join( - ["_" + x.lower() if x.isupper() else x for x in key] - ).lstrip("_") - - if key != converted_key: - del data[key] - - if key.lower() == "information": - for information_key, information_value in value.copy().items(): - converted_information_key = "".join( - ["_" + x.lower() if x.isupper() else x for x in information_key] - ).lstrip("_") - - data[converted_information_key] = information_value - - del data["information"] - else: - data[converted_key] = value - - return data + def __call__(cls, *args, **kwargs): + if cls not in cls._instances: + cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) + return cls._instances[cls] -def parse_user_comments_dict(response_data: dict) -> dict: - data = response_data.copy() +class TypeCache(metaclass=Singleton): + # Thanks to Pincer Devs. This class is from the Pincer Library. + cache = {} - for key, value in data.copy().items(): - data[key] = [SingleComment(**comment) for comment in value] + def __init__(self): + lcp = modules.copy() + for module in lcp: + if not module.startswith("melisa"): + continue - return data + TypeCache.cache.update(lcp[module].__dict__) -class ApiData(dict, typing.MutableMapping[KT, VT]): - """Base class used to represent received data from the API.""" +def _asdict_ignore_none(obj: Generic[T]) -> Union[Tuple, Dict, T]: + """ + Returns a dict from a dataclass that ignores + all values that are None + Modification of _asdict_inner from dataclasses + Parameters + ---------- + obj: Generic[T] + The object to convert + Returns + ------- + A dict without None values + """ - def __init__(self, **kwargs: VT) -> None: - super().__init__(**parse_response_dict(kwargs)) - self.__dict__ = self + print(obj) + + if _is_dataclass_instance(obj): + result = [] + for f in fields(obj): + value = _asdict_ignore_none(getattr(obj, f.name)) + + if isinstance(value, Enum): + result.append((f.name, value.value)) + elif not f.name.startswith("_"): + result.append((f.name, value)) + + return dict(result) + + elif isinstance(obj, tuple) and hasattr(obj, "_fields"): + return type(obj)(*[_asdict_ignore_none(v) for v in obj]) + + elif isinstance(obj, (list, tuple)): + return type(obj)(_asdict_ignore_none(v) for v in obj) + + elif isinstance(obj, datetime): + return str(round(obj.timestamp() * 1000)) + + elif isinstance(obj, dict): + return type(obj)( + (_asdict_ignore_none(k), _asdict_ignore_none(v)) for k, v in obj.items() + ) + else: + return copy.deepcopy(obj) -class SingleComment(ApiData): - """This model represents single comment""" +class APIObjectBase: + """ + Represents an object which has been fetched from the BotiCord API. + """ - user_id: str - """Comment's author Id (`str`)""" + def __attr_convert(self, attr_value: Dict, attr_type: T) -> T: + factory = attr_type - text: str - """Comment content""" + # Always use `__factory__` over __init__ + if getattr(attr_type, "__factory__", None): + factory = attr_type.__factory__ - vote: int - """Comment vote value (`-1,` `0`, `1`)""" + if attr_value is None: + return None - is_updated: bool - """Was comment updated?""" + if attr_type is not None and isinstance(attr_value, attr_type): + return attr_value - created_at: int - """Comment Creation date timestamp""" + if isinstance(attr_value, dict): + return factory(attr_value) - updated_at: int - """Last edit date timestamp""" + return factory(attr_value) - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) + def __post_init__(self): + TypeCache() + + attributes = chain.from_iterable( + get_type_hints(cls, globalns=TypeCache.cache).items() + for cls in chain(self.__class__.__bases__, (self,)) + ) + + for attr, attr_type in attributes: + # Ignore private attributes. + if attr.startswith("_"): + continue + + types = self.__get_types(attr, attr_type) + + types = tuple(filter(lambda tpe: tpe is not None, types)) + + if not types: + raise ValueError( + f"Attribute `{attr}` in `{type(self).__name__}` only " + "consisted of missing/optional type!" + ) + + specific_tp = types[0] + + attr_gotten = getattr(self, attr) + + if tp := get_origin(specific_tp): + specific_tp = tp + + if isinstance(specific_tp, EnumMeta) and not attr_gotten: + attr_value = None + elif tp == list and attr_gotten and (classes := get_args(types[0])): + attr_value = [ + self.__attr_convert(attr_item, classes[0]) + for attr_item in attr_gotten + ] + elif tp == dict and attr_gotten and (classes := get_args(types[0])): + attr_value = { + key: self.__attr_convert(value, classes[1]) + for key, value in attr_gotten.items() + } + else: + attr_value = self.__attr_convert(attr_gotten, specific_tp) + + setattr(self, attr, attr_value) + + def __get_types(self, attr: str, arg_type: type) -> Tuple[type]: + origin = get_origin(arg_type) + + if origin is Union: + # Ahh yes, typing module has no type annotations for this... + # noinspection PyTypeChecker + args: Tuple[type] = get_args(arg_type) + + if 2 <= len(args) < 4: + return args + + raise ValueError( + f"Attribute `{attr}` in `{type(self).__name__}` has too many " + f"or not enough arguments! (got {len(args)} expected 2-3)" + ) + + return (arg_type,) + + @classmethod + def __factory__(cls: Generic[T], *args, **kwargs) -> T: + return cls.from_dict(*args, **kwargs) + + def __repr__(self): + attrs = ", ".join( + f"{k}={v!r}" + for k, v in self.__dict__.items() + if v and not k.startswith("_") + ) + + return f"{type(self).__name__}({attrs})" + + def __str__(self): + if _name := getattr(self, "__name__", None): + return f"{_name} {self.__class__.__name__.lower()}" + + return super().__str__() + + def to_dict(self) -> Dict: + """ + Transform the current object to a dictionary representation. Parameters that + start with an underscore are not serialized. + """ + return _asdict_ignore_none(self) -class Bot(ApiData): - """This model represents a bot, returned from the BotiCord API""" +class BotLibrary(IntEnum): + """The library that the bot is based on""" + + DISCORD4J = 1 + DISCORDCR = 2 + DISCORDGO = 3 + DISCORDDOO = 4 + DSHARPPLUS = 5 + DISCORDJS = 6 + DISCORDNET = 7 + DISCORDPY = 8 + ERIS = 9 + JAVACORD = 10 + JDA = 11 + OTHER = 12 + + +class ResourceStatus(IntEnum): + """Bot status on monitoring""" + + HIDDEN = 0 + """Bot is hidden""" + + PUBLIC = 1 + """Bot is public""" + + BANNED = 2 + """Bot is banned""" + + PENDING = 3 + """Bor is pending""" + + +class BotTag(IntEnum): + """Tags of the bot""" + + MODERATION = 0 + """Moderation""" + + BOT = 1 + """Bot""" + + UTILITIES = 2 + """Utilities""" + + ENTERTAINMENT = 3 + """Entertainment""" + + MUSIC = 4 + """Music""" + + ECONOMY = 5 + """Economy""" + + LOGS = 6 + """Logs""" + + LEVELS = 7 + """Levels""" + + NSFW = 8 + """NSFW (18+)""" + + SETTINGS = 9 + """Settings""" + + ROLE_PLAY = 10 + """Role-Play""" + + MEMES = 11 + """Memes""" + + GAMES = 12 + """Games""" + + AI = 13 + """AI""" + + +@dataclass(repr=False) +class UserLinks(APIObjectBase): + """Links of the userk""" + + vk: Optional[str] + """vk.com""" + + telegram: Optional[str] + """t.me""" + + donate: Optional[str] + """Donate""" + + git: Optional[str] + """Link to git of the user""" + + custom: Optional[str] + """Custom link""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a UserLinks from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a UserLinks. + """ + + self: ResourceUp = super().__new__(cls) + + self.vk = data.get("vk") + self.telegram = data.get("telegram") + self.donate = data.get("donate") + self.git = data.get("git") + self.custon = data.get("custom") + + return self + + +@dataclass(repr=False) +class ResourceUp(APIObjectBase): + """Information about bump (bot/server)""" id: str - """Bot's Id""" + """Bump's id""" - short_code: typing.Optional[str] - """Bot's page short code""" + expires: datetime + """Expiration date. (ATTENTION! When using `to_dict()`, the data may not correspond to the actual data due to the peculiarities of the `datetime` module)""" - page_links: list - """List of bot's page urls""" + @classmethod + def from_dict(cls, data: dict): + """Generate a ResourceUp from the given data. - server: dict - """Bot's support server""" + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a ResourceUp. + """ - bumps: int - """Bumps count""" + self: ResourceUp = super().__new__(cls) - added: str - """How many times users have added the bot?""" + self.id = data["id"] + self.expires = datetime.fromtimestamp( + int(int(data["expires"]) / 1000), tz=timezone.utc + ) - prefix: str - """Bot's commands prefix""" - - permissions: int - """Bot's permissions""" - - tags: list - """Bot's search-tags""" - - developers: list - """List of bot's developers Ids""" - - links: typing.Optional[dict] - """Bot's social medias""" - - library: typing.Optional[str] - """Bot's library""" - - short_description: typing.Optional[str] - """Bot's short description""" - - long_description: typing.Optional[str] - """Bot's long description""" - - badge: typing.Optional[str] - """Bot's badge""" - - stats: dict - """Bot's stats""" - - status: str - """Bot's approval status""" - - def __init__(self, **kwargs): - super().__init__(**parse_with_information_dict(kwargs)) + return self -class Server(ApiData): - """This model represents a server, returned from the Boticord API""" +@dataclass(repr=False) +class ResourceRating(APIObjectBase): + """Rating of bot/server""" + + count: int + """Number of ratings""" + + rating: int + """Rating (from 1 to 5)""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a ResourceRating from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a ResourceRating. + """ + + self: ResourceRating = super().__new__(cls) + + self.count = data["count"] + self.rating = data["rating"] + + return self + + +@dataclass(repr=False) +class PartialUser(APIObjectBase): + """Partial user from BotiCord.""" + + username: str + """Username""" + + discriminator: str + """Discriminator""" + + avatar: Optional[str] + """Avatar of the user""" id: str - """Server's Id""" + """Id of the user""" - short_code: typing.Optional[str] - """Server's page short code""" + socials: UserLinks + """Links of the user""" - status: str - """Server's approval status""" + description: Optional[str] + """Description of the user""" - page_links: list - """List of server's page urls""" + short_description: Optional[str] + """Short description of the user""" - bot: dict - """Bot where this server is used for support users""" + status: Optional[str] + """Status of the user""" + + short_domain: Optional[str] + """Short domain""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a PartialUser from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a PartialUser. + """ + + self: PartialUser = super().__new__(cls) + + self.username = data["username"] + self.discriminator = data["discriminator"] + self.avatar = data.get("avatar") + self.id = data["id"] + self.socials = UserLinks.from_dict(data["socials"]) + self.description = data.get("description") + self.short_description = data.get("shortDescription") + self.status = data.get("status") + self.short_domain = data.get("shortDomain") + + return self + + +@dataclass(repr=False) +class ResourceBot(APIObjectBase): + """Bot published on BotiCord + + .. warning:: + + The result of the reverse conversion (`.to_dict()`) may not match the actual data.""" + + id: str + """ID of the bot""" name: str - """Name of the server""" + """Name of the bot""" - avatar: str - """Server's avatar""" + short_description: str + """Short description of the bot""" - members: list - """Members counts - `[all, online]`""" + description: str + """Description of the bot""" - owner: typing.Optional[str] - """Server's owner Id""" + avatar: Optional[str] + """Avatar of the bot""" - bumps: int - """Bumps count""" + short_link: Optional[str] + """Short link to the bot's page""" - tags: list - """Server's search-tags""" + invite_link: str + """Invite link""" - links: dict - """Server's social medias""" + premium_active: bool + """Is premium status active? (True/False)""" - short_description: typing.Optional[str] - """Server's short description""" + premium_splash_url: Optional[str] + """Link to the splash""" - long_description: typing.Optional[str] - """Server's long description""" + premium_auto_fetch: Optional[bool] + """Is auto-fetch enabled? (True/False)""" - badge: typing.Optional[str] - """Server's badge""" + premium_banner_url: Optional[str] + """Premium banner URL""" - def __init__(self, **kwargs): - super().__init__(**parse_with_information_dict(kwargs)) + owner: str + """Owner of the bot""" + status: ResourceStatus + """Status of the bot""" -class UserProfile(ApiData): - """This model represents profile of user, returned from the Boticord API""" + ratings: List[ResourceRating] + """Bot's ratings""" - id: str - """Id of User""" + prefix: str + """Prefix of the bot""" - status: str - """Status of user""" + discriminator: str + """Bot's discriminator""" - badge: typing.Optional[str] - """User's badge""" + created_date: datetime + """Date when the bot was published""" - short_code: typing.Optional[str] - """User's profile page short code""" + support_server_invite_link: Optional[str] + """Link to the support server""" - site: typing.Optional[str] - """User's website""" + library: Optional[BotLibrary] + """The library that the bot is based on""" - vk: typing.Optional[str] - """User's VK Profile""" + guilds: Optional[int] + """Number of guilds""" - steam: typing.Optional[str] - """User's steam account""" + shards: Optional[int] + """Number of shards""" - youtube: typing.Optional[str] - """User's youtube channel""" + members: Optional[int] + """Number of members""" - twitch: typing.Optional[str] - """User's twitch channel""" + website: Optional[str] + """Link to bot's website""" - git: typing.Optional[str] - """User's github/gitlab (or other git-service) profile""" + tags: List[BotTag] + """List of bot tags""" - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) + up_count: int + """Number of ups""" + ups: List[ResourceUp] + """List of bot's ups""" -class UserComments(ApiData): - """This model represents all the user's comments on every page""" + developers: List[PartialUser] + """List of bot's developers""" - bots: list - """Data from `get_bot_comments` method""" + @classmethod + def from_dict(cls, data: dict): + """Generate a ResourceBot from the given data. - servers: list - """Data from `get_server_comments` method""" + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a ResourceBot. + """ - def __init__(self, **kwargs): - super().__init__(**parse_user_comments_dict(kwargs)) + self: ResourceBot = super().__new__(cls) + self.id = data.get("id") + self.name = data.get("name") + self.short_description = data.get("shortDescription") + self.description = data.get("description") + self.avatar = data.get("avatar") + self.short_link = data.get("shortLink") + self.invite_link = data.get("inviteLink") + self.owner = data.get("owner") + self.prefix = data.get("prefix") + self.discriminator = data.get("discriminator") + self.support_server_invite_link = data.get("support_server_invite") + self.website = data.get("website") + self.up_count = data.get("upCount") -class SimpleBot(ApiData): - """This model represents a short bot information (`id`, `short`). - After that you can get more information about it using `get_bot_info` method.""" + self.premium_active = data["premium"].get("active") + self.premium_splash_url = data["premium"].get("splashURL") + self.premium_auto_fetch = data["premium"].get("autoFetch") + self.premium_banner_url = data["premium"].get("bannerURL") - id: str - """Bot's Id""" + self.status = ResourceStatus(data.get("status")) + self.ratings = [ + ResourceRating.from_dict(rating) for rating in data.get("ratings", []) + ] + self.created_date = datetime.strptime( + data["createdDate"], "%Y-%m-%dT%H:%M:%S.%f%z" + ) + self.library = ( + BotLibrary(data["library"]) if data.get("library") is not None else None + ) + self.tags = [BotTag(tag) for tag in data.get("tags", [])] + self.ups = [ResourceUp.from_dict(up) for up in data.get("ups", [])] + self.developers = [ + PartialUser.from_dict(dev) for dev in data.get("developers", []) + ] - short_code: typing.Optional[str] - """Bot's page short code""" + self.guilds = data.get("guilds") + self.shards = data.get("shards") + self.members = data.get("members") + + return self - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) - -class CommentData(ApiData): - """This model represents comment data (from webhook response)""" - - vote: dict - """Comment vote data""" - - old: typing.Optional[str] - """Old content of the comment""" - - new: typing.Optional[str] - """New content of the comment""" - - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) - - -def parse_webhook_response_dict(webhook_data: dict) -> dict: - data = webhook_data.copy() - - for key, value in data.copy().items(): - if key.lower() == "data": - for data_key, data_value in value.copy().items(): - if data_key == "comment": - data[data_key] = CommentData(**data_value) - else: - converted_data_key = "".join( - ["_" + x.lower() if x.isupper() else x for x in data_key] - ).lstrip("_") - - data[converted_data_key] = data_value - - del data["data"] - else: - data[key] = value - - return data - - -class BumpResponse(ApiData): - """This model represents a webhook response (`bot bump`).""" - - type: str - """Type of response (`bump`)""" - - user: str - """Id of user who did the action""" - - at: int - """Timestamp of the action""" - - def __init__(self, **kwargs): - super().__init__(**parse_webhook_response_dict(kwargs)) - - -class CommentResponse(ApiData): - """This model represents a webhook response (`comment`).""" - - type: str - """Type of response (`comment`)""" - - user: str - """Id of user who did the action""" - - comment: CommentData - """Information about the comment""" - - reason: typing.Optional[str] - """Is comment deleted? so, why?""" - - at: int - """Timestamp of the action""" - - def __init__(self, **kwargs): - super().__init__(**parse_webhook_response_dict(kwargs)) - - -class LinkDomain(IntEnum): - """Domain to short the link""" - - BCORD_CC = 1 - """``bcord.cc`` domain, default""" - - DISCORD_CAMP = 3 - """``discord.camp`` domain""" - - -class ShortedLink(ApiData): - id: int - """Id of shorted link""" - - code: str - """Code of shorted link""" - - owner_i_d: str - """Id of owner of shorted link""" - - domain: str - """Domain of shorted link""" - - views: int - """Link views count""" - - date: int - """Timestamp of link creation moment""" - - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) +class LinkDomain: + pass diff --git a/boticordpy/webhook.py b/boticordpy/webhook.py deleted file mode 100644 index 49aa084..0000000 --- a/boticordpy/webhook.py +++ /dev/null @@ -1,135 +0,0 @@ -import asyncio -import typing - -from .types import BumpResponse, CommentResponse - -from aiohttp import web -import aiohttp - - -class Webhook: - """Represents a client that can be used to work with BotiCord Webhooks. - IP of the server - your machine IP. (`0.0.0.0`) - - Args: - x_hook_key (:obj:`str`) - X-hook-key to check the auth of incoming request. - endpoint_name (:obj:`str`) - Name of endpoint (for example: `/bot`) - - Keyword Arguments: - loop: `asyncio loop` - """ - - __slots__ = ( - "_webserver", - "_listeners", - "_is_running", - "__app", - "_endpoint_name", - "_x_hook_key", - "_loop", - ) - - __app: web.Application - _webserver: web.TCPSite - - def __init__(self, x_hook_key: str, endpoint_name: str, **kwargs) -> None: - self._x_hook_key = x_hook_key - self._endpoint_name = endpoint_name - self._listeners = {} - self.__app = web.Application() - self._is_running = False - self._loop = kwargs.get("loop") or asyncio.get_event_loop() - - def listener(self, response_type: str): - """Decorator to set the listener. - Args: - response_type (:obj:`str`) - Type of response (Check reference page) - """ - - def inner(func): - if not asyncio.iscoroutinefunction(func): - raise TypeError(f"<{func.__qualname__}> must be a coroutine function") - self._listeners[response_type] = func - return func - - return inner - - def register_listener(self, response_type: str, callback: typing.Any): - """Method to set the listener. - Args: - response_type (:obj:`str`) - Type of response (Check reference page) - callback (:obj:`function`) - Coroutine Callback Function - """ - if not asyncio.iscoroutinefunction(callback): - raise TypeError(f"<{func.__qualname__}> must be a coroutine function") - - self._listeners[response_type] = callback - return self - - async def _interaction_handler(self, request: aiohttp.web.Request) -> web.Response: - """Interaction handler""" - auth = request.headers.get("X-Hook-Key") - - if auth == self._x_hook_key: - data = await request.json() - - responder = self._listeners.get(data["type"]) - - if responder is not None: - await responder( - ( - BumpResponse - if data["type"].endswith("_bump") - else CommentResponse - )(**data) - ) - - return web.Response(status=200) - - return web.Response(status=401) - - async def _run(self, port): - self.__app.router.add_post("/" + self._endpoint_name, self._interaction_handler) - - runner = web.AppRunner(self.__app) - await runner.setup() - - self._webserver = web.TCPSite(runner, "0.0.0.0", port) - await self._webserver.start() - - self._is_running = True - - def start(self, port: int) -> None: - """Method to start the webhook server - - Args: - port (:obj:`int`) - Port to start the webserver - """ - self._loop.create_task(self._run(port)) - - @property - def is_running(self) -> bool: - """If the server running?""" - return self._is_running - - @property - def listeners(self) -> dict: - """Dictionary of listeners (`type`: `callback function`)""" - return self._listeners - - @property - def app(self) -> web.Application: - """Web application that handles incoming requests""" - return self.__app - - async def close(self) -> None: - """Stop the webhooks server""" - await self._webserver.stop() - - self._is_running = False diff --git a/requirements.txt b/requirements.txt index ee4ba4f..a05ef8d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ aiohttp +typing_extensions \ No newline at end of file diff --git a/tests/test_convertation.py b/tests/test_convertation.py new file mode 100644 index 0000000..81d39df --- /dev/null +++ b/tests/test_convertation.py @@ -0,0 +1,43 @@ +from boticordpy import types + + +resource_up_dict = {"id": "arbuz123", "expires": "1685262170000"} +resource_rating_dict = {"count": 15, "rating": 5} +resource_bot_dict = { + "id": "947141336451153931", + "name": "BumpBot", + "status": 1, + "createdDate": "2023-05-22T22:29:23.264Z", + "premium": {}, +} + + +def test_resource_up_convertation(): + model_from_dict = types.ResourceUp.from_dict(resource_up_dict) + + assert model_from_dict.id == "arbuz123" + assert ( + model_from_dict.expires.strftime("%Y.%m.%d %H:%M:%S") == "2023.05.28 08:22:50" + ) + + dict_from_model = model_from_dict.to_dict() + + assert dict_from_model == resource_up_dict + + +def test_resource_rating_convertation(): + model_from_dict = types.ResourceRating.from_dict(resource_rating_dict) + + assert model_from_dict.count == 15 + assert model_from_dict.rating == 5 + + dict_from_model = model_from_dict.to_dict() + + assert dict_from_model == resource_rating_dict + + +def test_resource_bot_convertation(): + model_from_dict = types.ResourceBot.from_dict(resource_bot_dict) + + assert int(model_from_dict.created_date.timestamp()) == 1684794563 + assert model_from_dict.status.name == "PUBLIC" diff --git a/tests/test_converting.py b/tests/test_converting.py deleted file mode 100644 index 5073fe9..0000000 --- a/tests/test_converting.py +++ /dev/null @@ -1,141 +0,0 @@ -import pytest - -from boticordpy import types - -single_comment_dict = { - "userID": "525366699969478676", - "text": "aboba", - "vote": 1, - "isUpdated": False, - "createdAt": 1644388399, -} - -bot_data_dict = { - "id": "724663360934772797", - "shortCode": "kerdoku", - "links": [ - "https://boticord.top/bot/724663360934772797", - "https://bcord.cc/b/724663360934772797", - "https://myservers.me/b/724663360934772797", - "https://boticord.top/bot/kerdoku", - "https://bcord.cc/b/kerdoku", - "https://myservers.me/b/kerdoku", - ], - "server": {"id": "724668798874943529", "approved": True}, - "information": { - "bumps": 37, - "added": 1091, - "prefix": "?", - "permissions": 1544023111, - "tags": ["комбайн", "экономика", "модерация", "приветствия"], - "developers": ["585766846268047370"], - "links": {"discord": "5qXgJvr", "github": None, "site": "https://kerdoku.top"}, - "library": "discordpy", - "shortDescription": "Удобный и дружелюбный бот, который имеет крутой функционал!", - "longDescription": "wow", - "badge": None, - "stats": {"servers": 2558, "shards": 3, "users": 348986}, - "status": "APPROVED", - }, -} - -server_data_dict = { - "id": "722424773233213460", - "shortCode": "boticord", - "status": "ACCEPT_MEMBERS", - "links": [ - "https://boticord.top/server/722424773233213460", - "https://bcord.cc/s/722424773233213460", - "https://myservers.me/s/722424773233213460", - "https://boticord.top/server/boticord", - "https://bcord.cc/s/boticord", - "https://myservers.me/s/boticord", - ], - "bot": {"id": None, "approved": False}, - "information": { - "name": "BotiCord Community", - "avatar": "https://cdn.discordapp.com/icons/722424773233213460/060188f770836697846710b109272e4c.webp", - "members": [438, 0], - "bumps": 62, - "tags": [ - "аниме", - "игры", - "поддержка", - "комьюнити", - "сообщество", - "discord", - "дискорд сервера", - "дискорд боты", - ], - "links": { - "invite": "hkHjW8a", - "site": "https://boticord.top/", - "youtube": None, - "twitch": None, - "steam": None, - "vk": None, - }, - "shortDescription": "short text", - "longDescription": "long text", - "badge": "STAFF", - }, -} - -user_profile_dict = { - "id": "178404926869733376", - "status": '"Если вы не разделяете мою точку зрения, поздравляю — вам больше достанется." © Артемий Лебедев', - "badge": "STAFF", - "shortCode": "cipherka", - "site": "https://sqdsh.top/", - "vk": None, - "steam": "sadlycipherka", - "youtube": None, - "twitch": None, - "git": "https://git.sqdsh.top/me", -} - - -@pytest.fixture -def single_comment() -> types.SingleComment: - return types.SingleComment(**single_comment_dict) - - -@pytest.fixture -def bot_data() -> types.Bot: - return types.Bot(**bot_data_dict) - - -@pytest.fixture -def server_data() -> types.Server: - return types.Bot(**server_data_dict) - - -@pytest.fixture -def user_profile_data() -> types.UserProfile: - return types.UserProfile(**user_profile_dict) - - -def test_comment_dict_fields(single_comment: types.SingleComment) -> None: - for attr in single_comment: - assert single_comment.get(attr) == getattr(single_comment, attr) - - -def test_user_profile_dict_fields(user_profile_data: types.UserProfile) -> None: - for attr in user_profile_data: - assert user_profile_data.get(attr) == getattr(user_profile_data, attr) - - -def test_bot_dict_fields(bot_data: types.Bot) -> None: - for attr in bot_data: - if attr.lower() == "information": - assert bot_data["information"].get(attr) == getattr(bot_data, attr) - else: - assert bot_data[attr] == getattr(bot_data, attr) - - -def test_server_dict_fields(server_data: types.Server) -> None: - for attr in server_data: - if attr.lower() == "information": - assert server_data["information"].get(attr) == getattr(bot_data, attr) - else: - assert server_data[attr] == getattr(server_data, attr)