refactor(tg_bot): reorganize bot structure

src/docker-entrypoint.sh

@@ -1 +1,9 @@
+#!/bin/sh
+
+if [ "$DEV_MODE" = "False" ]; then
+      echo "RELEASE MODE"
+else
+      echo "DEBUG MODE"
+fi
+
 python main.py

src/main.py

@@ -1,6 +1,5 @@
 import asyncio
 
-import bot_start
 import tg_bot
 from container import AppContainer
 from core.settings import app_settings, image_settings
@@ -11,9 +10,9 @@ def main():
     app_container.settings.from_dict(app_settings.model_dump())
     app_container.image_settings.from_dict(image_settings.model_dump())
     app_container.init_resources()
-    app_container.wire(packages=[tg_bot], modules=[bot_start])
+    app_container.wire(packages=[tg_bot])
 
-    asyncio.run(bot_start.run())
+    asyncio.run(tg_bot.run())
 
 
 if __name__ == "__main__":

src/tg_bot/__init__.py

@@ -0,0 +1 @@
+from .main import run

src/tg_bot/commands.py

@@ -2,8 +2,6 @@
 
 from aiogram import Bot
 from aiogram.types import BotCommand, BotCommandScopeChat, BotCommandScopeDefault
-from container import AppContainer
-from dependency_injector.wiring import Provide, inject
 
 from .enums import CartographyCommandsEnum, GeodesyCommandsEnum, UtilCommandsEnum
 
@@ -32,11 +30,10 @@
 BotCommand_all = BotCommand_default + BotCommand_dev + BotCommand_admin
 
 
-@inject
 async def set_commands(
     bot: Bot,
-    dev_mode: bool = Provide[AppContainer.settings.dev_mode],
-    admin_id: int = Provide[AppContainer.settings.admin_id],
+    dev_mode: bool,
+    admin_id: int,
 ):
     await bot.set_my_commands(
         BotCommand_admin + BotCommand_default,

src/tg_bot/handlers/__init__.py

@@ -0,0 +1 @@
+from .main import register_all_handlers

src/tg_bot/handlers/business/__init__.py

@@ -0,0 +1 @@
+from .main import register_business_handlers

src/tg_bot/handlers/business/main.py

@@ -0,0 +1,12 @@
+from aiogram import Dispatcher
+
+from . import geography_coordinates, micrometer, middle_values, nomenclature_title
+
+
+def register_business_handlers(dp: Dispatcher) -> None:
+    dp.include_routers(
+        geography_coordinates.router,
+        micrometer.router,
+        middle_values.router,
+        nomenclature_title.router,
+    )

src/tg_bot/handlers/callbacks/__init__.py

@@ -0,0 +1 @@
+from .main import register_callback_handlers

src/tg_bot/handlers/callbacks/main.py

@@ -0,0 +1,7 @@
+from aiogram import Dispatcher
+
+from . import help
+
+
+def register_callback_handlers(dp: Dispatcher) -> None:
+    dp.include_routers(help.router)

src/tg_bot/handlers/dev/__init__.py

@@ -0,0 +1 @@
+from .main import register_utils_handlers

src/tg_bot/handlers/dev/main.py

@@ -0,0 +1,10 @@
+from aiogram import Dispatcher
+
+from . import dev_handlers, util_handlers
+
+
+def register_utils_handlers(dp: Dispatcher) -> None:
+    dp.include_routers(
+        util_handlers.router,
+        dev_handlers.router,
+    )

src/tg_bot/handlers/main.py

@@ -0,0 +1,10 @@
+from aiogram import Dispatcher
+
+from . import business, callbacks, dev
+
+
+def register_all_handlers(dp: Dispatcher, dev_mode: bool) -> None:
+    if dev_mode:
+        dev.register_utils_handlers(dp)
+    business.register_business_handlers(dp)
+    callbacks.register_callback_handlers(dp)

src/tg_bot/main.py

@@ -0,0 +1,21 @@
+from aiogram import Bot, Dispatcher
+from container import AppContainer
+from dependency_injector.wiring import Provide, inject
+
+from . import commands, handlers, middlewares
+
+
+@inject
+async def run(
+    bot_token: str = Provide[AppContainer.settings.bot_token],
+    dev_mode: bool = Provide[AppContainer.settings.dev_mode],
+    admin_id: bool = Provide[AppContainer.settings.admin_id],
+) -> None:
+    bot = Bot(bot_token, parse_mode="HTML")
+    dp = Dispatcher()
+
+    await commands.set_commands(bot=bot, dev_mode=dev_mode, admin_id=admin_id)
+    handlers.register_all_handlers(dp, dev_mode)
+    middlewares.register_all_middlewares(dp, dev_mode, admin_id)
+
+    await dp.start_polling(bot)

src/tg_bot/middlewares/__init__.py

@@ -0,0 +1 @@
+from .main import register_all_middlewares

src/tg_bot/middlewares.py → src/tg_bot/middlewares/error_handler.py

@@ -3,45 +3,11 @@
 import structlog
 from aiogram import BaseMiddleware
 from aiogram.types import CallbackQuery, Message, TelegramObject
-from container import AppContainer
-from dependency_injector.wiring import Provide, inject
 from errors import BaseMsgError
 
 logger = structlog.get_logger()
 
 
-class IsAdminMiddleWare(BaseMiddleware):
-    @staticmethod
-    @inject
-    def _is_admin(
-        user_id: int,
-        admin_id: int = Provide[AppContainer.settings.admin_id],
-    ) -> bool:
-        if user_id == admin_id:
-            return True
-        return False
-
-    async def __call__(
-        self,
-        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
-        event: TelegramObject,
-        data: Dict[str, Any],
-    ) -> Any:
-        if self._is_admin(event.chat.id):  # type: ignore
-            return await handler(event, data)
-
-
-class LoggingChatActions(BaseMiddleware):
-    async def __call__(
-        self,
-        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
-        event: TelegramObject,
-        data: Dict[str, Any],
-    ) -> Any:
-        logger.info(event.text, username=event.chat.username, chat_id=event.chat.id)  # type: ignore
-        return await handler(event, data)
-
-
 class ErrorHandlerMiddleware(BaseMiddleware):
     async def __call__(
         self,

src/tg_bot/middlewares/is_admin.py

@@ -0,0 +1,29 @@
+from typing import Any, Awaitable, Callable, Dict
+
+import structlog
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject
+
+logger = structlog.get_logger()
+
+
+class BlockNonAdminMiddleware(BaseMiddleware):
+    def __init__(self, admin_id: int) -> None:
+        self.admin_id = admin_id
+
+    def _is_admin(
+        self,
+        user_id: int,
+    ) -> bool:
+        if user_id == self.admin_id:
+            return True
+        return False
+
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: Dict[str, Any],
+    ) -> Any:
+        if self._is_admin(event.chat.id):  # type: ignore
+            return await handler(event, data)

src/tg_bot/middlewares/logging.py

@@ -0,0 +1,18 @@
+from typing import Any, Awaitable, Callable, Dict
+
+import structlog
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject
+
+logger = structlog.get_logger()
+
+
+class LogCommandsMiddleware(BaseMiddleware):
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: Dict[str, Any],
+    ) -> Any:
+        logger.info(event.text, username=event.chat.username, chat_id=event.chat.id)  # type: ignore
+        return await handler(event, data)

src/tg_bot/middlewares/main.py

@@ -0,0 +1,14 @@
+from aiogram import Dispatcher
+from aiogram.utils.chat_action import ChatActionMiddleware
+
+from .error_handler import ErrorHandlerMiddleware
+from .is_admin import BlockNonAdminMiddleware
+from .logging import LogCommandsMiddleware
+
+
+def register_all_middlewares(dp: Dispatcher, dev_mode: bool, admin_id: int) -> None:
+    if dev_mode:
+        dp.message.middleware(BlockNonAdminMiddleware(admin_id))
+    dp.message.middleware(ChatActionMiddleware())
+    dp.message.middleware(LogCommandsMiddleware())
+    dp.message.middleware(ErrorHandlerMiddleware())