diff --git a/.gitignore b/.gitignore index 3280530..bdf246b 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ boticordpy.egg-info _testing.py /.pytest_cache /docs/build/ +.vscode \ No newline at end of file diff --git a/ACKNOWLEDGEMENTS.txt b/ACKNOWLEDGEMENTS.txt new file mode 100644 index 0000000..b34de64 --- /dev/null +++ b/ACKNOWLEDGEMENTS.txt @@ -0,0 +1,8 @@ +These are the open source libraries we use: +- aiohttp (https://github.com/aio-libs/aiohttp) +- typing_extensions (https://github.com/python/typing_extensions) +- sphinx (https://github.com/sphinx-doc/sphinx) +- furo (https://github.com/pradyunsg/furo) + +It is also important to note that some developments of Melisa (https://github.com/MelisaDev/melisa) were used in the development of the project. +I would also like to express my gratitude to the former admin staff of the BotiCord service (until 06/02/2023) and the entire project community. \ No newline at end of file diff --git a/LICENSE.txt b/LICENSE.txt index 7ab55d1..0463ab7 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -1,4 +1,4 @@ -Copyright 2021 Victor Kotlin +Copyright 2021 - 2023 Viktor K (Marakarka) Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: diff --git a/README.md b/README.md index 0cf1b2f..fc3edaa 100644 --- a/README.md +++ b/README.md @@ -22,12 +22,12 @@ * Object-oriented * Full BotiCord API Coverage * Modern Pythonic API using `async`/`await` syntax -* BotiCord Webhooks +* BotiCord Websocket * It is not necessary to use any particular library used to interact with the Discord API.

Installation

-Python 3.6 or newer is required. +Python 3.8 or newer is required. Enter one of these commands to install the library: @@ -49,28 +49,33 @@ You can find other examples in an examples folder. ```py from discord.ext import commands - from boticordpy import BoticordClient bot = commands.Bot(command_prefix="!") +# Function that will return the current bot's stats. async def get_stats(): return {"servers": len(bot.guilds), "shards": 0, "users": len(bot.users)} +# Function that will be called if stats are posted successfully. async def on_success_posting(): - print("stats posting successfully") + print("wow stats posting works") -boticord_client = BoticordClient("your_api_token") + +boticord_client = BoticordClient( + "your_boticord_api_token", version=3 +) # <--- BotiCord API token autopost = ( boticord_client.autopost() .init_stats(get_stats) .on_success(on_success_posting) - .start() + .start("id_of_your_bot") # <--- ID of your bot ) -bot.run("bot token") +bot.run("bot token") # <--- Discord bot's token + ```

Links

diff --git a/boticordpy/__init__.py b/boticordpy/__init__.py index 5fe812c..596f461 100644 --- a/boticordpy/__init__.py +++ b/boticordpy/__init__.py @@ -2,17 +2,16 @@ Boticord API Wrapper ~~~~~~~~~~~~~~~~~~~ A basic wrapper for the BotiCord API. -:copyright: (c) 2022 Marakarka +:copyright: (c) 2023 Marakarka :license: MIT, see LICENSE for more details. """ __title__ = "boticordpy" __author__ = "Marakarka" __license__ = "MIT" -__copyright__ = "Copyright 2022 Marakarka" -__version__ = "2.2.2" +__copyright__ = "Copyright 2021 - 2023 Marakarka" +__version__ = "3.0.0" from .client import BoticordClient -from .webhook import Webhook - from .types import * +from .websocket import BotiCordWebsocket diff --git a/boticordpy/autopost.py b/boticordpy/autopost.py index 6c09de9..321e050 100644 --- a/boticordpy/autopost.py +++ b/boticordpy/autopost.py @@ -1,8 +1,11 @@ import asyncio import typing +import logging from . import exceptions as bexc +_logger = logging.getLogger("boticord.autopost") + class AutoPost: """ @@ -28,12 +31,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: """ @@ -65,6 +72,8 @@ class AutoPost: self._success = callback return func + _logger.info("Registering success callback") + return inner def on_error(self, callback: typing.Any = None): @@ -91,6 +100,8 @@ class AutoPost: self._error = callback return func + _logger.info("Registering error callback") + return inner def init_stats(self, callback: typing.Any = None): @@ -116,6 +127,8 @@ class AutoPost: self._stats = callback return func + _logger.info("Registered stats initialization function") + return inner @property @@ -145,7 +158,8 @@ 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) + _logger.info("Tried to post bot stats") except Exception as err: on_error = getattr(self, "_error", None) if on_error: @@ -160,14 +174,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") @@ -178,6 +199,9 @@ class AutoPost: task = asyncio.ensure_future(self._internal_loop()) self._task = task + + _logger.info("Started autoposting") + return task def stop(self) -> None: @@ -188,3 +212,5 @@ class AutoPost: return None self._stopped = True + + _logger.info("Stopped autoposting") diff --git a/boticordpy/client.py b/boticordpy/client.py index 9cbdb76..aab9970 100644 --- a/boticordpy/client.py +++ b/boticordpy/client.py @@ -1,18 +1,17 @@ import typing +import logging from . import types as boticord_types from .http import HttpClient from .autopost import AutoPost +from .exceptions import MeilisearchException + +_logger = logging.getLogger("boticord") 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,210 +19,179 @@ 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") + __slots__ = ("http", "_autopost", "_token", "_meilisearch_api_key") http: HttpClient - def __init__(self, token: str = None, version: int = 2): + def __init__(self, token: str = None, version: int = 3): self._token = token + self._meilisearch_api_key = None 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. """ + _logger.info("Requesting information about bot") + 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:`int`]) + Id of the bot to post stats of. servers ( :obj:`int` ) Bot's servers count shards ( :obj:`int` ) Bot's shards count users ( :obj:`int` ) Bot's users count - Returns: - :obj:`dict`: - Boticord API Response status - """ - response = await self.http.post_bot_stats( - {"servers": servers, "shards": shards, "users": users} - ) - return response - async def get_server_info(self, server_id: int) -> boticord_types.Server: + Returns: + :obj:`~.types.ResourceBot`: + ResourceBot object. + """ + _logger.info("Posting bot stats") + + response = await self.http.post_bot_stats( + bot_id, {"servers": servers, "shards": shards, "users": users} + ) + return boticord_types.ResourceBot.from_dict(response) + + async def get_server_info( + self, server_id: typing.Union[str, int] + ) -> boticord_types.ResourceServer: """Gets information about specified server. Args: - server_id (:obj:`int`) + server_id (Union[:obj:`str`, :obj:`int`]) Id of the server Returns: - :obj:`~.types.Server`: - Server object. + :obj:`~.types.ResourceServer`: + ResourceServer object. """ + _logger.info("Requesting information about server") + response = await self.http.get_server_info(server_id) - return boticord_types.Server(**response) + return boticord_types.ResourceServer.from_dict(response) - async def get_server_comments(self, server_id: int) -> list: - """Gets list of comments of specified server. - - Args: - server_id (:obj:`int`) - Id of the server - - Returns: - :obj:`list` [ :obj:`~.types.SingleComment` ]: - List of comments. - """ - response = await self.http.get_server_comments(server_id) - return [boticord_types.SingleComment(**comment) for comment in response] - - async def post_server_stats(self, payload: dict) -> dict: - """Post Server's stats. You must be Boticord-Service bot. - Payload is raw, because if you use it - you know what you are doing. - You can find more information about payload `in BotiCord API Docs `_ - - Args: - payload (:obj:`dict`) - Custom data (Use Boticord API docs.) - Returns: - :obj:`dict`: - Boticord API Response. - """ - 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: typing.Union[str, int] + ) -> boticord_types.UserProfile: """Gets information about specified user. Args: - user_id (:obj:`int`) + user_id (Union[:obj:`str`, :obj:`int`]) Id of the user Returns: :obj:`~.types.UserProfile`: - User Profile object. + UserProfile object. """ + _logger.info("Requesting information about user") + response = await self.http.get_user_info(user_id) - return boticord_types.UserProfile(**response) + return boticord_types.UserProfile.from_dict(response) - async def get_user_comments(self, user_id: int) -> boticord_types.UserComments: - """Gets comments of specified user. + async def __search_for(self, index, data): + """Search for something on BotiCord""" + if self._meilisearch_api_key is None: + token_response = await self.http.get_search_key() + self._meilisearch_api_key = token_response["key"] - Args: - user_id (:obj:`int`) - Id of the user + try: + response = await self.http.search_for( + index, self._meilisearch_api_key, data + ) + except MeilisearchException: + token_response = await self.http.get_search_key() + self._meilisearch_api_key = token_response["key"] + + response = await self.http.search_for( + index, self._meilisearch_api_key, data + ) + + return response["hits"] + + async def search_for_bots( + self, **kwargs + ) -> typing.List[boticord_types.MeiliIndexedBot]: + """Search for bots on BotiCord. + + Note: + You can find every keyword argument `here `_. Returns: - :obj:`~.types.UserComments`: - User comments on Bots and Servers pages. + List[:obj:`~.types.MeiliIndexedBot`]: + List of found bots """ - response = await self.http.get_user_comments(user_id) - return boticord_types.UserComments(**response) + _logger.info("Searching for bots on BotiCord") - async def get_user_bots(self, user_id: int) -> list: - """Gets list of bots of specified user. + response = await self.__search_for("bots", kwargs) + return [boticord_types.MeiliIndexedBot.from_dict(bot) for bot in response] - Args: - user_id (:obj:`int`) - Id of the user + async def search_for_servers( + self, **kwargs + ) -> typing.List[boticord_types.MeiliIndexedServer]: + """Search for servers on BotiCord. + + Note: + You can find every keyword argument `here `_. Returns: - :obj:`list` [ :obj:`~.types.SimpleBot` ]: - List of simple information about users bots. + List[:obj:`~.types.MeiliIndexedServer`]: + List of found servers """ - response = await self.http.get_user_bots(user_id) - return [boticord_types.SimpleBot(**bot) for bot in response] + _logger.info("Searching for servers on BotiCord") - async def get_my_shorted_links(self, *, code: str = None): - """Gets shorted links of an authorized user + response = await self.__search_for("servers", kwargs) + return [ + boticord_types.MeiliIndexedServer.from_dict(server) for server in response + ] - Args: - code (:obj:`str`) - Code of shorted link. Could be None. + async def search_for_comments( + self, **kwargs + ) -> typing.List[boticord_types.MeiliIndexedComment]: + """Search for comments on BotiCord. + + Note: + You can find every keyword argument `here `_. Returns: - Union[:obj:`list` [ :obj:`~.types.ShortedLink` ], :obj:`~types.ShortedLink`]: - List of shorted links if none else shorted link + List[:obj:`~.types.MeiliIndexedComment`]: + List of found comments """ - response = await self.http.get_my_shorted_links(code) + _logger.info("Searching for comments on BotiCord") - return ( - [boticord_types.ShortedLink(**link) for link in response] - if code is None - else boticord_types.ShortedLink(**response[0]) - ) - - async def create_shorted_link( - self, *, code: str, link: str, domain: boticord_types.LinkDomain = 1 - ): - """Creates new shorted link - - Args: - code (:obj:`str`) - Code of link to short. - link (:obj:`str`) - Link to short. - domain (:obj:`~.types.LinkDomain`) - Domain to use in shorted link - - Returns: - :obj:`~types.ShortedLink`: - Shorted Link - """ - response = await self.http.create_shorted_link(code, link, domain=domain) - - return boticord_types.ShortedLink(**response) - - async def delete_shorted_link( - self, code: str, domain: boticord_types.LinkDomain = 1 - ): - """Deletes shorted link - - Args: - code (:obj:`str`) - Code of link to delete. - domain (:obj:`~.types.LinkDomain`) - Domain that is used in shorted link - - Returns: - :obj:`bool`: - Is link deleted successfully? - """ - response = await self.http.delete_shorted_link(code, domain) - - return response.get("ok", False) + response = await self.__search_for("comments", kwargs) + return [ + boticord_types.MeiliIndexedComment.from_dict(comment) + for comment in response + ] def autopost(self) -> AutoPost: """Returns a helper instance for auto-posting. diff --git a/boticordpy/exceptions.py b/boticordpy/exceptions.py index be501c9..e1401c5 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. @@ -18,7 +21,7 @@ class InternalException(BoticordException): class HTTPException(BoticordException): - """Exception that's thrown when an HTTP request operation fails. + """Exception that's thrown when request to BotiCord API operation fails. Attributes ---------- @@ -29,26 +32,148 @@ 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 MeilisearchException(BoticordException): + """Exception that's thrown when request to Meilisearch API operation fails. + + Attributes + ---------- + response: + The response of the failed HTTP request. + """ + + def __init__(self, response): + self.response = response + + fmt = f"{self.response['code']} ({self.response['message']})" + + super().__init__(fmt) -class Forbidden(HTTPException): - """Exception that's thrown when status code 403 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 NotFound(HTTPException): - """Exception that's thrown when status code 404 occurs.""" +class HTTPErrors(IntEnum): + """Errors which BotiCord may return""" + UNKNOWN_ERROR = 0 + """Unknown error""" -class ToManyRequests(HTTPException): - """Exception that's thrown when status code 429 occurs.""" + INTERNAL_SERVER_ERROR = 1 + """Server error (>500)""" + RATE_LIMITED = 2 + """Too many requests""" -class ServerError(HTTPException): - """Exception that's thrown when status code 500 or 503 occurs.""" + NOT_FOUND = 3 + """Not found""" + + FORBIDDEN = 4 + """Access denied""" + + 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..4a62222 100644 --- a/boticordpy/http.py +++ b/boticordpy/http.py @@ -1,9 +1,9 @@ import asyncio +import typing import aiohttp from . import exceptions -from .types import LinkDomain class HttpClient: @@ -20,96 +20,66 @@ 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.boticord.top/v{version}" loop = kwargs.get("loop") or asyncio.get_event_loop() self.session = kwargs.get("session") or aiohttp.ClientSession(loop=loop) - async def make_request(self, method: str, endpoint: str, **kwargs): + async def make_request( + self, method: str, endpoint: str, *, meilisearch_token: str = None, **kwargs + ) -> dict: """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 + if meilisearch_token is not None: + kwargs["headers"]["Authorization"] = f"Bearer {meilisearch_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 (200, 201).__contains__(response.status): + return data["result"] if not meilisearch_token else data + else: + if not meilisearch_token: + raise exceptions.HTTPException( + {"status": response.status, "error": data["errors"][0]["code"]} + ) + else: + raise exceptions.MeilisearchException(data) - 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): + def get_server_info(self, server_id: typing.Union[str, int]): """Get information about specified server""" - return self.make_request("GET", f"server/{server_id}") + return self.make_request("GET", f"servers/{server_id}") - def get_server_comments(self, server_id: int): - """Get list of specified server comments""" - return self.make_request("GET", f"server/{server_id}/comments") + def get_user_info(self, user_id: typing.Union[str, int]): + """Get information about specified user""" + return self.make_request("GET", f"users/{user_id}") - def post_server_stats(self, payload: dict): - """Post server's stats""" - return self.make_request("POST", "server", json=payload) + def get_search_key(self): + """Get API key for Meilisearch""" + return self.make_request("GET", f"search-key") - def get_user_info(self, user_id: int): - """Get information about the user""" - return self.make_request("GET", f"profile/{user_id}") - - def get_user_comments(self, user_id: int): - """Get specified user's comments""" - return self.make_request("GET", f"user/{user_id}/comments") - - def get_user_bots(self, user_id: int): - """Get bots of specified user""" - return self.make_request("GET", f"bots/{user_id}") - - def get_my_shorted_links(self, code: str = None): - """Get shorted links of an authorized user""" - body = {"code": code} if code is not None else {} - - return self.make_request("POST", "links/get", json=body) - - def create_shorted_link(self, code: str, link: str, *, domain: LinkDomain = 1): - """Create new shorted link""" + def search_for(self, index: str, api_key: str, data: dict): + """Search for something on BotiCord.""" return self.make_request( "POST", - "links/create", - json={"code": code, "link": link, "domain": int(domain)}, - ) - - def delete_shorted_link(self, code: str, domain: LinkDomain = 1): - """Delete shorted link""" - return self.make_request( - "POST", "links/delete", json={"code": code, "domain": int(domain)} + f"search/indexes/{index}/search", + meilisearch_token=api_key, + json=data, ) diff --git a/boticordpy/types.py b/boticordpy/types.py index 8998a36..11a374c 100644 --- a/boticordpy/types.py +++ b/boticordpy/types.py @@ -1,375 +1,1092 @@ -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 + """Discord4j""" + + DISCORDCR = 2 + """Discordcr""" + + DISCORDGO = 3 + """DiscordGO""" + + DISCORDDOO = 4 + """Discordoo""" + + DSHARPPLUS = 5 + """DSharpPlus""" + + DISCORDJS = 6 + """Discord.js""" + + DISCORDNET = 7 + """Discord.Net""" + + DISCORDPY = 8 + """discord.py""" + + ERIS = 9 + """eris""" + + JAVACORD = 10 + """JavaCord""" + + JDA = 11 + """JDA""" + + OTHER = 12 + """Other""" + + NONE = 0 + """Bot's library doesn't specified""" + + +class ResourceStatus(IntEnum): + """Status of the project on monitoring""" + + HIDDEN = 0 + """is hidden""" + + PUBLIC = 1 + """is public""" + + BANNED = 2 + """is banned""" + + PENDING = 3 + """is pending""" + + +class ServerTag(IntEnum): + """Tags of the server""" + + SPEAKING = 130 + """Speaking""" + + FUN = 131 + """Fun""" + + GAMES = 132 + """Games""" + + CINEMA = 133 + """Cinema""" + + ANIME = 134 + """Anime""" + + ART = 135 + """Art""" + + CODING = 136 + """Coding""" + + MUSIC = 137 + """Music""" + + ADULT = 138 + """18+""" + + ROLEPLAY = 139 + """Role-Play""" + + HUMOUR = 140 + """Humour""" + + GENSHIN = 160 + """Genshin""" + + MINECRAFT = 161 + """Minecraft""" + + GTA = 162 + """GTA""" + + CS = 163 + """CS""" + + DOTA = 164 + """Dota""" + + AMONG_US = 165 + """Among Us""" + + FORTNITE = 166 + """Fortnite""" + + BRAWL_STARS = 167 + """Brawl Stars""" + + +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: UserLinks = super().__new__(cls) + data = data or {} + + self.vk = data.get("vk") + self.telegram = data.get("telegram") + self.donate = data.get("donate") + self.git = data.get("git") + self.custom = data.get("custom") + + return self + + +@dataclass(repr=False) +class UserBadge(APIObjectBase): + """Information about user's profile badge""" + + id: int + """Badge's ID""" + + name: str + """Badge's name""" + + asset_url: str + """Badge's icon URL""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a UserBadge from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a UserBadge. + """ + self: UserBadge = super().__new__(cls) + + self.id = data["id"] + self.name = data["name"] + self.asset_url = data["assetURL"] + + 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 + ) + + return self + + +@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 + """Id of the user""" + + socials: UserLinks + """Links of the user""" + + description: Optional[str] + """Description of the user""" + + short_description: Optional[str] + """Short description of the user""" + + 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: cls = super().__new__(cls) + + self.username = data["username"] + self.discriminator = data.get("discriminator") + self.avatar = data.get("avatar") + self.id = data["id"] + self.socials = UserLinks.from_dict(data.get("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 ResourceServer(APIObjectBase): + """Information about server from BotiCord. + + .. warning:: + + The result of the reverse conversion (`.to_dict()`) may not match the actual data. + """ + + id: str + """Server's ID""" + + name: str + """Server's name""" + + short_description: str + """Server's short description""" + + description: str + """Server's description""" + + avatar: Optional[str] + """Server's avatar""" + + short_link: Optional[str] + """Server's short link""" + + invite_link: str + """Server's invite link""" + + premium_active: bool + """Server's premium state""" + + premium_auto_fetch: Optional[bool] + """Server's premium auto fetch state""" + + premium_banner_url: Optional[str] + """Server's premium banner URL""" + + premium_splash_url: Optional[str] + """Server's premium splash URL""" + + standart_banner_id: int + """Server's standart banner ID""" + + owner: str + """Server's owner ID""" + + status: ResourceStatus + """Server's status""" + + ratings: List[ResourceRating] + """Server's ratings""" + + created_date: datetime + """Server's creation time""" + + members: Optional[int] + """Server's members count""" + + website: Optional[str] + """Server's website""" + + tags: List[ServerTag] + """Server's tags""" + + moderators: List[PartialUser] + """Server's moderators""" + + up_count: int + """Server's up count""" + + ups: Optional[ResourceUp] + """Server's ups""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a ResourceServer from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a ResourceServer.""" + + self = 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.website = data.get("website") + self.up_count = data.get("upCount") + self.standart_banner_id = data.get("standartBannerID") + + 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") + + self.status = ResourceStatus(data.get("status")) + self.ratings = [ + ResourceRating.from_dict(rating) for rating in data.get("ratings", []) + ] + self.created_date = datetime.strptime( + data.get("createdDate"), "%Y-%m-%dT%H:%M:%S.%f%z" + ) + self.tags = [ServerTag(tag) for tag in data.get("tags", [])] + self.ups = [ResourceUp.from_dict(up) for up in data.get("ups", [])] + self.moderators = [ + PartialUser.from_dict(mod) for mod in data.get("moderators", []) + ] + + self.members = data.get("memberCount") + + 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 bot""" + + short_description: str + """Short description of the bot""" + + description: str + """Description of the bot""" + + avatar: Optional[str] + """Avatar of the bot""" + + short_link: Optional[str] + """Short link to the bot's page""" + + standart_banner_id: int + """Server's standart banner ID""" + + invite_link: str + """Invite link""" + + premium_active: bool + """Is premium status active? (True/False)""" + + premium_splash_url: Optional[str] + """Link to the splash""" + + premium_auto_fetch: Optional[bool] + """Is auto-fetch enabled? (True/False)""" + + premium_banner_url: Optional[str] + """Premium banner URL""" + + owner: str + """Owner of the bot""" + + status: ResourceStatus + """Status of the bot""" + + ratings: List[ResourceRating] + """Bot's ratings""" prefix: str - """Bot's commands prefix""" + """Prefix of the bot""" - permissions: int - """Bot's permissions""" + discriminator: str + """Bot's discriminator""" - tags: list - """Bot's search-tags""" + created_date: datetime + """Date when the bot was published""" - developers: list - """List of bot's developers Ids""" + support_server_invite_link: Optional[str] + """Link to the support server""" - links: typing.Optional[dict] - """Bot's social medias""" + library: Optional[BotLibrary] + """The library that the bot is based on""" - library: typing.Optional[str] - """Bot's library""" + guilds: Optional[int] + """Number of guilds""" - short_description: typing.Optional[str] - """Bot's short description""" + shards: Optional[int] + """Number of shards""" - long_description: typing.Optional[str] - """Bot's long description""" + members: Optional[int] + """Number of members""" - badge: typing.Optional[str] - """Bot's badge""" + website: Optional[str] + """Link to bot's website""" - stats: dict - """Bot's stats""" + tags: List[BotTag] + """List of bot tags""" - status: str - """Bot's approval status""" + up_count: int + """Number of ups""" - def __init__(self, **kwargs): - super().__init__(**parse_with_information_dict(kwargs)) + ups: List[ResourceUp] + """List of bot's ups""" + + developers: List[PartialUser] + """List of bot's developers""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a ResourceBot from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a ResourceBot. + """ + + 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") + self.standart_banner_id = data.get("standartBannerID") + + 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") + + 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", []) + ] + + self.guilds = data.get("guilds") + self.shards = data.get("shards") + self.members = data.get("members") + + return self -class Server(ApiData): - """This model represents a server, returned from the Boticord API""" +@dataclass(repr=False) +class UserProfile(PartialUser): + """Information about user's profile from BotiCord.'""" + + badges: List[UserBadge] + """User's badges list.""" + + bots: List[ResourceBot] + """User's bots list""" + + servers: List[ResourceServer] + """User's servers list""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a UserProfile from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a UserProfile.""" + + self = super().from_dict(data) + + self.badges = [UserBadge.from_dict(badge) for badge in data.get("badges", [])] + self.bots = [ResourceBot.from_dict(bot) for bot in data.get("bots", [])] + self.servers = [ + ResourceServer.from_dict(server) for server in data.get("servers", []) + ] + + return self + + +@dataclass(repr=False) +class MeiliIndexedBot(APIObjectBase): + """Bot found on BotiCord + + .. warning:: + + The result of the reverse conversion (`.to_dict()`) may not match the actual data. + """ id: str - """Server's Id""" + """ID of the bot""" - short_code: typing.Optional[str] - """Server's page short code""" + name: str + """Name of the bot""" - status: str - """Server's approval status""" + short_description: str + """Short description of the bot""" - page_links: list - """List of server's page urls""" + description: str + """Description of the bot""" - bot: dict - """Bot where this server is used for support users""" + avatar: Optional[str] + """Avatar of the bot""" + + invite: str + """Invite link""" + + premium_active: bool + """Is premium status active? (True/False)""" + + premium_banner: Optional[str] + """Premium banner URL""" + + banner: int + """Standart banner""" + + rating: int + """Bot's rating""" + + discriminator: str + """Bot's discriminator""" + + library: Optional[BotLibrary] + """The library that the bot is based on""" + + guilds: Optional[int] + """Number of guilds""" + + shards: Optional[int] + """Number of shards""" + + members: Optional[int] + """Number of members""" + + tags: List[BotTag] + """List of bot tags""" + + ups: int + """List of bot's ups""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a MeiliIndexedBot from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a MeiliIndexedBot. + """ + + self: MeiliIndexedBot = 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.invite = data.get("invite") + self.discriminator = data.get("discriminator") + self.ups = data.get("ups") + self.rating = data.get("rating") + self.banner = data.get("banner") + + self.premium_active = data.get("premiumActive") + self.premium_banner = data.get("premiumBanner") + + 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.guilds = data.get("guilds") + self.shards = data.get("shards") + self.members = data.get("members") + + return self + + +@dataclass(repr=False) +class MeiliIndexedServer(APIObjectBase): + """Server found on BotiCord + + .. warning:: + + The result of the reverse conversion (`.to_dict()`) may not match the actual data. + """ + + id: str + """ID of the server""" name: str """Name of the server""" - avatar: str - """Server's avatar""" + short_description: str + """Short description of the server""" - members: list - """Members counts - `[all, online]`""" + description: str + """Description of the server""" - owner: typing.Optional[str] - """Server's owner Id""" + avatar: Optional[str] + """Avatar of the server""" - bumps: int - """Bumps count""" + invite: str + """Invite link""" - tags: list - """Server's search-tags""" + premium_active: bool + """Is premium status active? (True/False)""" - links: dict - """Server's social medias""" + premium_banner: Optional[str] + """Premium banner URL""" - short_description: typing.Optional[str] - """Server's short description""" + banner: int + """Standart banner""" - long_description: typing.Optional[str] - """Server's long description""" + discord_banner: Optional[str] + """Discord banner URL""" - badge: typing.Optional[str] - """Server's badge""" + rating: int + """Server's rating""" - def __init__(self, **kwargs): - super().__init__(**parse_with_information_dict(kwargs)) + members: Optional[int] + """Number of members""" + + tags: List[ServerTag] + """List of server tags""" + + ups: int + """List of server's ups""" + + @classmethod + def from_dict(cls, data: dict): + """Generate a MeiliIndexedServer from the given data. + + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a MeiliIndexedServer. + """ + + self: MeiliIndexedServer = 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.invite = data.get("invite") + self.ups = data.get("ups") + self.rating = data.get("rating") + self.banner = data.get("banner") + + self.premium_active = data.get("premiumActive") + self.premium_banner = data.get("premiumBanner") + self.discord_banner = data.get("discordBanner") + + self.tags = [ServerTag(tag) for tag in data.get("tags", [])] + + self.members = data.get("members") + + return self -class UserProfile(ApiData): - """This model represents profile of user, returned from the Boticord API""" +@dataclass(repr=False) +class MeiliIndexedComment(APIObjectBase): + """Comment found on BotiCord""" id: str - """Id of User""" + """ID of the comment""" - status: str - """Status of user""" + author: str + """Id of the author of the comment""" - badge: typing.Optional[str] - """User's badge""" + rating: int + """Comment's rating""" - short_code: typing.Optional[str] - """User's profile page short code""" + content: str + """Content of the comment""" - site: typing.Optional[str] - """User's website""" + resource: str + """Id of the resource""" - vk: typing.Optional[str] - """User's VK Profile""" + created: datetime + """When the comment was created""" - steam: typing.Optional[str] - """User's steam account""" + mod_reply: Optional[str] + """Reply to the comment""" - youtube: typing.Optional[str] - """User's youtube channel""" + @classmethod + def from_dict(cls, data: dict): + """Generate a MeiliIndexedComment from the given data. - twitch: typing.Optional[str] - """User's twitch channel""" + Parameters + ---------- + data: :class:`dict` + The dictionary to convert into a MeiliIndexedComment. + """ - git: typing.Optional[str] - """User's github/gitlab (or other git-service) profile""" + self: MeiliIndexedComment = super().__new__(cls) - def __init__(self, **kwargs): - super().__init__(**parse_response_dict(kwargs)) + self.id = data.get("id") + self.rating = data.get("rating") + self.author = data.get("author") + self.content = data.get("content") + self.resource = data.get("resource") + self.mod_reply = data.get("modReply") + self.created = datetime.utcfromtimestamp(data.get("created") / 1000) - -class UserComments(ApiData): - """This model represents all the user's comments on every page""" - - bots: list - """Data from `get_bot_comments` method""" - - servers: list - """Data from `get_server_comments` method""" - - def __init__(self, **kwargs): - super().__init__(**parse_user_comments_dict(kwargs)) - - -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.""" - - id: str - """Bot's Id""" - - short_code: typing.Optional[str] - """Bot's page short code""" - - 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)) + return self 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/boticordpy/websocket.py b/boticordpy/websocket.py new file mode 100644 index 0000000..3b2bb70 --- /dev/null +++ b/boticordpy/websocket.py @@ -0,0 +1,151 @@ +# Copyright Marakarka (Viktor K) 2021 - Present +# Full MIT License can be found in `LICENSE.txt` at the project root. + +import logging +import json +import asyncio +import typing + +import aiohttp + +_logger = logging.getLogger("boticord.websocket") + + +class BotiCordWebsocket: + """Represents a client that can be used to interact with the BotiCord by websocket connection.""" + + def __init__(self, token: str): + self.__session = None + self.loop = asyncio.get_event_loop() + self.ws = None + self._listeners = {} + self.not_closed = True + + self._token = token + + def listener(self): + """Decorator to set the listener. + + .. warning:: + + Callback functions must be a **coroutine**. If they aren't, then you might get unexpected + errors. In order to turn a function into a coroutine they must be ``async def`` + functions. + + For example: + + .. code-block:: python + + @websocket.listener() + async def comment_removed(data): + pass + """ + + def inner(func): + if not asyncio.iscoroutinefunction(func): + raise TypeError(f"<{func.__qualname__}> must be a coroutine function") + self._listeners[func.__qualname__] = func + _logger.debug(f"Listener {func.__qualname__} added successfully!") + return func + + return inner + + def register_listener(self, notification_type: str, callback: typing.Any): + """Method to set the listener. + + Args: + notify_type (:obj:`str`) + Type of notification (Check reference page) + callback (:obj:`function`) + Coroutine Callback Function + + .. warning:: + + Callback functions must be a **coroutine**. If they aren't, then you might get unexpected + errors. In order to turn a function into a coroutine they must be ``async def`` + functions. + """ + if not asyncio.iscoroutinefunction(callback): + raise TypeError(f"<{callback.__qualname__}> must be a coroutine function") + + self._listeners[notification_type] = callback + _logger.debug(f"Listener {callback.__qualname__} added successfully!") + return self + + async def connect(self) -> None: + """Connect to BotiCord.""" + try: + self.__session = aiohttp.ClientSession() + self.ws = await self.__session.ws_connect( + "wss://gateway.boticord.top/websocket/", + timeout=30.0, + ) + + _logger.info("Connected to BotiCord.") + + self.not_closed = True + + self.loop.create_task(self._receive()) + await self._send_identify() + except Exception as exc: + _logger.error("Connecting failed!") + + raise exc + + async def _send_identify(self) -> None: + await self.ws.send_json({"event": "auth", "data": {"token": self._token}}) + + async def _receive(self) -> None: + while self.not_closed: + async for msg in self.ws: + if msg.type == aiohttp.WSMsgType.TEXT: + await self._handle_data(msg.data) + else: + raise RuntimeError + + close_code = self.ws.close_code + + if close_code is not None: + await self._handle_close(close_code) + + async def _handle_data(self, data): + data = json.loads(data) + + if data["event"] == "hello": + _logger.info("Authorized successfully.") + self.loop.create_task(self._send_ping()) + elif data["event"] == "notify": + listener = self._listeners.get(data["data"]["type"]) + if listener: + self.loop.create_task(listener(data["data"])) + elif data["event"] == "pong": + _logger.info("Received pong-response.") + self.loop.create_task(self._send_ping()) + else: + _logger.error("An error has occurred.") + + async def _handle_close(self, code: int) -> None: + self.not_closed = False + await self.__session.close() + + if code == 4000: + _logger.info("Closed connection successfully.") + return + elif code == 1006: + _logger.error("Token is invalid.") + return + + _logger.info("Disconnected from BotiCord. Reconnecting...") + + await self.connect() + + async def _send_ping(self) -> None: + if not self.ws.closed: + await asyncio.sleep(45) + await self.ws.send_json({"event": "ping"}) + + async def close(self) -> None: + """Close websocket connection with BotiCord""" + if self.ws: + self.not_closed = False + await self.ws.close(code=4000) diff --git a/docs-requirements.txt b/docs-requirements.txt index 2ae7816..d1c21fe 100644 --- a/docs-requirements.txt +++ b/docs-requirements.txt @@ -1,2 +1,4 @@ -sphinxawesome_theme +furo +sphinxcontrib_trio +sphinx_design sphinx \ No newline at end of file diff --git a/docs/source/api.rst b/docs/source/api.rst index 706644a..6540202 100644 --- a/docs/source/api.rst +++ b/docs/source/api.rst @@ -14,5 +14,4 @@ API Reference for the boticordpy Module api/client api/autopost api/exceptions - api/types - api/webhook \ No newline at end of file + api/types \ No newline at end of file diff --git a/docs/source/api/autopost.rst b/docs/source/api/autopost.rst index 82e9f81..30ae03b 100644 --- a/docs/source/api/autopost.rst +++ b/docs/source/api/autopost.rst @@ -1,6 +1,6 @@ -#################### +########################### AutoPost API Reference -#################### +########################### .. automodule:: boticordpy.autopost :members: diff --git a/docs/source/api/client.rst b/docs/source/api/client.rst index cb589f2..6ea5f18 100644 --- a/docs/source/api/client.rst +++ b/docs/source/api/client.rst @@ -1,7 +1,11 @@ +.. currentmodule:: boticordpy + #################### Client API Reference #################### -.. automodule:: boticordpy.client - :members: +BoticordClient +----------------- + +.. autoclass:: BoticordClient :inherited-members: diff --git a/docs/source/api/exceptions.rst b/docs/source/api/exceptions.rst index 5fbedef..ac2cea0 100644 --- a/docs/source/api/exceptions.rst +++ b/docs/source/api/exceptions.rst @@ -1,7 +1,23 @@ -#################### -Exceptions API Reference -#################### +.. currentmodule:: boticordpy.exceptions -.. automodule:: boticordpy.exceptions +########################## +Exceptions API Reference +########################## + +.. autoclass:: BoticordException + :members: + +.. autoclass:: InternalException + :members: + +.. autoclass:: HTTPException + :members: + +.. autoclass:: MeilisearchException + :members: + +.. autoclass:: StatusCodes + :members: + +.. autoclass:: HTTPErrors :members: - :inherited-members: diff --git a/docs/source/api/types.rst b/docs/source/api/types.rst index dc0a263..959cb54 100644 --- a/docs/source/api/types.rst +++ b/docs/source/api/types.rst @@ -1,8 +1,71 @@ +.. currentmodule:: boticordpy.types + #################### Models API Reference #################### -We recommend you to read the `boticordpy/types.py `_ file, because it is much easier to read than here. - -.. automodule:: boticordpy.types +.. autoclass:: ResourceRating + :members: + +.. autoclass:: ResourceUp + :members: + +Enums +------- + +.. autoclass:: BotLibrary + :members: + +.. autoclass:: BotTag + :members: + +.. autoclass:: ServerTag + :members: + +.. autoclass:: ResourceStatus + :members: + + +Bots +------ + +.. autoclass:: ResourceBot + :members: + + +Servers +--------- + +.. autoclass:: ResourceServer + :members: + + +Users +------ + +.. autoclass:: UserLinks + :members: + +.. autoclass:: UserBadge + :members: + +.. autoclass:: PartialUser + :members: + +.. autoclass:: UserProfile + :members: + :exclude-members: to_dict + :inherited-members: + + +MeiliSearch +------------ + +.. autoclass:: MeiliIndexedBot + :members: + +.. autoclass:: MeiliIndexedServer + :members: + +.. autoclass:: MeiliIndexedComment :members: diff --git a/docs/source/api/webhook.rst b/docs/source/api/webhook.rst deleted file mode 100644 index 59198b0..0000000 --- a/docs/source/api/webhook.rst +++ /dev/null @@ -1,7 +0,0 @@ -#################### -Webhook API Reference -#################### - -.. automodule:: boticordpy.webhook - :members: - :inherited-members: diff --git a/docs/source/conf.py b/docs/source/conf.py index a79baa0..88ac24a 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -23,11 +23,11 @@ import os sys.path.insert(0, os.path.abspath("../..")) project = "BoticordPY" -copyright = "2022, Victor Kotlin (Marakarka)" -author = "Victor Kotlin (Marakarka)" +copyright = "2022 - 2023, Viktor K (Marakarka)" +author = "Viktor K (Marakarka)" # The full version, including alpha/beta/rc tags -release = "2.2.2" +release = "3.0.0a" # -- General configuration --------------------------------------------------- @@ -36,41 +36,47 @@ release = "2.2.2" # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # ones. extensions = [ + "sphinx_design", "sphinx.ext.napoleon", "sphinx.ext.autodoc", "sphinx.ext.viewcode", "sphinx.ext.autosectionlabel", "sphinx.ext.extlinks", + "sphinxcontrib_trio", ] +autodoc_default_options = { + "members": True, + "show-inheritance": True, + "member-order": "bysource", +} + # Add any paths that contain templates here, relative to this directory. templates_path = ["_templates"] -# List of patterns, relative to source directory, that match files and -# directories to ignore when looking for source files. -# This pattern also affects html_static_path and html_extra_path. +add_module_names = False + exclude_patterns = [] intersphinx_mapping = { "py": ("https://docs.python.org/3", None), - "discord": ("https://discordpy.readthedocs.io/en/latest/", None), "aiohttp": ("https://docs.aiohttp.org/en/stable/", None), } # -- Options for HTML output ------------------------------------------------- -# The theme to use for HTML and HTML Help pages. See the documentation for -# a list of builtin themes. -# -html_theme = "sphinxawesome_theme" - -html_theme_options = {} - -# Add any paths that contain custom static files (such as style sheets) here, -# relative to this directory. They are copied after the builtin static files, -# so a file named "custom.css" will overwrite the builtin "custom.css". +html_theme = "furo" +html_theme_options = { + "sidebar_hide_name": True, +} +pygments_style = "monokai" +default_dark_mode = True html_static_path = ["_static"] +html_css_files = ["custom.css"] - -def setup(app): - app.add_css_file("custom.css") +rst_prolog = """ +.. |coro| replace:: This function is a |coroutine_link|_. +.. |maybecoro| replace:: This function *could be a* |coroutine_link|_. +.. |coroutine_link| replace:: *coroutine* +.. _coroutine_link: https://docs.python.org/3/library/asyncio-task.html#coroutine +""" diff --git a/docs/source/index.rst b/docs/source/index.rst index 63629a7..1d2dd05 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -11,7 +11,7 @@ This is a documentation for wrapper for BotiCord API. quickstart api - other + websocket Links ===== diff --git a/docs/source/other.rst b/docs/source/other.rst deleted file mode 100644 index eb26853..0000000 --- a/docs/source/other.rst +++ /dev/null @@ -1,38 +0,0 @@ -.. currentmodule:: boticordpy - -.. other: - -Other Information -================= - -########## -Listeners -########## - -When you work with BotiCord Webhooks you may receive a lot of events. -To make it easier to handle them there is a list of the events you can receive: - -.. csv-table:: - :header: "BotiCord Events", "Meaning" - :widths: 20, 20 - - "test_webhook_message", "Test message." - "new_bot_comment", "On new bot comment" - "edit_bot_comment", "On bot comment edit" - "delete_bot_comment", "On bot comment delete" - "new_bot_bump", "On new bot bump" - "new_server_comment", "On new server comment" - "edit_server_comment", "On server comment edit" - "delete_server_comment", "On server comment delete" - "new_server_bump", "On new server bump" - - -################## -Callback functions -################## - -.. warning:: - - Callback functions must be a **coroutine**. If they aren't, then you might get unexpected - errors. In order to turn a function into a coroutine they must be ``async def`` - functions. diff --git a/docs/source/websocket.rst b/docs/source/websocket.rst new file mode 100644 index 0000000..5fcc5ec --- /dev/null +++ b/docs/source/websocket.rst @@ -0,0 +1,22 @@ +.. currentmodule:: boticordpy.websocket + +########### +WebSocket +########### + +BotiCord Websocket +------------------- + +.. autoclass:: BotiCordWebsocket + :exclude-members: listener + :inherited-members: + + .. automethod:: BotiCordWebsocket.listener() + :decorator: + + +Notification types +------------------- +.. function:: comment_removed(data) + + Called when comment is deleted. \ No newline at end of file diff --git a/examples/autopost.py b/examples/autopost.py index ac370f2..6d41e6f 100644 --- a/examples/autopost.py +++ b/examples/autopost.py @@ -15,15 +15,17 @@ async def get_stats(): # Function that will be called if stats are posted successfully. async def on_success_posting(): - print("stats posting successfully") + print("wow stats posting works") -boticord_client = BoticordClient("Bot your_api_token", version=2) +boticord_client = BoticordClient( + "your_boticord_api_token", version=3 +) # <--- BotiCord API token autopost = ( boticord_client.autopost() .init_stats(get_stats) .on_success(on_success_posting) - .start() + .start("id_of_your_bot") # <--- ID of your bot ) -bot.run("bot token") +bot.run("bot token") # <--- Discord bot's token diff --git a/examples/stats_melisa.py b/examples/stats_melisa.py new file mode 100644 index 0000000..a6bb181 --- /dev/null +++ b/examples/stats_melisa.py @@ -0,0 +1,16 @@ +import melisa +from boticordpy import BoticordClient + +bot = melisa.Bot("your_discord_bot_token") + +boticord = BoticordClient("your_boticord_api_token") + + +@bot.listen +async def on_message_create(message): + if message.content.startswith("!guilds"): + data = await boticord.get_bot_info(bot.user.id) + await bot.rest.create_message(message.channel.id, data.guilds) + + +bot.run_autosharded() diff --git a/examples/webhooks.py b/examples/webhooks.py deleted file mode 100644 index 5b937a6..0000000 --- a/examples/webhooks.py +++ /dev/null @@ -1,20 +0,0 @@ -# You can use any library to interact with the Discord API. -# This example uses discord.py. -# You can install it with `pip install discord.py`. - -from discord.ext import commands -from boticordpy import webhook - -bot = commands.Bot(command_prefix="!") - - -async def edit_bot_comment(data): - print(data.comment.new) - - -boticord_webhook = webhook.Webhook("x-hook-key", "bot").register_listener( - "edit_bot_comment", edit_bot_comment -) -boticord_webhook.start(5000) - -bot.run("bot_token") diff --git a/examples/websocket.py b/examples/websocket.py new file mode 100644 index 0000000..7505050 --- /dev/null +++ b/examples/websocket.py @@ -0,0 +1,23 @@ +# You can use any library to interact with the Discord API. +# This example uses discord.py. +# You can install it with `pip install discord.py`. + +from discord.ext import commands +from boticordpy import BotiCordWebsocket + +bot = commands.Bot(command_prefix="!") + +websocket = BotiCordWebsocket("your_boticord_api_token") # <--- BotiCord API token + + +@websocket.listener() +async def comment_removed(data): + print(data["payload"]) + + +@bot.event +async def on_ready(): + await websocket.connect() + + +bot.run("bot token") # <--- Discord bot's token 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/setuo.cfg b/setup.cfg similarity index 100% rename from setuo.cfg rename to setup.cfg diff --git a/setup.py b/setup.py index b0b7a5b..63e80aa 100644 --- a/setup.py +++ b/setup.py @@ -44,10 +44,11 @@ setup( }, packages=find_packages(), version=version, - python_requires=">= 3.6", + python_requires=">= 3.8", description="A Python wrapper for BotiCord API", long_description=README, long_description_content_type="text/markdown", + include_package_data=True, url="https://github.com/boticord/boticordpy", author="Marakarka", author_email="support@kerdoku.top", @@ -55,7 +56,9 @@ setup( classifiers=[ "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", ], - install_requires=["aiohttp"], + install_requires=["aiohttp", "typing_extensions"], ) diff --git a/tests/test_convertation.py b/tests/test_convertation.py new file mode 100644 index 0000000..2a5e55a --- /dev/null +++ b/tests/test_convertation.py @@ -0,0 +1,74 @@ +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": {}, +} +resource_server_dict = { + "id": "722424773233213460", + "name": "BotiCord.top", + "tags": [134, 132], + "status": 1, + "createdDate": "2023-05-23T15:16:45.387Z", + "premium": {}, +} +user_profile_dict = { + "id": "585766846268047370", + "username": "Marakarka", + "bots": [resource_bot_dict], + "shortDescription": None, +} + + +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" + + +def test_resource_server_convertation(): + model_from_dict = types.ResourceServer.from_dict(resource_server_dict) + + assert int(model_from_dict.created_date.timestamp()) == 1684855005 + assert model_from_dict.name == "BotiCord.top" + assert model_from_dict.tags[1].name == "GAMES" + + +def test_user_profile_convertation(): + model_from_dict = types.UserProfile.from_dict(user_profile_dict) + + assert model_from_dict.id == "585766846268047370" + assert model_from_dict.username == "Marakarka" + assert model_from_dict.short_description == None + assert model_from_dict.bots[0].id == "947141336451153931" 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)