diff --git a/boticordpy/types.py b/boticordpy/types.py index a4acad0..a3c3303 100644 --- a/boticordpy/types.py +++ b/boticordpy/types.py @@ -259,3 +259,61 @@ class SimpleBot(ApiData): def __init__(self, **kwargs): super().__init__(**parse_response_dict(kwargs)) + + +class CommentData(ApiData): + """This model represents comment data (from webhook response)""" + + vote: dict + old: typing.Optional[str] + new: typing.Optional[str] + + 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 + user: str + at: int + + def __init__(self, **kwargs): + super().__init__(**parse_webhook_response_dict(kwargs)) + + +class CommentResponse(ApiData): + """This model represents a webhook response (`comment`).""" + + type: str + user: str + comment: CommentData + reason: typing.Optional[str] + at: int + + def __init__(self, **kwargs): + super().__init__(**parse_webhook_response_dict(kwargs)) + diff --git a/boticordpy/webhook.py b/boticordpy/webhook.py new file mode 100644 index 0000000..9dfd727 --- /dev/null +++ b/boticordpy/webhook.py @@ -0,0 +1,94 @@ +import asyncio +import typing + +from .types import BumpResponse, CommentResponse + +from aiohttp import web +import aiohttp + + +class Webhook: + __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): + 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): + 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: + 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: + self._loop.create_task(self._run(port)) + + @property + def is_running(self) -> bool: + return self._is_running + + @property + def listeners(self) -> dict: + return self._listeners + + @property + def app(self) -> web.Application: + return self.__app + + async def close(self) -> None: + await self._webserver.stop() + + self._is_running = False diff --git a/examples/webhooks.py b/examples/webhooks.py new file mode 100644 index 0000000..79c675a --- /dev/null +++ b/examples/webhooks.py @@ -0,0 +1,16 @@ +# You can use disnake or nextcord or something like this. + +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")