feat: migrate to another music search service

README.md

@@ -15,7 +15,6 @@
 
 - 🎵 Поиск и прослушивание музыки
 - 🎧 Высокое качество звука
-- 🌅 Аудио с обложкой
 - 📱 Интуитивный интерфейс
 - 🌍 Поддержка нескольких языков (🇬🇧 English, 🇷🇺 Русский)
 - 🌍 Автоопределение языка пользователя

bot/__init__.py

@@ -0,0 +1 @@
+"""Bot package."""

bot/filters/__init__.py

@@ -1,6 +1,7 @@
 """Filters for the bot."""
+
+from .is_private_chat import IsPrivateChatFilter
 from .language import LanguageFilter
 from .not_subbed import NotSubbedFilter
 
-
-__all__ = ['LanguageFilter', 'NotSubbedFilter']
+__all__ = ["IsPrivateChatFilter", "LanguageFilter", "NotSubbedFilter"]

bot/filters/is_private_chat.py

@@ -0,0 +1,25 @@
+"""Filter for checking if the chat is private."""
+
+from __future__ import annotations
+
+from aiogram.enums import ChatType
+from aiogram.filters import Filter
+from aiogram.types import CallbackQuery, Message
+
+
+class IsPrivateChatFilter(Filter):
+    """Filter for checking if the chat is private."""
+
+    async def __call__(
+        self,
+        event: Message | CallbackQuery,
+    ) -> bool:
+        """Check if the chat is private."""
+        if isinstance(event, CallbackQuery):
+            message = event.message
+            if not isinstance(message, Message):
+                return False
+        else:
+            message = event
+
+        return message.chat.type == ChatType.PRIVATE

bot/filters/language.py

@@ -1,20 +1,30 @@
 """Check language filter."""
 
-from aiogram import types
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from aiogram.filters import Filter
-from database.models import User
+
 from locales import support_languages
 
+if TYPE_CHECKING:
+    from aiogram import types
+
+    from database.models import User
+
 
 class LanguageFilter(Filter):
-    """
-    A filter that checks if a user's language matches the specified language.
-    """
+    """checks if user's language matches the specified language."""
 
     async def __call__(
-        self, update: types.Message | types.CallbackQuery, user: User
+        self,
+        update: types.Message | types.CallbackQuery,  # noqa: ARG002
+        user: User,
     ) -> bool:
-        """
-        Check if the user's language matches the specified language.
-        """
-        return not support_languages.is_supported(user.language_code)
+        """Check if the user's language matches the specified language."""
+        language_code = user.language_code
+        if language_code is None:
+            return True
+
+        return not support_languages.is_supported(language_code)

bot/filters/not_subbed.py

@@ -1,35 +1,41 @@
 """Check if user is subbed to the channel."""
+
+from __future__ import annotations
+
 import logging
-from aiogram import types
-from aiogram.filters import Filter
-from aiogram import Bot
+from typing import TYPE_CHECKING, ClassVar
+
 from aiogram.enums import ChatMemberStatus
-from database.models import RequiredSubscriptions, User
+from aiogram.filters import Filter
+
 from database.crud import CRUD
+from database.models import RequiredSubscriptions, User
+
+if TYPE_CHECKING:
+    from aiogram import Bot, types
 
 logger = logging.getLogger(__name__)
 
 
 class NotSubbedFilter(Filter):
-    """
-    A filter that checks if a user is subbed to the channel.
-    """
+    """A filter that checks if a user is subbed to the channel."""
 
-    ALLOWED_STATUSES = {
+    ALLOWED_STATUSES: ClassVar[set[ChatMemberStatus]] = {
         ChatMemberStatus.MEMBER,
         ChatMemberStatus.ADMINISTRATOR,
         ChatMemberStatus.CREATOR,
     }
 
     async def __call__(
-        self, update: types.Message | types.CallbackQuery, user: User, bot: Bot
+        self,
+        update: types.Message | types.CallbackQuery,  # noqa: ARG002
+        user: User,
+        bot: Bot,
     ) -> bool:
-        """
-        Check if the user is not subscribed to any required channels.
-        Returns True if user is NOT subscribed to ANY required channel.
-        Returns False if user is subscribed to ALL required channels.
-        """
-        chats = await CRUD(RequiredSubscriptions).get_all()
+        """Check if the user is not subscribed to any required channels."""
+        chats: list[RequiredSubscriptions] = await CRUD(
+            RequiredSubscriptions,
+        ).get_all()
 
         if not chats:
             return False
@@ -40,21 +46,21 @@ async def __call__(
         return False
 
     async def _not_subscribe(
-        self, sub: RequiredSubscriptions, user: User, bot: Bot
+        self,
+        sub: RequiredSubscriptions,
+        user: User,
+        bot: Bot,
     ) -> bool:
-        """
-        Check if the user is subscribed to the channel.
-        Returns True if user is NOT subscribed.
-        Returns False if user is subscribed or if channel is not accessible.
-        """
+        """Check if the user is subscribed to the channel."""
         try:
             chat = await bot.get_chat(sub.chat_id)
-        except Exception as e:
-            logger.error(f"Failed to get chat {sub.chat_id}: {e}")
+        except Exception:
+            logger.exception("Failed to get chat %s", sub.chat_id)
             return False
 
         try:
             member = await chat.get_member(user.id)
-            return member.status not in self.ALLOWED_STATUSES
-        except Exception:
+        except Exception:  # noqa: BLE001
             return True
+
+        return member.status not in self.ALLOWED_STATUSES

bot/handlers/__init__.py

@@ -1,27 +1,27 @@
-"""Setup router for the bot."""
+"""Set up handlers for the bot."""
+
 from aiogram import Dispatcher, Router
-from . import (
-    get_track,
-    language,
-    menu,
-    faq,
-    search,
-    pages,
-    subscribe
-)
+
+from bot.filters import IsPrivateChatFilter
+
+from . import faq, get_track, language, menu, pages, search, subscribe
 
 
 def setup(dp: Dispatcher) -> None:
-    """Setup handlers for the bot."""
+    """Set up handlers for the bot."""
     router = Router()
 
+    # Add filters
+    router.message.filter(IsPrivateChatFilter())
+    router.callback_query.filter(IsPrivateChatFilter())
+
     # Register routers
     language.register(router)
     menu.register(router)
     faq.register(router)
-    subscribe.register(router)
     search.register(router)
     pages.register(router)
+    subscribe.register(router)
     get_track.register(router)
 
     dp.include_router(router)

bot/handlers/faq.py

@@ -1,7 +1,10 @@
 """FAQ handler for the bot."""
+
 import logging
-from aiogram import types, Router, F
+
+from aiogram import F, Router, types
 from aiogram.utils.i18n import gettext
+
 from bot.keyboards import inline
 
 logger = logging.getLogger(__name__)
@@ -10,14 +13,21 @@
 async def faq_handler(callback: types.CallbackQuery) -> None:
     """FAQ handler."""
     try:
+        if callback.message is None or isinstance(
+            callback.message,
+            types.InaccessibleMessage,
+        ):
+            await callback.answer(gettext("cannot_edit_message"))
+            return
+
         await callback.message.edit_text(
             gettext("faq"),
-            reply_markup=inline.get_back_keyboard(gettext, "menu")
+            reply_markup=inline.get_back_keyboard(gettext, "menu"),
         )
-    except Exception as e:
-        logger.error("Failed to send message: %s", e)
+    except Exception:
+        logger.exception("Failed to send message")
 
 
 def register(router: Router) -> None:
-    """Registers FAQ handler with the router."""
+    """Register FAQ handler with the router."""
     router.callback_query.register(faq_handler, F.data == "faq")

bot/handlers/get_track.py

@@ -1,79 +1,102 @@
 """Get track handler for the bot."""
+
 import logging
-from aiogram import types, Router, F, Bot
+
+from aiogram import Bot, F, Router, types
 from aiogram.types import BufferedInputFile
 from aiogram.utils.i18n import gettext
-from service import Music, Track
-from bot.utils import load_tracks_from_db
 
+from bot.utils import load_tracks_from_db
+from service import Music, Track
 
 logger = logging.getLogger(__name__)
 
 
 async def send_track(
     callback: types.CallbackQuery,
     bot: Bot,
-    track: Track
+    track: Track,
 ) -> None:
     """Send track."""
     try:
+        if callback.message is None or callback.message.chat is None:
+            await callback.answer(gettext("cannot_access_chat"))
+            return
+
         await bot.send_chat_action(callback.message.chat.id, "upload_document")
 
         async with Music() as service:
             audio_bytes = await service.get_audio_bytes(track)
-            thumbnail_bytes = await service.get_thumbnail_bytes(track)
 
             audio_file = BufferedInputFile(audio_bytes, filename=track.name)
-            thumbnail_file = BufferedInputFile(
-                thumbnail_bytes, filename=track.name
-            )
+
+        me = await bot.get_me()
+
+        if callback.message is None:
+            await callback.answer("Cannot send message")
+            return
 
         await callback.message.answer_audio(
             audio_file,
             title=track.title,
             performer=track.performer,
-            caption=gettext("promo_caption").format(username=bot._me.username),
-            thumbnail=thumbnail_file,
+            caption=gettext("promo_caption").format(username=me.username),
         )
-    except Exception as e:
-        await callback.message.answer(gettext("send_track_error"))
-        logger.error("Failed to send track: %s", e)
+    except Exception:
+        if callback.message is not None:
+            await callback.message.answer(gettext("send_track_error"))
+        else:
+            await callback.answer(gettext("send_track_error"))
+        logger.exception("Failed to send track")
 
 
 async def get_track_handler(callback: types.CallbackQuery, bot: Bot) -> None:
     """Get track handler."""
     try:
-        _, _, search_id, index = callback.data.split(":")
+        if callback.data is None:
+            await callback.answer(gettext("invalid_data"))
+            return
+
+        _, _, search_id_str, index = callback.data.split(":")
+        search_id = int(search_id_str)
         tracks: list[Track] = await load_tracks_from_db(search_id)
         track: Track = tracks[int(index)]
         await callback.answer(gettext("track_sending"))
         await send_track(callback, bot, track)
 
-    except Exception as e:
-        logger.error("Failed get track handler: %s", e)
+    except Exception:
+        logger.exception("Failed get track handler")
 
 
 async def get_all_from_page_handler(
-    callback: types.CallbackQuery, bot: Bot
+    callback: types.CallbackQuery,
+    bot: Bot,
 ) -> None:
     """Get all tracks from page handler."""
     try:
-        _, _, search_id, start_indx, end_indx = callback.data.split(":")
+        if callback.data is None:
+            await callback.answer(gettext("invalid_data"))
+            return
+
+        _, _, search_id_str, start_indx, end_indx = callback.data.split(":")
+        search_id = int(search_id_str)
         all_tracks: list[Track] = await load_tracks_from_db(search_id)
-        page_tracks = all_tracks[int(start_indx):int(end_indx)]
+        page_tracks = all_tracks[int(start_indx) : int(end_indx)]
         await callback.answer(gettext("track_sending"))
         for track in page_tracks:
             await send_track(callback, bot, track)
 
-    except Exception as e:
-        logger.error("Failed get all from page handler: %s", e)
+    except Exception:
+        logger.exception("Failed get all from page handler")
 
 
 def register(router: Router) -> None:
-    """Registers get track handler with the router."""
+    """Register get track handler with the router."""
     router.callback_query.register(
-        get_track_handler, F.data.startswith("track:get:")
+        get_track_handler,
+        F.data.startswith("track:get:"),
     )
     router.callback_query.register(
-        get_all_from_page_handler, F.data.startswith("track:all:")
+        get_all_from_page_handler,
+        F.data.startswith("track:all:"),
     )