some models and http

This commit is contained in:
grey-cat-1908 2022-02-10 15:30:28 +03:00
parent 00f574cd9a
commit bdf1658a60
21 changed files with 526 additions and 733 deletions

4
.flake8 Normal file
View file

@ -0,0 +1,4 @@
[flake8]
# For error codes, see this https://flake8.pycqa.org/en/latest/user/error-codes.html
ignore = F401, F403
max-line-length = 100

1
.gitignore vendored
View file

@ -8,3 +8,4 @@ dist
docs/_build docs/_build
boticordpy.egg-info boticordpy.egg-info
test.py test.py
/.pytest_cache

View file

@ -1,67 +1,3 @@
<h1 align="center">Boticordpy</h1> <h1 align="center">Boticordpy 2.0</h1>
<p align="center">Модуль для работы с <a href="https://boticord.top/">Boticord</a> API</p> <p align="center">Currently in development</p>
<p align="center">
<img src="https://img.shields.io/pypi/dm/boticordpy" alt="">
</p>
---
* [Документация](https://boticordpy.readthedocs.io/)
* [Исходный код](https://github.com/grey-cat-1908/boticordpy)
---
### Примеры
#### Без Когов
Публикуем статистику при запуске бота.
```Python
from discord.ext import commands
from boticordpy import BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "your-boticord-token")
@bot.event
async def on_ready():
stats = {"servers": len(bot.guilds), "shards": bot.shard_count, "users": len(bot.users)}
await boticord.Bots.post_stats(stats)
bot.run("your-bot-token")
```
#### С Когами
Ког с автоматической публикацией статистики раз в 15 минут + команда для публикации статистики для владельца бота.
```python
from discord.ext import commands
from boticordpy import BoticordClient
class BoticordCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.boticord = BoticordClient(self.bot, "your-boticord-token")
self.boticord.start_loop()
@commands.command(name="boticord-update")
@commands.is_owner()
async def boticord_update(self, ctx):
"""
This commands can be used by owner to post stats to boticord
"""
stats = {"servers": len(self.bot.guilds), "shards": 0, "users": len(self.bot.users)}
await self.boticord.Bots.post_stats(stats)
def setup(bot):
bot.add_cog(BoticordCog(bot))
```

View file

@ -2,16 +2,16 @@
Boticord API Wrapper Boticord API Wrapper
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
A basic wrapper for the Boticord API. A basic wrapper for the Boticord API.
:copyright: (c) 2021 Grey Cat :copyright: (c) 2022 Marakarka
:license: MIT, see LICENSE for more details. :license: MIT, see LICENSE for more details.
""" """
__title__ = 'boticordpy' __title__ = 'boticordpy'
__author__ = 'Grey Cat' __author__ = 'Marakarka'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright 2021 Grey Cat' __copyright__ = 'Copyright 2022 Marakarka'
__version__ = '2.0.0a' __version__ = '2.0.1a'
from .client import BoticordClient from .client import BoticordClient
from .webhook import BoticordWebhook
from .types import * from .types import *

View file

@ -1,97 +1,50 @@
from discord.ext import commands from . import types as boticord_types
from disnake.ext import commands as commandsnake from .http import HttpClient
import aiohttp
from typing import Union
import asyncio
from .modules import Bots, Servers, Users
class BoticordClient: class BoticordClient:
"""
This class is used to make it much easier to use the Boticord API.
You can pass `lib` parameter to specify the library. Supported: ["discordpy", "disnake"]
Parameters
----------
bot : :class:`commands.Bot` | :class:`commands.AutoShardedBot`
The discord.py Bot instance
token : :class:`str`
boticord api key
Attributes
----------
Bots : :class:`modules.bots.Bots`
:class:`modules.bots.Bots` with all arguments filled.
Servers : :class:`modules.servers.Servers`
:class:`modules.servers.Servers` with all arguments filled.
Users : :class:`modules.users.Users`
:class:`modules.users.Users` with all arguments filled.
"""
__slots__ = ( __slots__ = (
"Bots", "http"
"Servers",
"Users",
"bot",
"events",
"lib"
) )
bot: Union[commands.Bot, commands.AutoShardedBot, commandsnake.Bot, commandsnake.AutoShardedBot] http: HttpClient
def __init__(self, bot, token=None, **kwargs): def __init__(self, token=None, **kwargs):
loop = kwargs.get('loop') or asyncio.get_event_loop() self._token = token
session = kwargs.get('session') or aiohttp.ClientSession(loop=loop) self.http = HttpClient(token)
self.lib = kwargs.get('lib') or "discordpy"
self.events = {}
self.bot = bot
self.Bots = Bots(bot, token=token, loop=loop, session=session, lib=self.lib)
self.Servers = Servers(bot, token=token, loop=loop, session=session)
self.Users = Users(token=token, loop=loop, session=session)
def event(self, event_name: str): async def get_bot_info(self, bot_id: int):
""" response = await self.http.get_bot_info(bot_id)
A decorator that registers an event to listen to. return boticord_types.Bot(**response)
You can find all the events on Event Reference page.
Parameters async def get_bot_comments(self, bot_id: int):
---------- response = await self.http.get_bot_comments(bot_id)
event_name :class:`str` return [boticord_types.SingleComment(**comment) for comment in response]
boticord event name
"""
def inner(func):
if not asyncio.iscoroutinefunction(func):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
self.events[event_name] = func
return func
return inner
def start_loop(self, sleep_time: int = None) -> None: async def post_bot_stats(self, servers: int = 0, shards: int = 0, users: int = 0):
""" response = await self.http.post_bot_stats(servers, shards, users)
return response
Can be used to post stats automatically. async def get_server_info(self, server_id: int):
response = await self.http.get_server_info(server_id)
return boticord_types.Server(**response)
Parameters async def get_server_comments(self, server_id: int):
---------- response = await self.http.get_server_comments(server_id)
sleep_time: :class:`int` return [boticord_types.SingleComment(**comment) for comment in response]
stats posting interval - can be not specified or None (default interval - 15 minutes)
"""
self.bot.loop.create_task(self.__loop(sleep_time=sleep_time))
async def __loop(self, sleep_time: int = None) -> None: async def post_server_stats(self, payload: dict):
""" response = await self.post_server_stats(payload)
The internal loop used for automatically posting stats return response
"""
await self.bot.wait_until_ready()
while not self.bot.is_closed(): async def get_user_info(self, user_id: int):
response = await self.get_user_info(user_id)
return boticord_types.UserProfile(**response)
await self.Bots.post_stats() async def get_user_comments(self, user_id: int):
response = await self.get_user_comments(user_id)
return boticord_types.UserComments(**response)
if sleep_time is None: async def get_user_bots(self, user_id: int):
sleep_time = 900 response = await self.get_user_bots(user_id)
return [boticord_types.SimpleBot(**bot) for bot in response]
await asyncio.sleep(sleep_time)

View file

@ -1,39 +0,0 @@
from aiohttp import ClientResponse
from disnake.ext import commands as commandsnake
from discord.ext import commands
from typing import Union
import json
from . import exceptions
from . import types
class Config:
local_api = "https://boticord.top/api"
general_api = "https://api.boticord.top/v1"
http_exceptions = {401: exceptions.Unauthorized,
403: exceptions.Forbidden,
404: exceptions.NotFound,
429: exceptions.ToManyRequests,
500: exceptions.ServerError,
503: exceptions.ServerError}
events_list = {
"new_bot_comment": types.Comment,
"edit_bot_comment": types.EditedComment,
"delete_bot_comment": types.Comment,
"new_bot_bump": types.BotVote
}
libs = {
"discordpy": commands,
"disnake": commandsnake
}
async def _json_or_text(response: ClientResponse) -> Union[dict, str]:
text = await response.text()
if response.headers['Content-Type'] == 'application/json; charset=utf-8':
return json.loads(text)
return text

View file

@ -11,8 +11,6 @@ class HTTPException(BoticordException):
---------- ----------
response: response:
The response of the failed HTTP request. The response of the failed HTTP request.
message:
The text of the error. Could be an empty string.
""" """
def __init__(self, response): def __init__(self, response):

82
boticordpy/http.py Normal file
View file

@ -0,0 +1,82 @@
import asyncio
import aiohttp
from . import exceptions
class HttpClient:
def __init__(self, auth_token, **kwargs):
self.token = auth_token
self.API_URL = "https://api.boticord.top/v1/"
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):
kwargs["headers"] = {
"Content-Type": "application/json",
"Authorization": self.token
}
url = f"{self.API_URL}{endpoint}"
async with self.session.request(method,
url,
**kwargs) as response:
data = await response.json()
if response.status == 200:
return data
elif response.status == 401:
raise exceptions.Unauthorized(response)
elif response.status == 403:
raise exceptions.Forbidden(response)
elif response.status == 404:
raise exceptions.NotFound(response)
elif response.status == 429:
raise exceptions.ToManyRequests(response)
elif response.status == 500:
raise exceptions.ServerError(response)
elif response.status == 503:
raise exceptions.ServerError(response)
raise exceptions.HTTPException(response)
def get_bot_info(self, bot_id: int):
return self.make_request("GET", f"bot/{bot_id}")
def get_bot_comments(self, bot_id: int):
return self.make_request("GET", f"bot/{bot_id}/comments")
def post_bot_stats(self,
servers: int = 0,
shards: int = 0,
users: int = 0):
return self.make_request("POST", "stats", json={
"servers": servers,
"shards": shards,
"users": users
})
def get_server_info(self, server_id: int):
return self.make_request("GET", f"server/{server_id}")
def get_server_comments(self, server_id: int):
return self.make_request("GET", f"server/{server_id}/comments")
def post_server_stats(self, payload: dict):
return self.make_request("POST", "server", json=payload)
def get_user_info(self, user_id: int):
return self.make_request("GET", f"profile/{user_id}")
def get_user_comments(self, user_id: int):
return self.make_request("GET", f"user/{user_id}/comments")
def get_user_bots(self, user_id: int):
return self.make_request("GET", f"bots/{user_id}")

View file

@ -1,3 +0,0 @@
from .bots import Bots
from .servers import Servers
from .users import Users

View file

@ -1,89 +0,0 @@
from discord.ext import commands
import asyncio
import aiohttp
from ..config import Config, _json_or_text
class Bots:
"""
Class with methods to work with Boticord API Bots.
Parameters
----------
bot : :class:`commands.Bot` | :class:`commands.AutoShardedBot`
The discord.py Bot instance
"""
def __init__(self, bot, **kwargs):
self.bot = bot
self.token = kwargs.get('token')
self.loop = kwargs.get('loop') or asyncio.get_event_loop()
self.session = kwargs.get('session') or aiohttp.ClientSession(loop=self.loop)
self.lib = Config.libs.get(kwargs.get("lib"))
async def get_bot_info(self, bot_id: int):
"""
Returns information about discord bot with the given ID.
Parameters
----------
bot_id : :class:`int`
Discord Bot's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/bot/{bot_id}', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def get_bot_comments(self, bot_id: int):
"""
Returns comments of the discord bot with the given ID.
Parameters
----------
bot_id : :class:`int`
Discord Bot's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/bot/{bot_id}/comments', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def post_stats(self, stats: dict = None):
"""
Post stats to Boticord API.
Parameters
----------
stats: :class:`dict`
A dictionary of {``guilds``: :class:`int`, ``shards``: :class:`int`, ``users``: :class:`int`}
"""
if not self.token:
return "Require Authentication"
headers = {"Authorization": self.token}
if stats is None:
data_to_send = {"servers": len(self.bot.guilds), "users": len(self.bot.users)}
if isinstance(self.bot, self.lib.AutoShardedBot):
data_to_send["shards"] = self.bot.shard_count
async with self.session.post(f'{Config.general_api}/stats', headers=headers, json=stats) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data

View file

@ -1,101 +0,0 @@
import asyncio
import aiohttp
import discord
from ..config import Config, _json_or_text
class Servers:
"""
Class with methods to work with Boticord API Servers.
Parameters
----------
bot : :class:`commands.Bot` | :class:`commands.AutoShardedBot`
The discord.py Bot instance
"""
def __init__(self, bot, **kwargs):
self.bot = bot
self.token = kwargs.get('token')
self.loop = kwargs.get('loop') or asyncio.get_event_loop()
self.session = kwargs.get('session') or aiohttp.ClientSession(loop=self.loop)
async def get_server_info(self, server_id: int):
"""
Returns information about discord server with the given ID.
Parameters
----------
server_id : :class:`int`
Discord Server's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/server/{server_id}', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def get_server_comments(self, server_id: int):
"""
Returns comments of the discord server with the given ID.
Parameters
----------
server_id : :class:`int`
Discord Server's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/server/{server_id}/comments', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def post_server_stats(self, message: discord.Message, custom_stats: dict = None):
"""
Post server stats to Boticord API.
Parameters
----------
message: :class:`discord.Message`
Message object of used command.
custom_stats: :class:`dict`
Dict with custom server stats. (Optional)
"""
if not self.token:
return "Require Authentication"
if custom_stats is None:
guild = message.guild
guild_owner = guild.owner
stats = {
"server_id": str(guild.id),
"up": 1,
"status": 1,
"serverName": guild.name,
"serverAvatar": str(guild.icon_url),
"serverMembersAllCount": guild.member_count,
"serverMembersOnlineCount": 0,
"serverOwnerTag": guild_owner.name + "#" + guild_owner.discriminator,
"serverOwnerID": str(guild_owner.id)
}
else:
stats = custom_stats
headers = {"Authorization": self.token}
async with self.session.post(f'{Config.general_api}/server', headers=headers, json=stats) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data

View file

@ -1,71 +0,0 @@
import asyncio
import aiohttp
from ..config import Config, _json_or_text
class Users:
"""
Class with methods to work with Boticord API Users.
"""
def __init__(self, **kwargs):
self.token = kwargs.get('token')
self.loop = kwargs.get('loop') or asyncio.get_event_loop()
self.session = kwargs.get('session') or aiohttp.ClientSession(loop=self.loop)
async def get_user(self, user_id: int):
"""
Returns information about discord user with the given ID.
Parameters
----------
user_id : :class:`int`
Discord User's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/profile/{user_id}', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def get_user_comments(self, user_id: int):
"""
Returns comments of discord user with the given ID.
Parameters
----------
user_id : :class:`int`
Discord User's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/profile/{user_id}/comments', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data
async def get_user_bots(self, user_id: int):
"""
Returns bots of discord user with the given ID.
Parameters
----------
user_id : :class:`int`
Discord User's ID
"""
headers = {"Authorization": self.token}
async with self.session.get(f'{Config.general_api}/bots/{user_id}', headers=headers) as resp:
data = await _json_or_text(resp)
status = Config.http_exceptions.get(resp.status)
if status is not None:
raise status(resp)
return data

View file

@ -1,111 +1,261 @@
from datetime import datetime import typing
KT = typing.TypeVar("KT")
VT = typing.TypeVar("VT")
class CommentData: def parse_response_dict(input_data: dict) -> dict:
"""Model that represents edited comment text data. data = input_data.copy()
Attributes for key, value in data.copy().items():
----------- converted_key = "".join(
old : :class:`str` or :class:`None` ["_" + x.lower() if x.isupper() else x for x in key]
Old comment text. ).lstrip("_")
new : :class:`str` or :class:`None`
New comment text. if key != converted_key:
del data[key]
data[converted_key] = value
return data
def parse_with_information_dict(bot_data: dict) -> dict:
data = bot_data.copy()
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 parse_user_comments_dict(response_data: dict) -> dict:
data = response_data.copy()
for key, value in data.copy().items():
data[key] = [SingleComment(**comment) for comment in value]
return data
class ApiData(dict, typing.MutableMapping[KT, VT]):
"""Base class used to represent received data from the API.
""" """
__slots__ = "old", "new" def __init__(self, **kwargs: VT) -> None:
super().__init__(**parse_response_dict(kwargs))
old: str or None self.__dict__ = self
new: str or None
def __init__(self, raw_data):
self.old = raw_data.get("old")
self.new = raw_data.get("new")
class Comment: class SingleComment(ApiData):
"""Model that represents information about a comment. """This model represents single comment"""
Attributes user_id: str
----------- """Comment's author Id (`str`)"""
raw_data : :class:`dict`
Raw data from the Boticord API.
user_id : :class:`int`
ID of comment author.
comment : :class:`str`
Comment.
at : :class:`datetime.datetime`
The comment creation time.
"""
__slots__ = "raw_data", "user_id", "comment", "at" text: str
"""Comment content"""
raw_data: dict vote: int
user_id: int """Comment vote value (`-1,` `0`, `1`)"""
comment: str
at: datetime
def __init__(self, raw_data): is_updated: bool
self.raw_data = raw_data["data"] """Was comment updated?"""
self.user_id = int(self.raw_data["user"])
self.comment = self.raw_data["comment"]
self.at = datetime.fromtimestamp(self.raw_data["at"] / 1000)
def __repr__(self) -> str: created_at: int
name = self.__class__.__name__ """Comment Creation date timestamp"""
return (
f'<{name} user_id={self.user_id} comment={self.comment}>' updated_at: int
) """Last edit date timestamp"""
def __init__(self, **kwargs):
super().__init__(**parse_response_dict(kwargs))
class EditedComment(Comment): class Bot(ApiData):
"""Model that represents information about edited comment. """This model represents a bot, returned from the Boticord API"""
It is inherited from :class:`Comment` id: str
"""Bot's Id"""
Attributes short_code: typing.Optional[str]
----------- """Bot's page short code"""
comment : :class:`CommentData`
Comment.
"""
__slots__ = "raw_data", "user_id", "comment", "at" page_links: list
"""List of bot's page urls"""
comment: CommentData server: dict
"""Bot's support server"""
def __init__(self, raw_data): bumps: int
super().__init__(raw_data) """Bumps count"""
self.comment = CommentData(self.raw_data["comment"])
def __repr__(self) -> str: prefix: str
name = self.__class__.__name__ """How many times users have added the bot?"""
return (
f'<{name} user_id={self.user_id} comment={self.comment.new}>' 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[int]
"""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 BotVote: class Server(ApiData):
"""Model that represents information about bot's vote. """This model represents a server, returned from the Boticord API"""
Attributes id: str
----------- """Server's Id"""
raw_data : :class:`dict`
Raw data from the Boticord API.
user_id : :class:`int`
ID of user, who voted.
at : :class:`datetime.datetime`
Voting date.
"""
__slots__ = "raw_data", "user_id", "at" short_code: typing.Optional[str]
"""Server's page short code"""
raw_data: dict status: str
user_id: int """Server's approval status"""
at: datetime
def __init__(self, raw_data): page_links: list
self.raw_data = raw_data["data"] """List of server's page urls"""
self.user_id = int(self.raw_data["user"])
self.at = datetime.fromtimestamp(self.raw_data["at"] / 1000)
def __repr__(self) -> str: bot: dict
name = self.__class__.__name__ """Bot where this server is used for support users"""
return f'<{name} user_id={self.user_id}'
name: str
"""Name of the server"""
avatar: str
"""Server's avatar"""
members: list
"""Members counts - `[all, onlinw]`"""
owner: typing.Optional[str]
"""Server's owner Id"""
bumps: int
"""Bumps count"""
tags: list
"""Server's search-tags"""
links: dict
"""Server's social medias"""
short_description: typing.Optional[str]
"""Server's short description"""
long_description: typing.Optional[str]
"""Server's long description"""
badge: typing.Optional[str]
"""Server's badge"""
def __init__(self, **kwargs):
super().__init__(**parse_with_information_dict(kwargs))
class UserProfile(ApiData):
"""This model represents profile of user, returned from the Boticord API"""
id: str
"""Id of User"""
status: str
"""Status of user"""
badge: typing.Optional[str]
"""User's badge"""
short_code: typing.Optional[str]
"""User's profile page short code"""
site: typing.Optional[str]
"""User's website"""
vk: typing.Optional[str]
"""User's VK Profile"""
steam: typing.Optional[str]
"""User's steam account"""
youtube: typing.Optional[str]
"""User's youtube channel"""
twitch: typing.Optional[str]
"""User's twitch channel"""
git: typing.Optional[str]
"""User's github/gitlab (or other git-service) profile"""
def __init__(self, **kwargs):
super().__init__(**parse_response_dict(kwargs))
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]
def __init__(self, **kwargs):
super().__init__(**parse_response_dict(kwargs))

View file

@ -1,119 +0,0 @@
import sys
from typing import Dict, Union
if sys.version_info >= (3, 8):
from typing import TypedDict
else:
from typing_extensions import TypedDict
from discord.ext.commands import Bot, AutoShardedBot
from disnake.ext import commands as commandsnake
from aiohttp.web_urldispatcher import _WebHandler
from aiohttp import web
import aiohttp
from . import BoticordClient
from . import config
class _Webhook(TypedDict):
route: str
hook_key: str
func: "_WebHandler"
class BoticordWebhook:
"""
This class is used as a manager for the Boticord webhook.
Parameters
----------
bot :class:`commands.Bot` | :class:`commands.AutoShardedBot`
The discord.py Bot instance
"""
__app: web.Application
_webhooks: Dict[
str,
_Webhook,
]
_webserver: web.TCPSite
def __init__(self, bot: Union[Bot,
AutoShardedBot,
commandsnake.Bot,
commandsnake.AutoShardedBot], boticord_client: BoticordClient):
self.bot = bot
self.boticord_client = boticord_client
self._webhooks = {}
self.__app = web.Application()
def bot_webhook(self, route: str = "/bot", hook_key: str = "") -> "BoticordWebhook":
"""This method may be used to configure the route of boticord bot's webhook.
Parameters
----------
route : :class:`str`
Bot's webhook route. Must start with ``/``. Defaults - ``/bot``.
hook_key : :class:`str`
Webhook authorization key.
Returns
----------
:class:`BoticordWebhook`
"""
self._webhooks["bot"] = _Webhook(
route=route or "/bot",
hook_key=hook_key or "",
func=self._bot_webhook_interaction_handler,
)
return self
async def _bot_webhook_interaction_handler(self, request: aiohttp.web.Request) -> web.Response:
auth = request.headers.get("X-Hook-Key")
if auth == self._webhooks["bot"]["hook_key"]:
data = await request.json()
event_in_config = config.Config.events_list.get(data["type"])
if event_in_config is not None:
data_for_event = event_in_config(data)
else:
data_for_event = data
try:
await self.boticord_client.events[data["type"]](data_for_event)
except:
pass
return web.Response(status=200)
return web.Response(status=401)
async def _run(self, port: int):
for webhook in self._webhooks.values():
self.__app.router.add_post(webhook["route"], webhook["func"])
runner = web.AppRunner(self.__app)
await runner.setup()
self._webserver = web.TCPSite(runner, "0.0.0.0", port)
await self._webserver.start()
def run(self, port: int):
"""Runs the webhook.
Parameters
----------
port
The port to run the webhook on.
"""
self.bot.loop.create_task(self._run(port))
async def close(self) -> None:
"""Stops the webhook."""
await self._webserver.stop()

0
conftest.py Normal file
View file

View file

@ -1,14 +0,0 @@
from disnake.ext import commands
from boticordpy import BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "your-boticord-token", lib="disnake")
@bot.event
async def on_ready():
await boticord.Bots.post_stats()
bot.run("your-bot-token")

View file

@ -1,23 +0,0 @@
from discord.ext import commands
from boticordpy import BoticordClient
class BoticordCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.boticord = BoticordClient(self.bot, "your-boticord-token")
self.boticord.start_loop()
@commands.command(name="boticord-update")
@commands.is_owner()
async def boticord_update(self, ctx):
"""
This commands can be used by owner to post stats to boticord
"""
stats = {"servers": len(self.bot.guilds), "shards": 0, "users": len(self.bot.users)}
await self.boticord.Bots.post_stats(stats)
def setup(bot):
bot.add_cog(BoticordCog(bot))

View file

@ -1,14 +0,0 @@
from discord.ext import commands
from boticordpy import BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "your-boticord-token")
@bot.event
async def on_ready():
await boticord.Bots.post_stats()
bot.run("your-bot-token")

View file

@ -1,17 +0,0 @@
from discord.ext import commands
from boticordpy import BoticordWebhook, BoticordClient
bot = commands.Bot(command_prefix="!")
boticord = BoticordClient(bot, "boticord-api-token")
boticord_webhook = BoticordWebhook(bot, boticord).bot_webhook("/bot", "X-Hook-Key")
boticord_webhook.run(5000)
@boticord.event("edit_bot_comment")
async def on_boticord_comment_edit(data):
print(data)
bot.run("bot-token")

View file

@ -1,4 +1,2 @@
discord.py aiohttp~=3.7.4
aiohttp~=3.7.4.post0 setuptools==58.2.0
setuptools==58.2.0
disnake~=2.1.2

161
tests/test_converting.py Normal file
View file

@ -0,0 +1,161 @@
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)