Did some refactor, add YouTube as search variant

bot/filters/__init__.py

@@ -0,0 +1 @@
+from .search import ServiceSearchFilter

bot/filters/search.py

@@ -0,0 +1,13 @@
+from aiogram.filters import BaseFilter
+from aiogram.types import InlineQuery
+
+
+class ServiceSearchFilter(BaseFilter):
+    def __init__(self, service_letter: str):
+        self.service_letter = f'{service_letter}:'
+
+    async def __call__(self, inline_query: InlineQuery):
+        return (
+                inline_query.query.startswith(self.service_letter) and
+                inline_query.query != self.service_letter
+        )

bot/handlers/__init__.py

@@ -1,6 +1,7 @@
 from aiogram import Router
 from . import (
     initialize,
+    inline_song,
     inline_default,
     inline_empty,
     on_chosen,
@@ -14,6 +15,7 @@ router.chosen_inline_result.outer_middleware(SaveChosenMiddleware())
 
 router.include_routers(
     initialize.router,
+    inline_song.router,
     inline_default.router,
     inline_empty.router,
     on_chosen.router,

bot/handlers/inline_default/on_inline_default.py

@@ -2,7 +2,7 @@ from aiogram import Router, F
 
 from aiogram.types import InlineQuery
 
-from bot.markups.deezer import get_deezer_search_results
+from bot.results.deezer import get_deezer_search_results
 
 router = Router()
 

bot/handlers/inline_song/__init__.py

@@ -0,0 +1,10 @@
+from aiogram import Router
+
+from . import on_inline_spotify, on_inline_deezer, on_inline_youtube
+
+router = Router()
+router.include_routers(
+    on_inline_spotify.router,
+    on_inline_deezer.router,
+    on_inline_youtube.router,
+)

bot/handlers/inline_song/on_inline_deezer.py

@@ -0,0 +1,17 @@
+from aiogram import Router
+
+from aiogram.types import InlineQuery
+
+from bot.results.deezer import get_deezer_search_results
+from bot.filters import ServiceSearchFilter
+
+router = Router()
+
+
+@router.inline_query(ServiceSearchFilter('d'))
+async def search_deezer_inline_query(inline_query: InlineQuery):
+    await inline_query.answer(
+        await get_deezer_search_results(inline_query.query.removeprefix('d:')),
+        cache_time=0,
+        is_personal=True
+    )

bot/handlers/inline_song/on_inline_spotify.py

@@ -0,0 +1,17 @@
+from aiogram import Router
+
+from aiogram.types import InlineQuery
+
+from bot.results.spotify import get_spotify_search_results
+from bot.filters import ServiceSearchFilter
+
+router = Router()
+
+
+@router.inline_query(ServiceSearchFilter('s'))
+async def search_spotify_inline_query(inline_query: InlineQuery):
+    await inline_query.answer(
+        await get_spotify_search_results(inline_query.query.removeprefix('s:')),
+        cache_time=0,
+        is_personal=True
+    )

bot/handlers/inline_song/on_inline_youtube.py

@@ -0,0 +1,17 @@
+from aiogram import Router
+
+from aiogram.types import InlineQuery
+
+from bot.results.youtube import get_youtube_search_results
+from bot.filters import ServiceSearchFilter
+
+router = Router()
+
+
+@router.inline_query(ServiceSearchFilter('y'))
+async def search_youtube_inline_query(inline_query: InlineQuery):
+    await inline_query.answer(
+        await get_youtube_search_results(inline_query.query.removeprefix('y:')),
+        cache_time=0,
+        is_personal=True
+    )

bot/handlers/on_chosen/__init__.py

@@ -1,11 +1,12 @@
 from aiogram import Router
-from . import spotify, deezer
+from . import spotify, deezer, youtube
 
 router = Router()
 
 router.include_routers(
     spotify.router,
     deezer.router,
+    youtube.router,
 )
 
 __all__ = ['router']

bot/handlers/on_chosen/youtube.py

@@ -0,0 +1,49 @@
+from aiogram import Router, Bot, F
+from aiogram.types import (
+    BufferedInputFile, URLInputFile, InputMediaAudio,
+    ChosenInlineResult,
+)
+
+from bot.modules.youtube import youtube, AgeRestrictedError
+from bot.utils.config import config
+from bot.modules.database import db
+
+router = Router()
+
+
+@router.chosen_inline_result(F.result_id.startswith('yt::'))
+async def on_new_chosen(chosen_result: ChosenInlineResult, bot: Bot):
+    song = youtube.songs.from_id(chosen_result.result_id.removeprefix('yt::'))
+
+    try:
+        bytestream = await song.to_bytestream()
+    except AgeRestrictedError:
+        await bot.edit_message_caption(
+            inline_message_id=chosen_result.inline_message_id,
+            caption='🔞 This song is age restricted, so I can\'t download it. '
+                    'Try downloading it from Deezer or SoundCloud',
+            reply_markup=None
+        )
+        return
+
+    audio = await bot.send_audio(
+        chat_id=config.telegram.files_chat,
+        audio=BufferedInputFile(
+            file=bytestream.file,
+            filename=bytestream.filename,
+        ),
+        thumbnail=URLInputFile(song.thumbnail),
+        performer=song.all_artists,
+        title=song.name,
+        duration=bytestream.duration,
+    )
+
+    db.spotify[song.id] = audio.audio.file_id
+
+    await bot.edit_message_media(
+        inline_message_id=chosen_result.inline_message_id,
+        media=InputMediaAudio(media=audio.audio.file_id),
+        reply_markup=None
+    )
+
+    await db.occasionally_write()

bot/markups/spotify/search.py

@@ -1,33 +0,0 @@
-from aiogram.types import (
-    InlineQueryResultDocument, InlineQueryResultCachedAudio,
-    InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResult
-)
-
-from bot.modules.spotify import spotify
-from bot.modules.database import db
-
-
-async def get_spotify_search_results(query: str) -> list[
-    InlineQueryResultDocument | InlineQueryResultCachedAudio
-]:
-    return [
-        InlineQueryResultDocument(
-            id='spot::' + audio.id,
-            title=audio.name,
-            description=audio.all_artists,
-            thumb_url=audio.thumbnail,
-            document_url=audio.preview_url or audio.thumbnail,
-            mime_type='application/zip',
-            reply_markup=InlineKeyboardMarkup(
-                inline_keyboard=[
-                    [InlineKeyboardButton(text='Downloading...', callback_data='.')]
-                ]
-            ),
-            caption=audio.full_name,
-        ) if audio.id not in list(db.spotify.keys()) else
-        InlineQueryResultCachedAudio(
-            id='spotc::' + audio.id,
-            audio_file_id=db.spotify[audio.id],
-        )
-        for audio in spotify.songs.search(query, limit=50)
-    ]

bot/modules/common/__init__.py

@@ -0,0 +1 @@
+pass

bot/modules/common/song/__init__.py

@@ -0,0 +1 @@
+from .song import BaseSongItem

bot/modules/common/song/song.py

@@ -0,0 +1,21 @@
+from attrs import define
+
+
+@define
+class BaseSongItem:
+    name: str
+    id: str
+    artists: list[str]
+    preview_url: str | None
+    thumbnail: str
+
+    @property
+    def all_artists(self):
+        return ', '.join(self.artists)
+
+    @property
+    def full_name(self):
+        return f"{self.all_artists} - {self.name}"
+
+    def __str__(self):
+        return self.full_name

bot/modules/database/db.py

@@ -17,6 +17,7 @@ class Db(object):
         self.inline = DBDict('inline')
         self.spotify = DBDict('spotify')
         self.deezer = DBDict('deezer')
+        self.youtube = DBDict('youtube')
 
     async def write(self):
         await self.config.write()

bot/modules/deezer/song.py

@@ -2,43 +2,25 @@ from attrs import define
 
 from .driver import DeezerDriver
 
+from ..common.song import BaseSongItem
+
 
 @define
-class SongItem:
-    name: str
-    id: int
-    id_s: str
-    artist: str
-    preview_url: str | None
-    thumbnail: str
-
+class SongItem(BaseSongItem):
     @classmethod
     def from_deezer(cls, song_item: dict):
         return cls(
             name=song_item['title'],
-            id=song_item['id'],
-            id_s=str(song_item['id']),
-            artist=song_item['artist']['name'],
+            id=str(song_item['id']),
+            artists=[song_item['artist']['name']],
             preview_url=song_item.get('preview'),
             thumbnail=song_item['album']['cover_medium']
         )
 
-    @property
-    def full_name(self):
-        return f"{self.artist} - {self.name}"
-
-    def __str__(self):
-        return self.full_name
-
 
 @define
-class FullSongItem:
-    name: str
-    id: str
-    artists: list[str]
-    preview_url: str | None
+class FullSongItem(BaseSongItem):
     duration: int
-    thumbnail: str
     track_dict: dict
 
     @classmethod
@@ -60,17 +42,6 @@ class FullSongItem:
             track_dict=song_item
         )
 
-    @property
-    def all_artists(self):
-        return ', '.join(self.artists)
-
-    @property
-    def full_name(self):
-        return f"{self.all_artists} - {self.name}"
-
-    def __str__(self):
-        return self.full_name
-
 
 @define
 class Songs(object):

bot/modules/spotify/song.py

@@ -1,15 +1,11 @@
 from attrs import define
 import spotipy
 
+from ..common.song import BaseSongItem
+
 
 @define
-class SongItem:
-    name: str
-    id: str
-    artists: list[str]
-    preview_url: str | None
-    thumbnail: str
-
+class SongItem(BaseSongItem):
     @classmethod
     def from_spotify(cls, song_item: dict):
         return cls(
@@ -21,17 +17,6 @@ class SongItem:
             thumbnail=song_item['album']['images'][1]['url']
         )
 
-    @property
-    def all_artists(self):
-        return ', '.join(self.artists)
-
-    @property
-    def full_name(self):
-        return f"{self.all_artists} - {self.name}"
-
-    def __str__(self):
-        return f"{', '.join(self.artists)} - {self.name}"
-
 
 @define
 class Songs(object):

bot/modules/youtube/song.py

@@ -5,13 +5,12 @@ from .downloader import Downloader, YouTubeBytestream
 
 from typing import Awaitable
 
+from ..common.song import BaseSongItem
+
 
 @define
-class SongItem:
-    name: str
-    id: str
-    artists: list[str]
-    thumbnail: str
+class SongItem(BaseSongItem):
+    preview_url: None = None
 
     @classmethod
     def from_youtube(cls, song_item: dict):
@@ -22,16 +21,14 @@ class SongItem:
             thumbnail=song_item['thumbnails'][1]['url']
         )
 
-    @property
-    def all_artists(self):
-        return ', '.join(self.artists)
-
-    @property
-    def full_name(self):
-        return f"{self.all_artists} - {self.name}"
-
-    def __str__(self):
-        return self.full_name
+    @classmethod
+    def from_details(cls, details: dict):
+        return cls(
+            name=details['title'],
+            id=details['videoId'],
+            artists=details['author'].split(' & '),
+            thumbnail=details['thumbnail']['thumbnails'][1]['url']
+        )
 
     def to_bytestream(self) -> Awaitable[YouTubeBytestream]:
         return Downloader.from_id(self.id).to_bytestream()
@@ -58,4 +55,4 @@ class Songs(object):
         if r is None:
             return None
 
-        return SongItem.from_youtube(r)
+        return SongItem.from_details(r['videoDetails'])

bot/results/__init__.py

@@ -0,0 +1 @@
+pass

bot/results/common/__init__.py

@@ -0,0 +1 @@
+pass

bot/results/common/search.py

@@ -3,18 +3,25 @@ from aiogram.types import (
     InlineKeyboardMarkup, InlineKeyboardButton
 )
 
-from bot.modules.deezer import deezer
-from bot.modules.database import db
+from bot.modules.database.db import DBDict
+
+from bot.modules.common.song import BaseSongItem
+from typing import TypeVar
 
 
-async def get_deezer_search_results(query: str) -> list[
-    InlineQueryResultDocument | InlineQueryResultCachedAudio
-]:
-    return [
+BaseSongT = TypeVar('BaseSongT', bound=BaseSongItem)
+
+
+async def get_common_search_result(
+        audio: BaseSongT,
+        db_table: DBDict,
+        service_id: str
+) -> InlineQueryResultDocument | InlineQueryResultCachedAudio:
+    return (
         InlineQueryResultDocument(
-            id='deez::' + audio.id_s,
+            id=f'{service_id}::' + audio.id,
             title=audio.name,
-            description=audio.artist,
+            description=audio.all_artists,
             thumb_url=audio.thumbnail,
             document_url=audio.preview_url or audio.thumbnail,
             mime_type='application/zip',
@@ -24,10 +31,9 @@ async def get_deezer_search_results(query: str) -> list[
                 ]
             ),
             caption=audio.full_name,
-        ) if audio.id_s not in list(db.deezer.keys()) else
+        ) if audio.id not in list(db_table.keys()) else
         InlineQueryResultCachedAudio(
-            id='deezc::' + audio.id_s,
-            audio_file_id=db.deezer[audio.id_s],
+            id=f'{service_id}c::' + audio.id,
+            audio_file_id=db_table[audio.id],
         )
-        for audio in await deezer.songs.search(query, limit=50)
-    ]
+    )

bot/results/deezer/search.py

@@ -0,0 +1,21 @@
+from aiogram.types import (
+    InlineQueryResultDocument, InlineQueryResultCachedAudio
+)
+
+from bot.modules.deezer import deezer
+from bot.modules.database import db
+
+from ..common.search import get_common_search_result
+
+
+async def get_deezer_search_results(query: str) -> list[
+    InlineQueryResultDocument | InlineQueryResultCachedAudio
+]:
+    return [
+        await get_common_search_result(
+            audio=audio,
+            db_table=db.deezer,
+            service_id='deez'
+        )
+        for audio in await deezer.songs.search(query, limit=50)
+    ]

bot/results/spotify/search.py

@@ -0,0 +1,21 @@
+from aiogram.types import (
+    InlineQueryResultDocument, InlineQueryResultCachedAudio
+)
+
+from bot.modules.spotify import spotify
+from bot.modules.database import db
+
+from ..common.search import get_common_search_result
+
+
+async def get_spotify_search_results(query: str) -> list[
+    InlineQueryResultDocument | InlineQueryResultCachedAudio
+]:
+    return [
+        await get_common_search_result(
+            audio=audio,
+            db_table=db.spotify,
+            service_id='spot'
+        )
+        for audio in spotify.songs.search(query, limit=50)
+    ]

bot/results/youtube/__init__.py

@@ -0,0 +1,6 @@
+from .search import get_youtube_search_results
+
+
+__all__ = [
+    'get_youtube_search_results'
+]

bot/results/youtube/search.py

@@ -0,0 +1,21 @@
+from aiogram.types import (
+    InlineQueryResultDocument, InlineQueryResultCachedAudio
+)
+
+from bot.modules.youtube import youtube
+from bot.modules.database import db
+
+from ..common.search import get_common_search_result
+
+
+async def get_youtube_search_results(query: str) -> list[
+    InlineQueryResultDocument | InlineQueryResultCachedAudio
+]:
+    return [
+        await get_common_search_result(
+            audio=audio,
+            db_table=db.youtube,
+            service_id='yt'
+        )
+        for audio in youtube.songs.search(query, limit=40)
+    ]