Refactoring

Author: BushlanovDev

bot/configs/config.py

@@ -21,7 +21,7 @@ class Settings(BaseSettings):
     tg_bot: TgBot
     logging: dict[str, Any] = {
         'version': 1,
-        'disable_existing_loggers': True,
+        'disable_existing_loggers': False,
         'formatters': {
             'default': {
                 'format': '%(levelname)-8s [%(asctime)s] [%(name)s] %(message)s',

bot/handlers/handlers.py

@@ -14,40 +14,47 @@ def __init__(self, bot: Bot):
 
     async def start_command(self, message: Message, ctx: HandlerContext, user_lang: str | None):
         await message.answer(
-            text=ctx.lexicon.get_text('Hi user! This start message.', user_lang),
+            text=ctx.lexicon.get_text('start_message', user_lang),
             reply_markup=ctx.keyboards.get_start_button(ctx.lexicon, user_lang),
         )
 
     async def help_command(self, message: Message, ctx: HandlerContext, user_lang: str | None):
-        await message.answer(text=ctx.lexicon.get_text('This help message.', user_lang))
+        await message.answer(text=ctx.lexicon.get_text('help_message', user_lang))
 
-    async def answer(self, message: Message, answer: str):
-        await self.__send_message(message.chat.id, answer, message.message_id)
+    async def answer(self, message: Message, ctx: HandlerContext, user_lang: str | None):
+        text = ctx.lexicon.get_text('default_answer', user_lang)
+        await self.__send_message(message.chat.id, text, message.message_id)
 
-    async def answer_fsm(self, message: Message, state: FSMContext):
-        await message.answer('Waiting Step 1')
+    async def answer_fsm(self, message: Message, state: FSMContext, ctx: HandlerContext, user_lang: str | None):
+        await message.answer(ctx.lexicon.get_text('fsm_wait_step_1', user_lang))
         await state.set_state(BotState.waiting_step_1)
 
-    async def answer_fsm_state_1(self, message: Message, state: FSMContext):
+    async def answer_fsm_state_1(self, message: Message, state: FSMContext, ctx: HandlerContext, user_lang: str | None):
         await state.update_data(step_1=message.text)
-        await message.answer('Waiting Step 2')
+        await message.answer(ctx.lexicon.get_text('fsm_wait_step_2', user_lang))
         await state.set_state(BotState.waiting_step_2)
 
-    async def answer_fsm_state_2(self, message: Message, state: FSMContext):
+    async def answer_fsm_state_2(self, message: Message, state: FSMContext, ctx: HandlerContext, user_lang: str | None):
         await state.update_data(step_2=message.text)
         data = await state.get_data()
-        await message.answer(f'Step 1: {data["step_1"]}\nStep 2: {data["step_2"]}')
+        await message.answer(
+            ctx.lexicon.get_text('fsm_result', user_lang).format(
+                step_1=data.get('step_1', ''),
+                step_2=data.get('step_2', ''),
+            )
+        )
         await state.clear()
 
     async def answer_inline_button(self, message: Message, ctx: HandlerContext, user_lang: str | None):
         callback = SaveCallbackFactory(message_id=message.message_id).pack()
         await message.answer(
-            text=ctx.lexicon.get_text('Inline button message'),
+            text=ctx.lexicon.get_text('inline_prompt', user_lang),
             reply_markup=ctx.keyboards.get_inline_button(ctx.lexicon, user_lang, callback),
         )
 
     async def reply(self, message: Message, ctx: HandlerContext):
-        await message.reply(text=ctx.bot_service.upper(message.text))
+        text = message.text or ''
+        await message.reply(text=ctx.bot_service.upper(text))
 
     async def process_any_inline_button_press(self, callback: CallbackQuery, callback_data: SaveCallbackFactory):
         await callback.message.answer(text=callback_data.pack())

bot/i18n/en.py

@@ -1 +1,15 @@
-LEXICON_EN: dict[str, str] = {}
+LEXICON_EN: dict[str, str] = {
+    'start_message': 'Hi user! This is the start message.',
+    'help_message': 'This is the help message.',
+    'start_button_hi': 'Hi',
+    'start_button_test': 'Test',
+    'start_button_inline': 'Inline button',
+    'start_button_fsm': 'FSM',
+    'inline_prompt': 'Here is an inline button.',
+    'inline_save_button': 'Save',
+    'fsm_wait_step_1': 'Waiting for step 1',
+    'fsm_wait_step_2': 'Waiting for step 2',
+    'fsm_result': 'Step 1: {step_1}\nStep 2: {step_2}',
+    'default_answer': 'Hello',
+    'throttled_message': 'Too many requests. Please slow down.',
+}

bot/i18n/lexicon.py

@@ -1,21 +1,31 @@
+from collections.abc import Mapping
+
 from bot.i18n.en import LEXICON_EN
 from bot.i18n.ru import LEXICON_RU
 
 
 class Lexicon:
     def __init__(self, default_language: str):
         self.default_language = default_language
+        self._lexicons: dict[str, Mapping[str, str]] = {
+            'en': LEXICON_EN,
+            'ru': LEXICON_RU,
+        }
+
+    def get_text(self, key: str, lang: str | None = None) -> str:
+        language = lang or self.default_language
+        translation = self._lexicons.get(language, {})
+
+        if key in translation:
+            return translation[key]
 
-    def get_text(self, param: str, lang: str | None = None) -> str:
-        if lang is None:
-            lang = self.default_language
+        default_translation = self._lexicons.get(self.default_language, {})
+        if key in default_translation:
+            return default_translation[key]
 
-        match lang:
-            case 'ru':
-                ret = LEXICON_RU.get(param, param)
-            case 'en':
-                ret = LEXICON_EN.get(param, param)
-            case _:
-                ret = LEXICON_EN.get(param, param)
+        fallback_translation = self._lexicons.get('en', {})
+        return fallback_translation.get(key, key)
 
-        return ret
+    def get_translations(self, key: str) -> set[str]:
+        values = {translations[key] for translations in self._lexicons.values() if key in translations}
+        return values or {key}

bot/i18n/ru.py

@@ -1,6 +1,15 @@
 LEXICON_RU: dict[str, str] = {
-    'hi': 'привет',
-    'Save': 'Сохранить',
-    'Hi user! This start message.': 'Привет, пользователь! Это стартовое сообщение.',
-    'This help message.': 'Это справочное сообщение.',
+    'start_message': 'Привет, пользователь! Это стартовое сообщение.',
+    'help_message': 'Это справочное сообщение.',
+    'start_button_hi': 'Привет',
+    'start_button_test': 'Тест',
+    'start_button_inline': 'Инлайн кнопка',
+    'start_button_fsm': 'FSM',
+    'inline_prompt': 'Сообщение с инлайн-кнопкой.',
+    'inline_save_button': 'Сохранить',
+    'fsm_wait_step_1': 'Ожидаю шаг 1',
+    'fsm_wait_step_2': 'Ожидаю шаг 2',
+    'fsm_result': 'Шаг 1: {step_1}\nШаг 2: {step_2}',
+    'default_answer': 'Привет',
+    'throttled_message': 'Слишком много запросов. Пожалуйста, снизьте частоту.',
 }

bot/keyboards/keyboards.py

@@ -9,8 +9,14 @@ def __init__(self):
 
     def get_start_button(self, lexicon: Lexicon, user_lang: str | None) -> ReplyKeyboardMarkup:
         kb_list = [
-            [KeyboardButton(text=lexicon.get_text('hi', user_lang)), KeyboardButton(text='test')],
-            [KeyboardButton(text='inline button'), KeyboardButton(text='fsm')],
+            [
+                KeyboardButton(text=lexicon.get_text('start_button_hi', user_lang)),
+                KeyboardButton(text=lexicon.get_text('start_button_test', user_lang)),
+            ],
+            [
+                KeyboardButton(text=lexicon.get_text('start_button_inline', user_lang)),
+                KeyboardButton(text=lexicon.get_text('start_button_fsm', user_lang)),
+            ],
         ]
 
         return ReplyKeyboardMarkup(keyboard=kb_list, resize_keyboard=True, one_time_keyboard=True)
@@ -20,7 +26,10 @@ def get_inline_button(
     ) -> InlineKeyboardMarkup:
         inline_kb_list = [
             [
-                InlineKeyboardButton(text=lexicon.get_text('Save', user_lang), callback_data=callback_data),
+                InlineKeyboardButton(
+                    text=lexicon.get_text('inline_save_button', user_lang),
+                    callback_data=callback_data,
+                ),
             ],
         ]
 

bot/middlewares/throttling.py

@@ -3,12 +3,12 @@
 from typing import Any
 
 from aiogram import BaseMiddleware
-from aiogram.types import TelegramObject, User
+from aiogram.types import CallbackQuery, Message, TelegramObject, Update, User
 from cachetools import TTLCache
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger('main')
 
-CACHE = TTLCache(maxsize=10_000, ttl=2)  # Максимальный размер кэша - 10000 ключей, время жизни ключа - 2 секунды
+CACHE = TTLCache(maxsize=10_000, ttl=2)
 
 
 class ThrottlingMiddleware(BaseMiddleware):
@@ -23,8 +23,37 @@ async def __call__(
         if user is not None:
             if user.id in CACHE:
                 logger.debug(f'Throttling for user {user.id}')
+                text = 'Too many requests. Please slow down.'
+                lexicon = data.get('lexicon')
+                if lexicon is not None:
+                    text = lexicon.get_text('throttled_message', data.get('user_lang'))
+                await self._notify_throttled(event, text)
                 return
 
             CACHE[user.id] = True
 
         return await handler(event, data)
+
+    @staticmethod
+    async def _notify_throttled(event: TelegramObject, text: str) -> None:
+        if isinstance(event, Message):
+            await event.answer(text)
+            return
+
+        if isinstance(event, CallbackQuery):
+            if event.message is not None:
+                await event.message.answer(text)
+            await event.answer()
+            return
+
+        if isinstance(event, Update):
+            if event.message is not None:
+                await event.message.answer(text)
+                return
+            if event.callback_query is not None:
+                if event.callback_query.message is not None:
+                    await event.callback_query.message.answer(text)
+                await event.callback_query.answer()
+                return
+
+        logger.debug('Throttling notification skipped: unsupported event type %s', type(event))

main.py

@@ -1,4 +1,5 @@
 import logging
+import logging.config
 from typing import Callable, Optional
 
 import uvicorn
@@ -36,11 +37,7 @@ async def startup_handler() -> None:
                 drop_pending_updates=True,
             )
 
-        await bot.set_my_commands(
-            [
-                BotCommand(command='/help', description='Help'),
-            ]
-        )
+        await bot.set_my_commands([BotCommand(command='/help', description='Help')])
 
         logger.debug('startup_handler')
 
@@ -78,15 +75,18 @@ def create_dispatcher(storage: Optional[BaseStorage] = None) -> Dispatcher:
     return Dispatcher(storage=storage)
 
 
-def register_handlers(dp: Dispatcher, hd: Handlers) -> None:
+def register_handlers(dp: Dispatcher, hd: Handlers, lexicon: Lexicon) -> None:
     dp.message.register(hd.start_command, CommandStart())
     dp.message.register(hd.help_command, Command(commands='help'))
-    dp.message.register(hd.answer, F.text == 'hi')
-    dp.message.register(hd.answer_inline_button, F.text == 'inline button')
-    dp.message.register(hd.answer_fsm, F.text == 'fsm')
+    dp.message.register(hd.answer, F.text.in_(tuple(sorted(lexicon.get_translations('start_button_hi')))))
+    dp.message.register(
+        hd.answer_inline_button,
+        F.text.in_(tuple(sorted(lexicon.get_translations('start_button_inline')))),
+    )
+    dp.message.register(hd.answer_fsm, F.text.in_(tuple(sorted(lexicon.get_translations('start_button_fsm')))))
     dp.message.register(hd.answer_fsm_state_1, BotState.waiting_step_1)
     dp.message.register(hd.answer_fsm_state_2, BotState.waiting_step_2)
-    dp.message.register(hd.reply)
+    dp.message.register(hd.reply, F.text)
     dp.callback_query.register(hd.process_any_inline_button_press, SaveCallbackFactory.filter())
 
 
@@ -96,16 +96,12 @@ def register_middlewares(dp: Dispatcher) -> None:
     dp.update.outer_middleware(ContextMiddleware())
 
 
-def register_workflow_data(dp: Dispatcher, settings: Settings) -> None:
-    dp.workflow_data.update({'answer': 'Hello', 'lexicon': Lexicon(settings.app.default_language)})
-    dp.workflow_data.update(
-        {
-            'answer': 'Hello',
-            'lexicon': Lexicon(settings.app.default_language),
-            'keyboards': Keyboards(),
-            'bot_service': BotService(),
-        }
-    )
+def register_workflow_data(dp: Dispatcher, lexicon: Lexicon) -> None:
+    dp.workflow_data.update({
+        'lexicon': lexicon,
+        'keyboards': Keyboards(),
+        'bot_service': BotService(),
+    })
 
 
 def create_app(bot: Bot, dp: Dispatcher, settings: App) -> FastAPI:
@@ -124,9 +120,10 @@ def main() -> None:
 
     bot = create_bot(settings.tg_bot)
     dp = create_dispatcher(MemoryStorage())
+    lexicon = Lexicon(settings.app.default_language)
     app = create_app(bot, dp, settings.app)
-    register_handlers(dp, Handlers(bot))
-    register_workflow_data(dp, settings)
+    register_workflow_data(dp, lexicon)
+    register_handlers(dp, Handlers(bot), lexicon)
     register_middlewares(dp)
 
     logger.debug('Application is running')