add link shortener

This commit is contained in:
grey-cat-1908 2022-04-17 20:32:39 +03:00
parent e101a948c4
commit f4a136a348
7 changed files with 149 additions and 37 deletions

2
.gitignore vendored
View file

@ -7,6 +7,6 @@ __pycache__
dist dist
docs/_build docs/_build
boticordpy.egg-info boticordpy.egg-info
test.py _testing.py
/.pytest_cache /.pytest_cache
/docs/build/ /docs/build/

View file

@ -1,16 +1,16 @@
""" """
Boticord API Wrapper Boticord API Wrapper
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
A basic wrapper for the Boticord API. A basic wrapper for the BotiCord API.
:copyright: (c) 2022 Marakarka :copyright: (c) 2022 Marakarka
:license: MIT, see LICENSE for more details. :license: MIT, see LICENSE for more details.
""" """
__title__ = 'boticordpy' __title__ = "boticordpy"
__author__ = 'Marakarka' __author__ = "Marakarka"
__license__ = 'MIT' __license__ = "MIT"
__copyright__ = 'Copyright 2022 Marakarka' __copyright__ = "Copyright 2022 Marakarka"
__version__ = '2.1.0' __version__ = "2.1.0"
from .client import BoticordClient from .client import BoticordClient
from .webhook import Webhook from .webhook import Webhook

View file

@ -20,7 +20,7 @@ class AutoPost:
"_error", "_error",
"_stats", "_stats",
"_task", "_task",
"_stopped" "_stopped",
) )
_success: typing.Any _success: typing.Any
@ -134,7 +134,9 @@ class AutoPost:
Boticord recommends not to set interval lower than 900 seconds! Boticord recommends not to set interval lower than 900 seconds!
""" """
if seconds < 900: if seconds < 900:
raise ValueError("no. Boticord recommends not to set interval lower than 900 seconds!") raise ValueError(
"no. Boticord recommends not to set interval lower than 900 seconds!"
)
self._interval = seconds self._interval = seconds
return self return self
@ -170,7 +172,9 @@ class AutoPost:
raise bexc.InternalException("You must provide stats") raise bexc.InternalException("You must provide stats")
if self.is_running: if self.is_running:
raise bexc.InternalException("Automatically stats posting is already running") raise bexc.InternalException(
"Automatically stats posting is already running"
)
task = asyncio.ensure_future(self._internal_loop()) task = asyncio.ensure_future(self._internal_loop())
self._task = task self._task = task
@ -178,7 +182,7 @@ class AutoPost:
def stop(self) -> None: def stop(self) -> None:
""" """
Stops the autopost. Stops the autopost.
""" """
if not self.is_running: if not self.is_running:
return None return None

View file

@ -15,11 +15,8 @@ class BoticordClient:
token (:obj:`str`) token (:obj:`str`)
Your bot's Boticord API Token. Your bot's Boticord API Token.
""" """
__slots__ = (
"http", __slots__ = ("http", "_autopost", "_token")
"_autopost",
"_token"
)
http: HttpClient http: HttpClient
@ -56,7 +53,9 @@ class BoticordClient:
response = await self.http.get_bot_comments(bot_id) response = await self.http.get_bot_comments(bot_id)
return [boticord_types.SingleComment(**comment) for comment in response] return [boticord_types.SingleComment(**comment) for comment in response]
async def post_bot_stats(self, servers: int = 0, shards: int = 0, users: int = 0) -> dict: async def post_bot_stats(
self, servers: int = 0, shards: int = 0, users: int = 0
) -> dict:
"""Post Bot's stats. """Post Bot's stats.
Args: Args:
@ -70,11 +69,9 @@ class BoticordClient:
:obj:`dict`: :obj:`dict`:
Boticord API Response status Boticord API Response status
""" """
response = await self.http.post_bot_stats({ response = await self.http.post_bot_stats(
"servers": servers, {"servers": servers, "shards": shards, "users": users}
"shards": shards, )
"users": users
})
return response return response
async def get_server_info(self, server_id: int) -> boticord_types.Server: async def get_server_info(self, server_id: int) -> boticord_types.Server:
@ -160,6 +157,61 @@ class BoticordClient:
response = await self.http.get_user_bots(user_id) response = await self.http.get_user_bots(user_id)
return [boticord_types.SimpleBot(**bot) for bot in response] return [boticord_types.SimpleBot(**bot) for bot in response]
async def get_my_shorted_links(self, *, code: str = None):
"""Gets shorted links of an authorized user
Args:
code (:obj:`str`)
Code of shorted link. Could be None.
Returns:
Union[:obj:`list` [ :obj:`~.types.ShortedLink` ], :obj:`~types.ShortedLink`]:
List of shorted links if none else shorted link
"""
response = await self.http.get_my_shorted_links(code)
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)
def autopost(self) -> AutoPost: def autopost(self) -> AutoPost:
"""Returns a helper instance for auto-posting. """Returns a helper instance for auto-posting.

View file

@ -3,6 +3,7 @@ import asyncio
import aiohttp import aiohttp
from . import exceptions from . import exceptions
from .types import LinkDomain
class HttpClient: class HttpClient:
@ -23,26 +24,21 @@ class HttpClient:
self.token = auth_token self.token = auth_token
self.API_URL = "https://api.boticord.top/v1/" self.API_URL = "https://api.boticord.top/v1/"
loop = kwargs.get('loop') or asyncio.get_event_loop() loop = kwargs.get("loop") or asyncio.get_event_loop()
self.session = kwargs.get('session') or aiohttp.ClientSession(loop=loop) self.session = kwargs.get("session") or aiohttp.ClientSession(loop=loop)
async def make_request(self, async def make_request(self, method: str, endpoint: str, **kwargs):
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 "Authorization": self.token,
} }
url = f"{self.API_URL}{endpoint}" url = f"{self.API_URL}{endpoint}"
async with self.session.request(method, async with self.session.request(method, url, **kwargs) as response:
url,
**kwargs) as response:
data = await response.json() data = await response.json()
if response.status == 200: if response.status == 200:
@ -97,3 +93,21 @@ class HttpClient:
def get_user_bots(self, user_id: int): def get_user_bots(self, user_id: int):
"""Get bots of specified user""" """Get bots of specified user"""
return self.make_request("GET", f"bots/{user_id}") 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"""
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)}
)

View file

@ -1,4 +1,5 @@
import typing import typing
from enum import IntEnum
KT = typing.TypeVar("KT") KT = typing.TypeVar("KT")
VT = typing.TypeVar("VT") VT = typing.TypeVar("VT")
@ -59,8 +60,7 @@ def parse_user_comments_dict(response_data: dict) -> dict:
class ApiData(dict, typing.MutableMapping[KT, VT]): class ApiData(dict, typing.MutableMapping[KT, VT]):
"""Base class used to represent received data from the API. """Base class used to represent received data from the API."""
"""
def __init__(self, **kwargs: VT) -> None: def __init__(self, **kwargs: VT) -> None:
super().__init__(**parse_response_dict(kwargs)) super().__init__(**parse_response_dict(kwargs))
@ -94,6 +94,7 @@ class SingleComment(ApiData):
class Bot(ApiData): class Bot(ApiData):
"""This model represents a bot, returned from the BotiCord API""" """This model represents a bot, returned from the BotiCord API"""
id: str id: str
"""Bot's Id""" """Bot's Id"""
@ -340,3 +341,38 @@ class CommentResponse(ApiData):
def __init__(self, **kwargs): def __init__(self, **kwargs):
super().__init__(**parse_webhook_response_dict(kwargs)) super().__init__(**parse_webhook_response_dict(kwargs))
class LinkDomain(IntEnum):
"""Domain to short the link"""
BCORD_CC = 1
"""``bcord.cc`` domain, default"""
MYSERVERS_ME = 2
"""``myservers.me`` domain"""
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

@ -20,6 +20,7 @@ class Webhook:
Keyword Arguments: Keyword Arguments:
loop: `asyncio loop` loop: `asyncio loop`
""" """
__slots__ = ( __slots__ = (
"_webserver", "_webserver",
"_listeners", "_listeners",
@ -27,7 +28,7 @@ class Webhook:
"__app", "__app",
"_endpoint_name", "_endpoint_name",
"_x_hook_key", "_x_hook_key",
"_loop" "_loop",
) )
__app: web.Application __app: web.Application
@ -39,7 +40,7 @@ class Webhook:
self._listeners = {} self._listeners = {}
self.__app = web.Application() self.__app = web.Application()
self._is_running = False self._is_running = False
self._loop = kwargs.get('loop') or asyncio.get_event_loop() self._loop = kwargs.get("loop") or asyncio.get_event_loop()
def listener(self, response_type: str): def listener(self, response_type: str):
"""Decorator to set the listener. """Decorator to set the listener.
@ -47,6 +48,7 @@ class Webhook:
response_type (:obj:`str`) response_type (:obj:`str`)
Type of response (Check reference page) Type of response (Check reference page)
""" """
def inner(func): def inner(func):
if not asyncio.iscoroutinefunction(func): if not asyncio.iscoroutinefunction(func):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function") raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
@ -80,7 +82,11 @@ class Webhook:
if responder is not None: if responder is not None:
await responder( await responder(
(BumpResponse if data["type"].endswith("_bump") else CommentResponse)(**data) (
BumpResponse
if data["type"].endswith("_bump")
else CommentResponse
)(**data)
) )
return web.Response(status=200) return web.Response(status=200)