rewrited bots

This commit is contained in:
grey-cat-1908 2023-06-04 10:42:33 +03:00
parent cf683fa14f
commit 7b9e341adc
10 changed files with 732 additions and 660 deletions

View file

@ -13,6 +13,5 @@ __copyright__ = "Copyright 2022 Marakarka"
__version__ = "2.2.2" __version__ = "2.2.2"
from .client import BoticordClient from .client import BoticordClient
from .webhook import Webhook
from .types import * from .types import *

View file

@ -28,12 +28,16 @@ class AutoPost:
_stats: typing.Any _stats: typing.Any
_task: typing.Optional["asyncio.Task[None]"] _task: typing.Optional["asyncio.Task[None]"]
bot_id: str
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
self._stopped: bool = False self._stopped: bool = False
self._interval: int = 900 self._interval: int = 900
self._task: typing.Optional["asyncio.Task[None]"] = None self._task: typing.Optional["asyncio.Task[None]"] = None
self.bot_id = None
@property @property
def is_running(self) -> bool: def is_running(self) -> bool:
""" """
@ -145,7 +149,7 @@ class AutoPost:
while True: while True:
stats = await self._stats() stats = await self._stats()
try: try:
await self.client.http.post_bot_stats(stats) await self.client.http.post_bot_stats(self.bot_id, stats)
except Exception as err: except Exception as err:
on_error = getattr(self, "_error", None) on_error = getattr(self, "_error", None)
if on_error: if on_error:
@ -160,14 +164,21 @@ class AutoPost:
await asyncio.sleep(self._interval) await asyncio.sleep(self._interval)
def start(self): def start(self, bot_id: typing.Union[str, int]):
""" """
Starts the loop. Starts the loop.
Args:
bot_id ( Union[:obj:`int`, :obj:`str`] )
Id of the bot to send stats of.
Raises: Raises:
:obj:`~.exceptions.InternalException` :obj:`~.exceptions.InternalException`
If there's no callback (for getting stats) provided or the autopost is already running. 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"): if not hasattr(self, "_stats"):
raise bexc.InternalException("You must provide stats") raise bexc.InternalException("You must provide stats")

View file

@ -8,11 +8,6 @@ from .autopost import AutoPost
class BoticordClient: class BoticordClient:
"""Represents a client that can be used to interact with the BotiCord API. """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 <https://docs.boticord.top/topics/v1vsv2/>`_
Note: Note:
Remember that every http method can return an http exception. Remember that every http method can return an http exception.
@ -20,52 +15,47 @@ class BoticordClient:
token (:obj:`str`) token (:obj:`str`)
Your bot's Boticord API Token. Your bot's Boticord API Token.
version (:obj:`int`) version (:obj:`int`)
BotiCord API version (Default: 2) BotiCord API version (Default: 3)
""" """
__slots__ = ("http", "_autopost", "_token") __slots__ = ("http", "_autopost", "_token")
http: HttpClient http: HttpClient
def __init__(self, token: str = None, version: int = 2): def __init__(self, token: str = None, version: int = 3):
self._token = token self._token = token
self._autopost: typing.Optional[AutoPost] = None self._autopost: typing.Optional[AutoPost] = None
self.http = HttpClient(token, version) 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. """Gets information about specified bot.
Args: Args:
bot_id (:obj:`int`) bot_id (Union[:obj:`str`, :obj:`int`])
Id of the bot Id of the bot
Returns: Returns:
:obj:`~.types.Bot`: :obj:`~.types.ResourceBot`:
Bot object. ResourceBot object.
""" """
response = await self.http.get_bot_info(bot_id) response = await self.http.get_bot_info(bot_id)
return boticord_types.Bot(**response) return boticord_types.ResourceBot.from_dict(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]
async def post_bot_stats( async def post_bot_stats(
self, servers: int = 0, shards: int = 0, users: int = 0 self,
) -> dict: bot_id: typing.Union[str, int],
*,
servers: int = 0,
shards: int = 0,
users: int = 0,
) -> boticord_types.ResourceBot:
"""Post Bot's stats. """Post Bot's stats.
Args: Args:
bot_id (Union[:obj:`str`, :obj:`)
Id of the bot to post stats of.
servers ( :obj:`int` ) servers ( :obj:`int` )
Bot's servers count Bot's servers count
shards ( :obj:`int` ) shards ( :obj:`int` )
@ -73,15 +63,15 @@ class BoticordClient:
users ( :obj:`int` ) users ( :obj:`int` )
Bot's users count Bot's users count
Returns: Returns:
:obj:`dict`: :obj:`~.types.ResourceBot`:
Boticord API Response status ResourceBot object.
""" """
response = await self.http.post_bot_stats( 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. """Gets information about specified server.
Args: Args:
@ -124,7 +114,7 @@ class BoticordClient:
response = await self.http.post_server_stats(payload) response = await self.http.post_server_stats(payload)
return response 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. """Gets information about specified user.
Args: Args:
@ -138,7 +128,7 @@ class BoticordClient:
response = await self.http.get_user_info(user_id) response = await self.http.get_user_info(user_id)
return boticord_types.UserProfile(**response) 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. """Gets comments of specified user.
Args: Args:
@ -185,9 +175,7 @@ class BoticordClient:
else boticord_types.ShortedLink(**response[0]) else boticord_types.ShortedLink(**response[0])
) )
async def create_shorted_link( async def create_shorted_link(self, *, code: str, link: str, domain=1):
self, *, code: str, link: str, domain: boticord_types.LinkDomain = 1
):
"""Creates new shorted link """Creates new shorted link
Args: Args:
@ -206,9 +194,7 @@ class BoticordClient:
return boticord_types.ShortedLink(**response) return boticord_types.ShortedLink(**response)
async def delete_shorted_link( async def delete_shorted_link(self, code: str, domain=1):
self, code: str, domain: boticord_types.LinkDomain = 1
):
"""Deletes shorted link """Deletes shorted link
Args: Args:

View file

@ -1,3 +1,6 @@
from enum import IntEnum
class BoticordException(Exception): class BoticordException(Exception):
"""Base exception class for boticordpy. """Base exception class for boticordpy.
This could be caught to handle any exceptions thrown from this library. This could be caught to handle any exceptions thrown from this library.
@ -29,26 +32,131 @@ class HTTPException(BoticordException):
def __init__(self, response): def __init__(self, response):
self.response = 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) super().__init__(fmt)
class Unauthorized(HTTPException): class StatusCodes(IntEnum):
"""Exception that's thrown when status code 401 occurs.""" """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): class HTTPErrors(IntEnum):
"""Exception that's thrown when status code 403 occurs.""" """Errors which BotiCord may return"""
UNKNOWN_ERROR = 0
"""Unknown error"""
class NotFound(HTTPException): INTERNAL_SERVER_ERROR = 1
"""Exception that's thrown when status code 404 occurs.""" """Server error (>500)"""
RATE_LIMITED = 2
"""Too many requests"""
class ToManyRequests(HTTPException): NOT_FOUND = 3
"""Exception that's thrown when status code 429 occurs.""" """Not found"""
FORBIDDEN = 4
"""Access denied"""
class ServerError(HTTPException): BAD_REQUEST = 5
"""Exception that's thrown when status code 500 or 503 occurs.""" """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"""

View file

@ -1,4 +1,5 @@
import asyncio import asyncio
import typing
import aiohttp import aiohttp
@ -20,9 +21,9 @@ class HttpClient:
loop: `asyncio loop` 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.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() 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): async def make_request(self, method: str, endpoint: str, **kwargs):
"""Send requests to the API""" """Send requests to the API"""
kwargs["headers"] = { kwargs["headers"] = {"Content-Type": "application/json"}
"Content-Type": "application/json",
"Authorization": self.token, if self.token is not None:
} kwargs["headers"]["Authorization"] = self.token
url = f"{self.API_URL}{endpoint}" url = f"{self.API_URL}{endpoint}"
async with self.session.request(method, url, **kwargs) as response: async with self.session.request(method, url, **kwargs) as response:
data = await response.json() data = await response.json()
if response.status == 200: if response.status == 200 or response.status == 201:
return data return data["result"]
elif response.status == 401: else:
raise exceptions.Unauthorized(response) raise exceptions.HTTPException(
elif response.status == 403: {"status": response.status, "error": data["errors"][0]["code"]}
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)
raise exceptions.HTTPException(response) def get_bot_info(self, bot_id: typing.Union[str, int]):
def get_bot_info(self, bot_id: int):
"""Get information about the specified bot""" """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): def post_bot_stats(self, bot_id: typing.Union[str, int], stats: dict):
"""Get list of specified bot comments"""
return self.make_request("GET", f"bot/{bot_id}/comments")
def post_bot_stats(self, stats: dict):
"""Post bot's stats""" """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: int):
"""Get information about specified server""" """Get information about specified server"""

View file

@ -1,375 +1,588 @@
import typing from datetime import datetime, timezone
from enum import IntEnum 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") from typing_extensions import get_type_hints
VT = typing.TypeVar("VT")
def parse_response_dict(input_data: dict) -> dict: KT = TypeVar("KT")
data = input_data.copy() VT = TypeVar("VT")
T = TypeVar("T")
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
def parse_with_information_dict(bot_data: dict) -> dict: class Singleton(type):
data = bot_data.copy() # Thanks to this stackoverflow answer (method 3):
# https://stackoverflow.com/q/6760685/12668716
_instances = {}
for key, value in data.copy().items(): def __call__(cls, *args, **kwargs):
if key.lower() == "links": if cls not in cls._instances:
converted_key = "page_links" cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
else: return cls._instances[cls]
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 parse_user_comments_dict(response_data: dict) -> dict: class TypeCache(metaclass=Singleton):
data = response_data.copy() # Thanks to Pincer Devs. This class is from the Pincer Library.
cache = {}
for key, value in data.copy().items(): def __init__(self):
data[key] = [SingleComment(**comment) for comment in value] 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]): def _asdict_ignore_none(obj: Generic[T]) -> Union[Tuple, Dict, T]:
"""Base class used to represent received data from the API.""" """
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: print(obj)
super().__init__(**parse_response_dict(kwargs))
self.__dict__ = self 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): class APIObjectBase:
"""This model represents single comment""" """
Represents an object which has been fetched from the BotiCord API.
"""
user_id: str def __attr_convert(self, attr_value: Dict, attr_type: T) -> T:
"""Comment's author Id (`str`)""" factory = attr_type
text: str # Always use `__factory__` over __init__
"""Comment content""" if getattr(attr_type, "__factory__", None):
factory = attr_type.__factory__
vote: int if attr_value is None:
"""Comment vote value (`-1,` `0`, `1`)""" return None
is_updated: bool if attr_type is not None and isinstance(attr_value, attr_type):
"""Was comment updated?""" return attr_value
created_at: int if isinstance(attr_value, dict):
"""Comment Creation date timestamp""" return factory(attr_value)
updated_at: int return factory(attr_value)
"""Last edit date timestamp"""
def __init__(self, **kwargs): def __post_init__(self):
super().__init__(**parse_response_dict(kwargs)) 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): class BotLibrary(IntEnum):
"""This model represents a bot, returned from the BotiCord API""" """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 id: str
"""Bot's Id""" """Bump's id"""
short_code: typing.Optional[str] expires: datetime
"""Bot's page short code""" """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 @classmethod
"""List of bot's page urls""" def from_dict(cls, data: dict):
"""Generate a ResourceUp from the given data.
server: dict Parameters
"""Bot's support server""" ----------
data: :class:`dict`
The dictionary to convert into a ResourceUp.
"""
bumps: int self: ResourceUp = super().__new__(cls)
"""Bumps count"""
added: str self.id = data["id"]
"""How many times users have added the bot?""" self.expires = datetime.fromtimestamp(
int(int(data["expires"]) / 1000), tz=timezone.utc
)
prefix: str return self
"""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))
class Server(ApiData): @dataclass(repr=False)
"""This model represents a server, returned from the Boticord API""" 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: str
"""Server's Id""" """Id of the user"""
short_code: typing.Optional[str] socials: UserLinks
"""Server's page short code""" """Links of the user"""
status: str description: Optional[str]
"""Server's approval status""" """Description of the user"""
page_links: list short_description: Optional[str]
"""List of server's page urls""" """Short description of the user"""
bot: dict status: Optional[str]
"""Bot where this server is used for support users""" """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: str
"""Name of the server""" """Name of the bot"""
avatar: str short_description: str
"""Server's avatar""" """Short description of the bot"""
members: list description: str
"""Members counts - `[all, online]`""" """Description of the bot"""
owner: typing.Optional[str] avatar: Optional[str]
"""Server's owner Id""" """Avatar of the bot"""
bumps: int short_link: Optional[str]
"""Bumps count""" """Short link to the bot's page"""
tags: list invite_link: str
"""Server's search-tags""" """Invite link"""
links: dict premium_active: bool
"""Server's social medias""" """Is premium status active? (True/False)"""
short_description: typing.Optional[str] premium_splash_url: Optional[str]
"""Server's short description""" """Link to the splash"""
long_description: typing.Optional[str] premium_auto_fetch: Optional[bool]
"""Server's long description""" """Is auto-fetch enabled? (True/False)"""
badge: typing.Optional[str] premium_banner_url: Optional[str]
"""Server's badge""" """Premium banner URL"""
def __init__(self, **kwargs): owner: str
super().__init__(**parse_with_information_dict(kwargs)) """Owner of the bot"""
status: ResourceStatus
"""Status of the bot"""
class UserProfile(ApiData): ratings: List[ResourceRating]
"""This model represents profile of user, returned from the Boticord API""" """Bot's ratings"""
id: str prefix: str
"""Id of User""" """Prefix of the bot"""
status: str discriminator: str
"""Status of user""" """Bot's discriminator"""
badge: typing.Optional[str] created_date: datetime
"""User's badge""" """Date when the bot was published"""
short_code: typing.Optional[str] support_server_invite_link: Optional[str]
"""User's profile page short code""" """Link to the support server"""
site: typing.Optional[str] library: Optional[BotLibrary]
"""User's website""" """The library that the bot is based on"""
vk: typing.Optional[str] guilds: Optional[int]
"""User's VK Profile""" """Number of guilds"""
steam: typing.Optional[str] shards: Optional[int]
"""User's steam account""" """Number of shards"""
youtube: typing.Optional[str] members: Optional[int]
"""User's youtube channel""" """Number of members"""
twitch: typing.Optional[str] website: Optional[str]
"""User's twitch channel""" """Link to bot's website"""
git: typing.Optional[str] tags: List[BotTag]
"""User's github/gitlab (or other git-service) profile""" """List of bot tags"""
def __init__(self, **kwargs): up_count: int
super().__init__(**parse_response_dict(kwargs)) """Number of ups"""
ups: List[ResourceUp]
"""List of bot's ups"""
class UserComments(ApiData): developers: List[PartialUser]
"""This model represents all the user's comments on every page""" """List of bot's developers"""
bots: list @classmethod
"""Data from `get_bot_comments` method""" def from_dict(cls, data: dict):
"""Generate a ResourceBot from the given data.
servers: list Parameters
"""Data from `get_server_comments` method""" ----------
data: :class:`dict`
The dictionary to convert into a ResourceBot.
"""
def __init__(self, **kwargs): self: ResourceBot = super().__new__(cls)
super().__init__(**parse_user_comments_dict(kwargs))
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): self.premium_active = data["premium"].get("active")
"""This model represents a short bot information (`id`, `short`). self.premium_splash_url = data["premium"].get("splashURL")
After that you can get more information about it using `get_bot_info` method.""" self.premium_auto_fetch = data["premium"].get("autoFetch")
self.premium_banner_url = data["premium"].get("bannerURL")
id: str self.status = ResourceStatus(data.get("status"))
"""Bot's Id""" 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] self.guilds = data.get("guilds")
"""Bot's page short code""" self.shards = data.get("shards")
self.members = data.get("members")
return self
def __init__(self, **kwargs):
super().__init__(**parse_response_dict(kwargs))
class LinkDomain:
class CommentData(ApiData): pass
"""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))

View file

@ -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

View file

@ -1 +1,2 @@
aiohttp aiohttp
typing_extensions

View file

@ -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"

View file

@ -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)