tgbots/AnyMusicBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

used black

Browse Source
This commit is contained in:
hhh
2024-11-02 00:10:24 +02:00
parent 1b1f217b75
commit e0a3d256d5
79 changed files with 658 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bot/modules/common/song/song.py
View File

@@ -11,7 +11,7 @@ class BaseSongItem:
@property
def all_artists(self):
return ', '.join(self.artists)
return ", ".join(self.artists)
@property
def full_name(self):

2
bot/modules/database/__init__.py
View File

@@ -3,4 +3,4 @@ from .db import Db
db = Db()
__all__ = ['db']
__all__ = ["db"]

20
bot/modules/database/db.py
View File

@@ -3,13 +3,13 @@ from .db_model import DBDict
class Db(object):
def __init__(self):
self.fsm = DBDict('fsm')
self.config = DBDict('config')
self.inline = DBDict('inline')
self.errors = DBDict('errors')
self.settings = DBDict('settings')
self.spotify = DBDict('spotify')
self.deezer = DBDict('deezer')
self.youtube = DBDict('youtube')
self.soundcloud = DBDict('soundcloud')
self.recoded = DBDict('recoded')
self.fsm = DBDict("fsm")
self.config = DBDict("config")
self.inline = DBDict("inline")
self.errors = DBDict("errors")
self.settings = DBDict("settings")
self.spotify = DBDict("spotify")
self.deezer = DBDict("deezer")
self.youtube = DBDict("youtube")
self.soundcloud = DBDict("soundcloud")
self.recoded = DBDict("recoded")

2
bot/modules/deezer/__init__.py
View File

@@ -7,4 +7,4 @@ deezer = Deezer(
arl=config.tokens.deezer.arl,
)
__all__ = ['deezer', 'DeezerBytestream']
__all__ = ["deezer", "DeezerBytestream"]

31
bot/modules/deezer/downloader.py
View File

@@ -17,10 +17,7 @@ class DeezerBytestream:
@classmethod
def from_bytestream(
cls,
bytestream: BytesIO,
filename: str,
full_song: FullSongItem
cls, bytestream: BytesIO, filename: str, full_song: FullSongItem
):
bytestream.seek(0)
return cls(
@@ -38,21 +35,18 @@ class Downloader:
song: FullSongItem
@classmethod
async def build(
cls,
song_id: str,
driver: DeezerDriver
):
async def build(cls, song_id: str, driver: DeezerDriver):
track = await driver.reverse_get_track(song_id)
try:
return cls(
song_id=str(song_id),
driver=driver,
track=track['results'],
song=await FullSongItem.from_deezer(track)
track=track["results"],
song=await FullSongItem.from_deezer(track),
)
except KeyError:
from icecream import ic
ic(track)
await driver.renew_engine()
return await cls.build(song_id, driver)
@@ -65,7 +59,7 @@ class Downloader:
audio = BytesIO()
async for chunk in self.driver.engine.get_data_iter(
await self._get_download_url(quality=quality)
await self._get_download_url(quality=quality)
):
if i % 3 > 0 or len(chunk) < 2 * 1024:
audio.write(chunk)
@@ -76,18 +70,16 @@ class Downloader:
return DeezerBytestream.from_bytestream(
filename=self.song.full_name + track_formats.TRACK_FORMAT_MAP[quality].ext,
bytestream=audio,
full_song=self.song
full_song=self.song,
)
async def _get_download_url(self, quality: str = 'MP3_128'):
async def _get_download_url(self, quality: str = "MP3_128"):
md5_origin = self.track["MD5_ORIGIN"]
track_id = self.track["SNG_ID"]
media_version = self.track["MEDIA_VERSION"]
url_decrypter = UrlDecrypter(
md5_origin=md5_origin,
track_id=track_id,
media_version=media_version
md5_origin=md5_origin, track_id=track_id, media_version=media_version
)
return url_decrypter.get_url_for(track_formats.TRACK_FORMAT_MAP[quality])
@@ -98,7 +90,4 @@ class DownloaderBuilder:
driver: DeezerDriver
async def from_id(self, song_id: str):
return await Downloader.build(
song_id=song_id,
driver=self.driver
)
return await Downloader.build(song_id=song_id, driver=self.driver)

19
bot/modules/deezer/driver.py
View File

@@ -10,30 +10,19 @@ class DeezerDriver:
engine: DeezerEngine
async def get_track(self, track_id: int | str):
data = await self.engine.call_legacy_api(
f'track/{track_id}'
)
data = await self.engine.call_legacy_api(f"track/{track_id}")
return data
async def reverse_get_track(self, track_id: str):
return await self.engine.call_api(
'song.getData',
params={
'SNG_ID': track_id
}
)
return await self.engine.call_api("song.getData", params={"SNG_ID": track_id})
async def search(self, query: str, limit: int = 30):
data = await self.engine.call_legacy_api(
'search/track',
params={
'q': clean_query(query),
'limit': limit
}
"search/track", params={"q": clean_query(query), "limit": limit}
)
return data['data']
return data["data"]
async def renew_engine(self):
self.engine = await self.engine.from_arl(self.engine.arl)

57
bot/modules/deezer/engine.py
View File

@@ -7,13 +7,13 @@ from attrs import define
HTTP_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36",
"(KHTML, like Gecko) Chrome/67.0.3396.99 Safari/537.36",
"Content-Language": "en-US",
"Cache-Control": "max-age=0",
"Accept": "*/*",
"Accept-Charset": "utf-8,ISO-8859-1;q=0.7,*;q=0.3",
"Accept-Language": "en-US,en;q=0.9,en-US;q=0.8,en;q=0.7",
"Connection": 'keep-alive'
"Connection": "keep-alive",
}
@@ -25,28 +25,22 @@ class DeezerEngine:
@classmethod
async def from_arl(cls, arl: str):
cookies = {'arl': arl}
cookies = {"arl": arl}
data, cookies = await cls(cookies).call_api(
'deezer.getUserData', get_cookies=True
"deezer.getUserData", get_cookies=True
)
data = data['results']
token = data['checkForm']
data = data["results"]
token = data["checkForm"]
return cls(
cookies=cookies,
arl=arl,
token=token
)
return cls(cookies=cookies, arl=arl, token=token)
async def call_legacy_api(
self, request_point: str, params: dict = None
):
async def call_legacy_api(self, request_point: str, params: dict = None):
async with aiohttp.ClientSession(cookies=self.cookies) as session:
async with session.get(
f"https://api.deezer.com/{request_point}",
params=params,
headers=HTTP_HEADERS
f"https://api.deezer.com/{request_point}",
params=params,
headers=HTTP_HEADERS,
) as r:
return await r.json()
@@ -63,31 +57,26 @@ class DeezerEngine:
async def get_data_iter(self, url: str):
async with aiohttp.ClientSession(
cookies=self.cookies,
headers=HTTP_HEADERS
cookies=self.cookies, headers=HTTP_HEADERS
) as session:
r = await session.get(
url,
allow_redirects=True
)
r = await session.get(url, allow_redirects=True)
async for chunk in self._iter_exact_chunks(r):
yield chunk
async def call_api(
self, method: str, params: dict = None,
get_cookies: bool = False
self, method: str, params: dict = None, get_cookies: bool = False
):
async with aiohttp.ClientSession(cookies=self.cookies) as session:
async with session.post(
f"https://www.deezer.com/ajax/gw-light.php",
params={
'method': method,
'api_version': '1.0',
'input': '3',
'api_token': self.token or 'null',
},
headers=HTTP_HEADERS,
json=params
f"https://www.deezer.com/ajax/gw-light.php",
params={
"method": method,
"api_version": "1.0",
"input": "3",
"api_token": self.token or "null",
},
headers=HTTP_HEADERS,
json=params,
) as r:
if not get_cookies:
return await r.json()

38
bot/modules/deezer/song.py
View File

@@ -10,11 +10,11 @@ class SongItem(BaseSongItem):
@classmethod
def from_deezer(cls, song_item: dict):
return cls(
name=song_item['title'],
id=str(song_item['id']),
artists=[song_item['artist']['name']],
preview_url=song_item.get('preview'),
thumbnail=song_item['album']['cover_medium']
name=song_item["title"],
id=str(song_item["id"]),
artists=[song_item["artist"]["name"]],
preview_url=song_item.get("preview"),
thumbnail=song_item["album"]["cover_medium"],
)
@@ -25,21 +25,23 @@ class FullSongItem(BaseSongItem):
@classmethod
async def from_deezer(cls, song_item: dict):
if song_item.get('results'):
song_item = song_item['results']
if song_item.get("results"):
song_item = song_item["results"]
return cls(
name=song_item['SNG_TITLE'],
id=song_item['SNG_ID'],
artists=[artist['ART_NAME'] for artist in song_item['ARTISTS']],
preview_url=(song_item.get('MEDIA').get('HREF')
if type(song_item.get('MEDIA')) is dict and
song_item.get('MEDIA').get('TYPE') == 'preview'
else None),
thumbnail=f'https://e-cdns-images.dzcdn.net/images/cover/'
f'{song_item["ALB_PICTURE"]}/320x320.jpg',
duration=int(song_item['DURATION']),
track_dict=song_item
name=song_item["SNG_TITLE"],
id=song_item["SNG_ID"],
artists=[artist["ART_NAME"] for artist in song_item["ARTISTS"]],
preview_url=(
song_item.get("MEDIA").get("HREF")
if type(song_item.get("MEDIA")) is dict
and song_item.get("MEDIA").get("TYPE") == "preview"
else None
),
thumbnail=f"https://e-cdns-images.dzcdn.net/images/cover/"
f'{song_item["ALB_PICTURE"]}/320x320.jpg',
duration=int(song_item["DURATION"]),
track_dict=song_item,
)

35
bot/modules/deezer/track_formats.py
View File

@@ -19,32 +19,11 @@ class TrackFormat:
TRACK_FORMAT_MAP = {
FLAC: TrackFormat(
code=9,
ext=".flac"
),
MP3_128: TrackFormat(
code=1,
ext=".mp3"
),
MP3_256: TrackFormat(
code=5,
ext=".mp3"
),
MP3_320: TrackFormat(
code=3,
ext=".mp3"
),
MP4_RA1: TrackFormat(
code=13,
ext=".mp4"
),
MP4_RA2: TrackFormat(
code=14,
ext=".mp4"
),
MP4_RA3: TrackFormat(
code=15,
ext=".mp3"
)
FLAC: TrackFormat(code=9, ext=".flac"),
MP3_128: TrackFormat(code=1, ext=".mp3"),
MP3_256: TrackFormat(code=5, ext=".mp3"),
MP3_320: TrackFormat(code=3, ext=".mp3"),
MP4_RA1: TrackFormat(code=13, ext=".mp4"),
MP4_RA2: TrackFormat(code=14, ext=".mp4"),
MP4_RA3: TrackFormat(code=15, ext=".mp3"),
}

24
bot/modules/deezer/util.py
View File

@@ -34,20 +34,20 @@ class UrlDecrypter:
media_version: str
def get_url_for(self, track_format: TrackFormat):
step1 = (f'{self.md5_origin}¤{track_format.code}¤'
f'{self.track_id}¤{self.media_version}')
step1 = (
f"{self.md5_origin}¤{track_format.code}¤"
f"{self.track_id}¤{self.media_version}"
)
m = hashlib.md5()
m.update(bytes([ord(x) for x in step1]))
step2 = f'{m.hexdigest()}¤{step1}¤'
step2 = f"{m.hexdigest()}¤{step1}¤"
step2 = step2.ljust(80, " ")
cipher = Cipher(
algorithm=algorithms.AES(
key=bytes('jo6aey6haid2Teih', 'ascii')
),
algorithm=algorithms.AES(key=bytes("jo6aey6haid2Teih", "ascii")),
mode=modes.ECB(),
backend=default_backend()
backend=default_backend(),
)
encryptor = cipher.encryptor()
@@ -55,7 +55,7 @@ class UrlDecrypter:
cdn = self.md5_origin[0]
return f'https://e-cdns-proxy-{cdn}.dzcdn.net/mobile/1/{step3}'
return f"https://e-cdns-proxy-{cdn}.dzcdn.net/mobile/1/{step3}"
@define
@@ -69,12 +69,10 @@ class ChunkDecrypter:
cipher = Cipher(
algorithms.Blowfish(get_blowfish_key(track_id)),
modes.CBC(bytes([i for i in range(8)])),
default_backend()
default_backend(),
)
return cls(
cipher=cipher
)
return cls(cipher=cipher)
def decrypt_chunk(self, chunk: bytes):
decryptor = self.cipher.decryptor()
@@ -82,7 +80,7 @@ class ChunkDecrypter:
def get_blowfish_key(track_id: str):
secret = 'g4el58wc0zvf9na1'
secret = "g4el58wc0zvf9na1"
m = hashlib.md5()
m.update(bytes([ord(x) for x in track_id]))

10
bot/modules/error/handler.py
View File

@@ -42,9 +42,9 @@ async def on_error(event: ErrorEvent, bot: Bot):
await bot.edit_message_caption(
inline_message_id=event.update.chosen_inline_result.inline_message_id,
caption=f'💔 <b>ERROR</b> occurred. Use this code to get more information: '
f'<code>{error_id}</code>',
parse_mode='HTML',
caption=f"💔 <b>ERROR</b> occurred. Use this code to get more information: "
f"<code>{error_id}</code>",
parse_mode="HTML",
)
else:
@@ -53,7 +53,7 @@ async def on_error(event: ErrorEvent, bot: Bot):
exception=pretty_exception,
)
console.print(f'[red]{error_id} occurred[/]')
console.print(f"[red]{error_id} occurred[/]")
console.print(event)
console.print(traceback)
console.print(f'-{error_id} occurred-')
console.print(f"-{error_id} occurred-")

20
bot/modules/error/pretty.py
View File

@@ -13,12 +13,14 @@ class PrettyException:
🐊 <code>{e.__traceback__.tb_frame.f_code.co_filename.replace(os.getcwd(), "")}\r
</code>:{e.__traceback__.tb_frame.f_lineno}
"""
self.short = (f'{e.__class__.__name__}: '
f'{"".join(traceback.format_exception_only(e)).strip()}')
self.short = (
f"{e.__class__.__name__}: "
f'{"".join(traceback.format_exception_only(e)).strip()}'
)
self.pretty_exception = (f"{self.long}\n\n"
f"⬇️ Trace:"
f"{self.get_full_stack()}")
self.pretty_exception = (
f"{self.long}\n\n" f"⬇️ Trace:" f"{self.get_full_stack()}"
)
@staticmethod
def get_full_stack():
@@ -40,9 +42,11 @@ class PrettyException:
full_stack = "\n".join(
[
format_line(line)
if re.search(line_regex, line)
else f"<code>{line}</code>"
(
format_line(line)
if re.search(line_regex, line)
else f"<code>{line}</code>"
)
for line in full_stack.splitlines()
]
)

8
bot/modules/fsm/in_db.py
View File

@@ -18,14 +18,14 @@ class MemoryStorageRecord:
class StorageDict(DefaultDict):
def __init__(self, default_factory=None) -> None:
if type(db.fsm.get('fsm')) is not dict:
db.fsm['fsm'] = dict()
if type(db.fsm.get("fsm")) is not dict:
db.fsm["fsm"] = dict()
super().__init__(default_factory, db.fsm['fsm'])
super().__init__(default_factory, db.fsm["fsm"])
def __setitem__(self, key, value):
super().__setitem__(key, value)
db.fsm['fsm'] = dict(self)
db.fsm["fsm"] = dict(self)
class InDbStorage(BaseStorage):

64
bot/modules/settings/model.py
View File

@@ -11,46 +11,32 @@ class Setting:
settings_strings: dict[str, Setting] = {
'search_preview': Setting(
name='Search preview',
description='Show only covers (better display), '
'or add 30 seconds of track preview whenever possible?',
choices={
'cover': 'Cover picture',
'preview': 'Audio preview'
},
"search_preview": Setting(
name="Search preview",
description="Show only covers (better display), "
"or add 30 seconds of track preview whenever possible?",
choices={"cover": "Cover picture", "preview": "Audio preview"},
),
'recode_youtube': Setting(
name='Recode YouTube (and Spotify)',
description='Recode when downloading from YouTube (and Spotify) to '
'more compatible format (may take some time)',
choices={
'no': 'Send original file',
'yes': 'Recode to libmp3lame'
},
"recode_youtube": Setting(
name="Recode YouTube (and Spotify)",
description="Recode when downloading from YouTube (and Spotify) to "
"more compatible format (may take some time)",
choices={"no": "Send original file", "yes": "Recode to libmp3lame"},
),
'exact_spotify_search': Setting(
name='Only exact Spotify matches',
description='When searching on Youtube from Spotify, show only exact matches, '
'may protect against inaccurate matches, but at the same time it '
'can lose reuploaded tracks. Should be enabled always, except in '
'situations where the track is not found on both YouTube and '
'Deezer',
choices={
'yes': 'Only exact matches',
'no': 'Fuzzy matches also'
},
"exact_spotify_search": Setting(
name="Only exact Spotify matches",
description="When searching on Youtube from Spotify, show only exact matches, "
"may protect against inaccurate matches, but at the same time it "
"can lose reuploaded tracks. Should be enabled always, except in "
"situations where the track is not found on both YouTube and "
"Deezer",
choices={"yes": "Only exact matches", "no": "Fuzzy matches also"},
),
"default_search_provider": Setting(
name="Default search provider",
description="Which service to use when searching without service filter",
choices={"d": "Deezer", "c": "SoundCloud", "y": "YouTube", "s": "Spotify"},
),
'default_search_provider': Setting(
name='Default search provider',
description='Which service to use when searching without service filter',
choices={
'd': 'Deezer',
'c': 'SoundCloud',
'y': 'YouTube',
's': 'Spotify'
}
)
}
@@ -64,8 +50,8 @@ class UserSettings:
if db.settings.get(self.user_id) is None:
db.settings[self.user_id] = dict(
(setting, list(settings_strings[setting].choices)[0]) for setting in
settings_strings
(setting, list(settings_strings[setting].choices)[0])
for setting in settings_strings
)
def __getitem__(self, item):

2
bot/modules/soundcloud/__init__.py
View File

@@ -7,4 +7,4 @@ soundcloud = SoundCloud(
client_id=config.tokens.soundcloud.client_id,
)
__all__ = ['soundcloud', 'SoundCloudBytestream']
__all__ = ["soundcloud", "SoundCloudBytestream"]

65
bot/modules/soundcloud/downloader.py
View File

@@ -15,18 +15,9 @@ class SoundCloudBytestream:
song: SongItem
@classmethod
def from_bytes(
cls,
bytes_: bytes,
filename: str,
duration: int,
song: SongItem
):
def from_bytes(cls, bytes_: bytes, filename: str, duration: int, song: SongItem):
return cls(
file=bytes_,
filename=filename,
duration=int(duration / 1000),
song=song
file=bytes_, filename=filename, duration=int(duration / 1000), song=song
)
@@ -40,60 +31,53 @@ class Downloader:
song: SongItem
@classmethod
async def build(
cls,
song_id: str,
driver: SoundCloudDriver
):
async def build(cls, song_id: str, driver: SoundCloudDriver):
track = await driver.get_track(song_id)
song = SongItem.from_soundcloud(track)
if url := cls._try_get_progressive(track['media']['transcodings']):
if url := cls._try_get_progressive(track["media"]["transcodings"]):
method = cls._progressive
else:
url = track['media']['transcodings'][0]['url']
method = cls._hls if \
(track['media']['transcodings'][0]['format']['protocol']
== 'hls') else cls._progressive
url = track["media"]["transcodings"][0]["url"]
method = (
cls._hls
if (track["media"]["transcodings"][0]["format"]["protocol"] == "hls")
else cls._progressive
)
return cls(
driver=driver,
duration=track['duration'],
duration=track["duration"],
method=method,
download_url=url,
filename=f'{track["title"]}.mp3',
song=song
song=song,
)
@staticmethod
def _try_get_progressive(urls: list) -> str | None:
for transcode in urls:
if transcode['format']['protocol'] == 'progressive':
return transcode['url']
if transcode["format"]["protocol"] == "progressive":
return transcode["url"]
async def _progressive(self, url: str) -> bytes:
return await self.driver.engine.read_data(
url=(await self.driver.engine.get(
url
))['url']
url=(await self.driver.engine.get(url))["url"]
)
async def _hls(self, url: str) -> bytes:
m3u8_obj = m3u8.loads(
(await self.driver.engine.read_data(
(await self.driver.engine.get(
url=url
))['url']
)).decode()
(
await self.driver.engine.read_data(
(await self.driver.engine.get(url=url))["url"]
)
).decode()
)
content = bytearray()
for segment in m3u8_obj.files:
content.extend(
await self.driver.engine.read_data(
url=segment,
append_client_id=False
)
await self.driver.engine.read_data(url=segment, append_client_id=False)
)
return content
@@ -103,7 +87,7 @@ class Downloader:
bytes_=await self.method(self, self.download_url),
filename=self.filename,
duration=self.duration,
song=self.song
song=self.song,
)
@@ -112,7 +96,4 @@ class DownloaderBuilder:
driver: SoundCloudDriver
async def from_id(self, song_id: str):
return await Downloader.build(
song_id=song_id,
driver=self.driver
)
return await Downloader.build(song_id=song_id, driver=self.driver)

21
bot/modules/soundcloud/driver.py
View File

@@ -8,23 +8,12 @@ class SoundCloudDriver:
engine: SoundCloudEngine
async def get_track(self, track_id: int | str):
return await self.engine.call(
f'tracks/{track_id}'
)
return await self.engine.call(f"tracks/{track_id}")
async def search(self, query: str, limit: int = 30):
return (await self.engine.call(
'search/tracks',
params={
'q': query,
'limit': limit
}
))['collection']
return (
await self.engine.call("search/tracks", params={"q": query, "limit": limit})
)["collection"]
async def resolve_url(self, url: str):
return await self.engine.call(
'resolve',
params={
'url': url
}
)
return await self.engine.call("resolve", params={"url": url})

30
bot/modules/soundcloud/engine.py
View File

@@ -8,27 +8,33 @@ class SoundCloudEngine:
async def call(self, request_point: str, params: dict = None):
return await self.get(
url=f'https://api-v2.soundcloud.com/{request_point}',
params=params
url=f"https://api-v2.soundcloud.com/{request_point}", params=params
)
async def get(self, url: str, params: dict = None):
async with aiohttp.ClientSession() as session:
async with session.get(
url,
params=(params or {}) | {
'client_id': self.client_id,
},
url,
params=(params or {})
| {
"client_id": self.client_id,
},
) as r:
return await r.json()
async def read_data(self, url: str, params: dict = None,
append_client_id: bool = True):
async def read_data(
self, url: str, params: dict = None, append_client_id: bool = True
):
async with aiohttp.ClientSession() as session:
async with session.get(
url,
params=(params or {}) | ({
'client_id': self.client_id,
} if append_client_id else {}),
url,
params=(params or {})
| (
{
"client_id": self.client_id,
}
if append_client_id
else {}
),
) as r:
return await r.content.read()

14
bot/modules/soundcloud/song.py
View File

@@ -9,13 +9,15 @@ class SongItem(BaseSongItem):
@classmethod
def from_soundcloud(cls, song_item: dict):
return cls(
name=song_item['title'],
id=str(song_item['id']),
name=song_item["title"],
id=str(song_item["id"]),
artists=[],
thumbnail=(song_item['artwork_url'] or song_item['user']['avatar_url'] or
'https://soundcloud.com/images/default_avatar_large.png')
.replace('large.jpg', 't300x300.jpg'),
preview_url=None
thumbnail=(
song_item["artwork_url"]
or song_item["user"]["avatar_url"]
or "https://soundcloud.com/images/default_avatar_large.png"
).replace("large.jpg", "t300x300.jpg"),
preview_url=None,
)
@property

4
bot/modules/spotify/__init__.py
View File

@@ -4,7 +4,7 @@ from bot.utils.config import config
spotify = Spotify(
client_id=config.tokens.spotify.client_id,
client_secret=config.tokens.spotify.client_secret
client_secret=config.tokens.spotify.client_secret,
)
__all__ = ['spotify']
__all__ = ["spotify"]

17
bot/modules/spotify/song.py
View File

@@ -9,12 +9,15 @@ class SongItem(BaseSongItem):
@classmethod
def from_spotify(cls, song_item: dict):
return cls(
name=song_item['name'],
id=song_item['id'],
artists=[artist['name'] for artist in song_item['artists']],
preview_url=song_item['preview_url'].split('?')[0] if
song_item['preview_url'] is not None else None,
thumbnail=song_item['album']['images'][1]['url']
name=song_item["name"],
id=song_item["id"],
artists=[artist["name"] for artist in song_item["artists"]],
preview_url=(
song_item["preview_url"].split("?")[0]
if song_item["preview_url"] is not None
else None
),
thumbnail=song_item["album"]["images"][1]["url"],
)
@@ -28,7 +31,7 @@ class Songs(object):
if r is None:
return None
return [SongItem.from_spotify(item) for item in r['tracks']['items']]
return [SongItem.from_spotify(item) for item in r["tracks"]["items"]]
def from_id(self, song_id: str) -> SongItem | None:
r = self.spotify.track(song_id)

5
bot/modules/spotify/spotify.py
View File

@@ -8,11 +8,10 @@ class Spotify(object):
def __init__(self, client_id, client_secret):
self.spotify = spotipy.Spotify(
client_credentials_manager=SpotifyClientCredentials(
client_id=client_id,
client_secret=client_secret
client_id=client_id, client_secret=client_secret
),
backoff_factor=0.1,
retries=10
retries=10,
)
self.songs = Songs(self.spotify)

30
bot/modules/url/id_getter.py
View File

@@ -10,26 +10,28 @@ async def get_url_after_redirect(url: str) -> str:
async def get_id(recognised: RecognisedService):
if recognised.name == 'yt':
return recognised.parse_result.path.replace('/', '') if (
recognised.parse_result.netloc.endswith('youtu.be')
) else recognised.parse_result.query.split('=')[1].split('&')[0]
if recognised.name == "yt":
return (
recognised.parse_result.path.replace("/", "")
if (recognised.parse_result.netloc.endswith("youtu.be"))
else recognised.parse_result.query.split("=")[1].split("&")[0]
)
elif recognised.name == 'spot':
if recognised.parse_result.netloc.endswith('open.spotify.com'):
return recognised.parse_result.path.split('/')[2]
elif recognised.name == "spot":
if recognised.parse_result.netloc.endswith("open.spotify.com"):
return recognised.parse_result.path.split("/")[2]
else:
url = await get_url_after_redirect(recognised.parse_result.geturl())
return url.split('/')[-1].split('?')[0]
return url.split("/")[-1].split("?")[0]
elif recognised.name == 'deez':
if recognised.parse_result.netloc.endswith('deezer.com'):
return recognised.parse_result.path.split('/')[-1]
elif recognised.name == "deez":
if recognised.parse_result.netloc.endswith("deezer.com"):
return recognised.parse_result.path.split("/")[-1]
else:
url = await get_url_after_redirect(recognised.parse_result.geturl())
return url.split('/')[-1].split('?')[0]
return url.split("/")[-1].split("?")[0]
elif recognised.name == 'sc':
if not recognised.parse_result.netloc.startswith('on'):
elif recognised.name == "sc":
if not recognised.parse_result.netloc.startswith("on"):
return recognised.parse_result.geturl()
return await get_url_after_redirect(recognised.parse_result.geturl())

26
bot/modules/url/recognise.py
View File

@@ -14,7 +14,7 @@ from bot.modules.soundcloud import soundcloud
@dataclass
class RecognisedService:
name: Literal['yt', 'spot', 'deez', 'sc']
name: Literal["yt", "spot", "deez", "sc"]
db_table: DBDict
by_id_func: Callable | Awaitable
parse_result: ParseResult
@@ -22,33 +22,33 @@ class RecognisedService:
def recognise_music_service(url: str) -> RecognisedService | None:
url = urlparse(url)
if url.netloc.endswith('youtube.com') or url.netloc.endswith('youtu.be'):
if url.netloc.endswith("youtube.com") or url.netloc.endswith("youtu.be"):
return RecognisedService(
name='yt',
name="yt",
db_table=db.youtube,
by_id_func=youtube.songs.from_id,
parse_result=url
parse_result=url,
)
elif url.netloc.endswith('open.spotify.com') or url.netloc.endswith('spotify.link'):
elif url.netloc.endswith("open.spotify.com") or url.netloc.endswith("spotify.link"):
return RecognisedService(
name='spot',
name="spot",
db_table=db.spotify,
by_id_func=spotify.songs.from_id,
parse_result=url
parse_result=url,
)
elif url.netloc.endswith('deezer.page.link') or url.netloc.endswith('deezer.com'):
elif url.netloc.endswith("deezer.page.link") or url.netloc.endswith("deezer.com"):
return RecognisedService(
name='deez',
name="deez",
db_table=db.deezer,
by_id_func=deezer.songs.from_id,
parse_result=url
parse_result=url,
)
elif url.netloc.endswith('soundcloud.com'):
elif url.netloc.endswith("soundcloud.com"):
return RecognisedService(
name='sc',
name="sc",
db_table=db.soundcloud,
by_id_func=soundcloud.songs.from_url,
parse_result=url
parse_result=url,
)
else:
return None

28
bot/modules/youtube/song.py
View File

@@ -15,19 +15,19 @@ class SongItem(BaseSongItem):
@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']
name=song_item["title"],
id=song_item["videoId"],
artists=[artist["name"] for artist in song_item["artists"]],
thumbnail=song_item["thumbnails"][1]["url"],
)
@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']
name=details["title"],
id=details["videoId"],
artists=details["author"].split(" & "),
thumbnail=details["thumbnail"]["thumbnails"][1]["url"],
)
def to_bytestream(self) -> Awaitable[YouTubeBytestream]:
@@ -39,16 +39,10 @@ class Songs(object):
ytm: ytmusicapi.YTMusic
def search(
self,
query: str,
limit: int = 10,
exact_match: bool = False
self, query: str, limit: int = 10, exact_match: bool = False
) -> list[SongItem] | None:
r = self.ytm.search(
query,
limit=limit,
filter='songs',
ignore_spelling=exact_match
query, limit=limit, filter="songs", ignore_spelling=exact_match
)
if r is None:
@@ -68,4 +62,4 @@ class Songs(object):
if r is None:
return None
return SongItem.from_details(r['videoDetails'])
return SongItem.from_details(r["videoDetails"])

4
bot/modules/youtube/youtube.py
View File

@@ -9,6 +9,4 @@ class YouTube(object):
self.ytm = ytmusicapi.YTMusic()
self.download = Downloader
self.songs = Songs(
self.ytm
)
self.songs = Songs(self.ytm)