Fix youtube downloading wrong track, saving exceptions, attempt to fix deezer

This commit is contained in:
BarsTiger
2023-10-30 23:39:54 +02:00
parent 2ae18aacae
commit 8cd956388e
13 changed files with 272 additions and 39 deletions

View File

@@ -15,6 +15,7 @@ class Db(object):
self.fsm = DBDict('fsm')
self.config = DBDict('config')
self.inline = DBDict('inline')
self.errors = DBDict('errors')
self.spotify = DBDict('spotify')
self.deezer = DBDict('deezer')
self.youtube = DBDict('youtube')

View File

@@ -44,12 +44,18 @@ class Downloader:
driver: DeezerDriver
):
track = await driver.reverse_get_track(song_id)
return cls(
song_id=str(song_id),
driver=driver,
track=track['results'],
song=await FullSongItem.from_deezer(track)
)
try:
return cls(
song_id=str(song_id),
driver=driver,
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)
async def to_bytestream(self) -> DeezerBytestream:
quality = track_formats.MP3_128

View File

@@ -34,3 +34,6 @@ class DeezerDriver:
)
return data['data']
async def renew_engine(self):
self.engine = await self.engine.from_arl(self.engine.arl)

View File

@@ -20,6 +20,7 @@ HTTP_HEADERS = {
@define
class DeezerEngine:
cookies: dict
arl: str = None
token: str = None
@classmethod
@@ -34,6 +35,7 @@ class DeezerEngine:
return cls(
cookies=cookies,
arl=arl,
token=token
)

View File

@@ -0,0 +1 @@
from .handler import on_error

View File

@@ -0,0 +1,53 @@
from bot.common import console
from aiogram.types.error_event import ErrorEvent
from aiogram import Bot
from rich.traceback import Traceback
from bot.modules.database import db
from dataclasses import dataclass
@dataclass
class Error:
traceback: Traceback
inline_message_id: str | None = None
async def on_error(event: ErrorEvent, bot: Bot):
import os
import base64
error_id = base64.urlsafe_b64encode(os.urandom(6)).decode()
traceback = Traceback.from_exception(
type(event.exception),
event.exception,
event.exception.__traceback__,
show_locals=True,
max_frames=1,
)
if event.update.chosen_inline_result:
db.errors[error_id] = Error(
traceback=traceback,
inline_message_id=event.update.chosen_inline_result.inline_message_id,
)
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',
)
else:
db.errors[error_id] = Error(
traceback=traceback,
)
console.print(f'[red]{error_id} occurred[/]')
console.print(event)
console.print(traceback)
console.print(f'-{error_id}-')

View File

@@ -0,0 +1,50 @@
import os
import traceback
import contextlib
import re
class PrettyException:
def __init__(self, e: Exception):
self.pretty_exception = f"""
❌ Error! Report it to admins:
🐊 <code>{e.__traceback__.tb_frame.f_code.co_filename.replace(os.getcwd(), "")}\r
</code>:{e.__traceback__.tb_frame.f_lineno}
😍 {e.__class__.__name__}
👉 {"".join(traceback.format_exception_only(e)).strip()}
⬇️ Trace:
{self.get_full_stack()}
"""
@staticmethod
def get_full_stack():
full_stack = traceback.format_exc().replace(
"Traceback (most recent call last):\n", ""
)
line_regex = r' File "(.*?)", line ([0-9]+), in (.+)'
def format_line(line: str) -> str:
filename_, lineno_, name_ = re.search(line_regex, line).groups()
with contextlib.suppress(Exception):
filename_ = os.path.basename(filename_)
return (
f"🤯 <code>{filename_}:{lineno_}</code> (<b>in</b>"
f" <code>{name_}</code> call)"
)
full_stack = "\n".join(
[
format_line(line)
if re.search(line_regex, line)
else f"<code>{line}</code>"
for line in full_stack.splitlines()
]
)
return full_stack
def __str__(self):
return self.pretty_exception

View File

@@ -38,16 +38,26 @@ class SongItem(BaseSongItem):
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')
def search(
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
)
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]
def search_one(self, query: str, exact_match: bool = False) -> SongItem | None:
return (self.search(query, limit=1, exact_match=exact_match) or [None])[0]
def from_id(self, song_id: str) -> SongItem | None:
r = self.ytm.get_song(song_id)