webhooks

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

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

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