Add downloading song by query

bot/handlers/on_chosen/spotify.py

@@ -1,11 +1,11 @@
 from aiogram import Router, Bot, F
 from aiogram.types import (
-    FSInputFile, URLInputFile, InputMediaAudio,
+    BufferedInputFile, URLInputFile, InputMediaAudio,
     ChosenInlineResult,
 )
 
 from bot.modules.spotify import spotify
-from rich import print
+from bot.modules.youtube import youtube
 from bot.utils.config import config
 from bot.modules.database import db
 
@@ -14,22 +14,27 @@ router = Router()
 
 @router.chosen_inline_result(F.result_id.startswith('spot::'))
 async def on_new_chosen(chosen_result: ChosenInlineResult, bot: Bot):
-    print('TEST: DOWNLOADING NEW')
     song = spotify.songs.from_id(chosen_result.result_id.removeprefix('spot::'))
 
+    bytestream = youtube.songs.search_one(song.full_name).to_bytestream()
+
     audio = await bot.send_audio(
         chat_id=config.telegram.files_chat,
-        audio=FSInputFile('tests/test.mp3'),
+        audio=BufferedInputFile(
+            file=bytestream.file,
+            filename=bytestream.filename,
+        ),
         thumbnail=URLInputFile(song.thumbnail),
         performer=song.all_artists,
-        title=song.name,
+        title=song.name
     )
 
     db.spotify[song.id] = audio.audio.file_id
-    await db.occasionally_write()
 
     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/modules/youtube/__init__.py

@@ -0,0 +1,7 @@
+from .youtube import YouTube
+
+
+youtube = YouTube()
+
+
+__all__ = ['youtube']

bot/modules/youtube/downloader.py

@@ -0,0 +1,59 @@
+from attrs import define
+from pytube import YouTube, Stream
+
+from pydub import AudioSegment
+from io import BytesIO
+
+
+@define
+class YouTubeBytestream:
+    file: bytes
+    filename: str
+
+    @classmethod
+    def from_bytestream(
+            cls,
+            bytestream: BytesIO,
+            filename: str
+    ):
+        bytestream.seek(0)
+        return cls(
+            file=bytestream.read(),
+            filename=filename
+        )
+
+    @property
+    def dict(self):
+        return {
+            "file": self.file,
+            "filename": self.filename
+        }
+
+
+@define
+class Downloader:
+    audio_stream: Stream
+    filename: str
+
+    @classmethod
+    def from_id(cls, yt_id: str):
+        video = YouTube.from_id(yt_id)
+        audio_stream = video.streams.filter(
+            only_audio=True,
+        ).order_by('abr').desc().first()
+        return cls(
+            audio_stream=audio_stream,
+            filename=f'{audio_stream.default_filename}.mp3',
+        )
+
+    def to_bytestream(self):
+        audio_io = BytesIO()
+        self.audio_stream.stream_to_buffer(audio_io)
+        audio_io.seek(0)
+
+        return YouTubeBytestream.from_bytestream(
+            AudioSegment.from_file(
+                file=audio_io
+            ).export(BytesIO(), format='mp3', codec='libmp3lame'),
+            self.filename,
+        )

bot/modules/youtube/song.py

@@ -0,0 +1,51 @@
+from attrs import define
+import ytmusicapi
+
+from .downloader import Downloader
+
+
+@define
+class SongItem:
+    name: str
+    id: str
+    artists: list[str]
+    thumbnail: str
+
+    @classmethod
+    def from_youtube(cls, song_item: dict):
+        return cls(
+            name=song_item['title'],
+            id=song_item['videoId'],
+            artists=[artist['name'] for artist in song_item['artists']],
+            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 f"{', '.join(self.artists)} - {self.name}"
+
+    def to_bytestream(self):
+        return Downloader.from_id(self.id).to_bytestream()
+
+
+@define
+class Songs(object):
+    ytm: ytmusicapi.YTMusic
+
+    def search(self, query: str, limit: int = 10) -> list[SongItem] | None:
+        r = self.ytm.search(query, limit=limit, filter='songs')
+
+        if r is None:
+            return None
+
+        return [SongItem.from_youtube(song_item) for song_item in r]
+
+    def search_one(self, query: str) -> SongItem | None:
+        return (self.search(query, limit=1) or [None])[0]

bot/modules/youtube/youtube.py

@@ -0,0 +1,14 @@
+import ytmusicapi
+
+from .song import Songs
+from .downloader import Downloader
+
+
+class YouTube(object):
+    def __init__(self):
+        self.ytm = ytmusicapi.YTMusic()
+
+        self.download = Downloader
+        self.songs = Songs(
+            self.ytm
+        )

pyproject.toml

@@ -1,5 +1,5 @@
 [tool.poetry]
-name = "AnyMusicBot"
+name = "anymusicbot"
 version = "0.1.0"
 description = ""
 authors = ["BarsTiger"]
@@ -15,6 +15,9 @@ shazamio = { path = "lib/ShazamIO" }
 sqlitedict = "^2.1.0"
 spotipy = "^2.23.0"
 attrs = "^23.1.0"
+ytmusicapi = "^1.3.0"
+pytube = "^15.0.0"
+pydub = "^0.25.1"
 
 
 [build-system]