feat(modules): added errors handling module

locales/bot.pot

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PyKorone 1.0\n"
 "Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
-"POT-Creation-Date: 2024-07-18 18:53-0300\n"
+"POT-Creation-Date: 2024-07-18 21:42-0300\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <LL@li.org>\n"

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
     "magic-filter>=1.0.12",
     "tomlkit>=0.13.0",
     "sentry-sdk>=2.10.0",
+    "better-exceptions>=0.3.3",
 ]
 readme = "README.md"
 license = { file = "LICENSE" }

src/korone/decorators/factory.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING
 
 from hydrogram.filters import Filter
+from hydrogram.handlers import ErrorHandler
 
 from korone.handlers.callback_query_handler import KoroneCallbackQueryHandler
 from korone.handlers.message_handler import KoroneMessageHandler
@@ -17,7 +18,7 @@
 @dataclass(frozen=True, slots=True)
 class HandlerObject:
     func: Callable
-    filters: Filter
+    filters: Filter | None
     group: int
     event: Callable
 
@@ -31,9 +32,10 @@ def __init__(self, event_name: str) -> None:
         self.events_observed: dict[str, type[Handler]] = {
             "message": KoroneMessageHandler,
             "callback_query": KoroneCallbackQueryHandler,
+            "error": ErrorHandler,
         }
 
-    def __call__(self, filters: Filter, group: int = 0) -> Callable:
+    def __call__(self, filters: Filter | None = None, group: int = 0) -> Callable:
         def wrapper(func: Callable) -> Callable:
             func.on = self.event_name
             func.group = group

src/korone/decorators/router.py

@@ -9,11 +9,12 @@ class RouterError(Exception):
 
 
 class Router:
-    __slots__ = ("callback_query", "message")
+    __slots__ = ("callback_query", "error", "message")
 
     def __init__(self) -> None:
         self.message = Factory("message")
         self.callback_query = Factory("callback_query")
+        self.error = Factory("error")
 
     def __getattr__(self, name: str):
         msg = f"Event of type: '{name}' is not supported by the Korone."

src/korone/modules/core.py

@@ -41,7 +41,7 @@ def add_modules_to_dict() -> None:
             add_handlers(module_name, handlers_path)
 
 
-def get_method_callable(cls: type, key: str) -> Callable[..., Any]:  # numpydoc ignore=PR02
+def get_method_callable(cls: type, key: str) -> Callable[..., Any]:
     method = cls.__dict__.get(key)
     if isinstance(method, staticmethod):
         return method.__func__

src/korone/modules/errors/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>

src/korone/modules/errors/handlers/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>

src/korone/modules/errors/handlers/catcher.py

@@ -0,0 +1,91 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+import html
+import sys
+from typing import Any
+
+from better_exceptions import format_exception
+from hydrogram import Client
+from hydrogram.errors import ChatWriteForbidden, FloodWait
+from hydrogram.types import (
+    CallbackQuery,
+    InlineKeyboardButton,
+    InlineKeyboardMarkup,
+    Message,
+    Update,
+)
+from sentry_sdk import capture_exception
+
+from korone.decorators import router
+from korone.handlers.abstract import MessageHandler
+from korone.utils.logging import logger
+
+IGNORED_EXCEPTIONS: tuple[type[Exception], ...] = (FloodWait, ChatWriteForbidden)
+
+
+class ErrorsHandler(MessageHandler):
+    @router.error()
+    async def handle(self, client: Client, update: Update, exception: Exception) -> None:
+        if isinstance(exception, IGNORED_EXCEPTIONS):
+            return
+
+        etype, value, tb = sys.exc_info()
+
+        if not (etype and value and tb):
+            await logger.aerror("Failed to retrieve exception information", exception=exception)
+            return
+
+        sentry_event_id = self.capture_sentry(exception)
+        await self.log_to_console(etype, value, tb, sentry_event_id=sentry_event_id)
+
+        if isinstance(update, Message):
+            chat = update.chat
+        elif isinstance(update, CallbackQuery):
+            chat = update.message.chat
+        else:
+            await logger.aerror("Unhandled update type", update=update)
+            return
+
+        message_data = self.message_data(value, sentry_event_id)
+
+        await client.send_message(chat.id, **message_data)
+
+    @staticmethod
+    def capture_sentry(exception: Exception) -> str | None:
+        return capture_exception(exception)
+
+    @staticmethod
+    async def log_to_console(etype: type, value: BaseException, tb: Any, **kwargs: Any) -> None:
+        formatted_exception = format_exception(etype, value, tb)
+        await logger.aerror("".join(formatted_exception))
+        await logger.aerror("Additional error data", **kwargs)
+
+    @staticmethod
+    def get_error_message(exception: BaseException) -> str:
+        return " ".join(f"<i>{html.escape(x)}</i>" for x in exception.args)
+
+    def message_data(
+        self, exception: BaseException, sentry_event_id: str | None
+    ) -> dict[str, Any]:
+        text = "An error occurred while processing this update. :/"
+
+        error_message = self.get_error_message(exception)
+        text += f"\n<blockquote>{error_message}</blockquote>\n"
+
+        if sentry_event_id:
+            text += f"Reference ID: {sentry_event_id}"
+
+        return {
+            "text": text,
+            "reply_markup": InlineKeyboardMarkup(
+                inline_keyboard=[
+                    [
+                        InlineKeyboardButton(
+                            text="🐞 Report This Error",
+                            url="https://github.com/HitaloM/PyKorone/issues",
+                        )
+                    ]
+                ]
+            ),
+        }

src/korone/modules/errors/handlers/test.py

@@ -0,0 +1,17 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2024 Hitalo M. <https://github.com/HitaloM>
+
+from hydrogram import Client
+from hydrogram.types import Message
+
+from korone.decorators import router
+from korone.filters import Command, IsSudo
+from korone.handlers.abstract import MessageHandler
+
+
+class ErrorTest(MessageHandler):
+    @staticmethod
+    @router.message(Command("error") & IsSudo)
+    async def handle(client: Client, message: Message) -> None:
+        msg = "Error Test!"
+        raise Exception(msg)