cleanup
This commit is contained in:
parent
119f40daf8
commit
f2ba97b9d1
@ -18,3 +18,25 @@ class LyricsResult:
|
|||||||
lyrics: Union[str, list]
|
lyrics: Union[str, list]
|
||||||
confidence: int
|
confidence: int
|
||||||
time: float = 0.00
|
time: float = 0.00
|
||||||
|
|
||||||
|
|
||||||
|
"""
|
||||||
|
Generic
|
||||||
|
"""
|
||||||
|
class InvalidLyricSearchResponseException(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
"""
|
||||||
|
Genius
|
||||||
|
"""
|
||||||
|
class InvalidGeniusResponseException(
|
||||||
|
InvalidLyricSearchResponseException):
|
||||||
|
pass
|
||||||
|
|
||||||
|
"""
|
||||||
|
LRCLib
|
||||||
|
"""
|
||||||
|
|
||||||
|
class InvalidLRCLibResponseException(
|
||||||
|
InvalidLyricSearchResponseException):
|
||||||
|
pass
|
@ -9,16 +9,12 @@ from bs4 import BeautifulSoup, ResultSet # type: ignore
|
|||||||
import html as htm
|
import html as htm
|
||||||
from . import private, common, cache, redis_cache
|
from . import private, common, cache, redis_cache
|
||||||
from lyric_search import utils
|
from lyric_search import utils
|
||||||
from lyric_search.constructors import LyricsResult
|
from lyric_search.constructors import (
|
||||||
|
LyricsResult, InvalidGeniusResponseException)
|
||||||
|
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
log_level = logging.getLevelName(logger.level)
|
log_level = logging.getLevelName(logger.level)
|
||||||
|
|
||||||
class InvalidResponseException(Exception):
|
|
||||||
"""
|
|
||||||
InvalidResponseException
|
|
||||||
"""
|
|
||||||
|
|
||||||
class Genius:
|
class Genius:
|
||||||
"""
|
"""
|
||||||
Genius Search Module
|
Genius Search Module
|
||||||
@ -60,23 +56,23 @@ class Genius:
|
|||||||
text: Optional[str] = await request.text()
|
text: Optional[str] = await request.text()
|
||||||
|
|
||||||
if not text:
|
if not text:
|
||||||
raise InvalidResponseException("No search response.")
|
raise InvalidGeniusResponseException("No search response.")
|
||||||
|
|
||||||
if len(text) < 100:
|
if len(text) < 100:
|
||||||
raise InvalidResponseException("Search response text was invalid (len < 100 chars.)")
|
raise InvalidGeniusResponseException("Search response text was invalid (len < 100 chars.)")
|
||||||
search_data = await request.json()
|
search_data = await request.json()
|
||||||
|
|
||||||
if not isinstance(search_data, dict):
|
if not isinstance(search_data, dict):
|
||||||
raise InvalidResponseException("Invalid JSON.")
|
raise InvalidGeniusResponseException("Invalid JSON.")
|
||||||
|
|
||||||
if not isinstance(search_data['response'], dict):
|
if not isinstance(search_data['response'], dict):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find response key.\n{search_data}")
|
raise InvalidGeniusResponseException(f"Invalid JSON: Cannot find response key.\n{search_data}")
|
||||||
|
|
||||||
if not isinstance(search_data['response']['sections'], list):
|
if not isinstance(search_data['response']['sections'], list):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find response->sections key.\n{search_data}")
|
raise InvalidGeniusResponseException(f"Invalid JSON: Cannot find response->sections key.\n{search_data}")
|
||||||
|
|
||||||
if not isinstance(search_data['response']['sections'][0]['hits'], list):
|
if not isinstance(search_data['response']['sections'][0]['hits'], list):
|
||||||
raise InvalidResponseException("Invalid JSON: Cannot find response->sections[0]->hits key.")
|
raise InvalidGeniusResponseException("Invalid JSON: Cannot find response->sections[0]->hits key.")
|
||||||
|
|
||||||
possible_matches: list = search_data['response']['sections'][0]['hits']
|
possible_matches: list = search_data['response']['sections'][0]['hits']
|
||||||
to_scrape: list[tuple] = [
|
to_scrape: list[tuple] = [
|
||||||
@ -98,10 +94,10 @@ class Genius:
|
|||||||
scrape_text: Optional[str] = await scrape_request.text()
|
scrape_text: Optional[str] = await scrape_request.text()
|
||||||
|
|
||||||
if not scrape_text:
|
if not scrape_text:
|
||||||
raise InvalidResponseException("No scrape response.")
|
raise InvalidGeniusResponseException("No scrape response.")
|
||||||
|
|
||||||
if len(scrape_text) < 100:
|
if len(scrape_text) < 100:
|
||||||
raise InvalidResponseException("Scrape response was invalid (len < 100 chars.)")
|
raise InvalidGeniusResponseException("Scrape response was invalid (len < 100 chars.)")
|
||||||
|
|
||||||
|
|
||||||
html = BeautifulSoup(htm.unescape(scrape_text).replace('<br/>', '\n'), "html.parser")
|
html = BeautifulSoup(htm.unescape(scrape_text).replace('<br/>', '\n'), "html.parser")
|
||||||
|
@ -8,15 +8,11 @@ from aiohttp import ClientTimeout, ClientSession
|
|||||||
from lyric_search import utils
|
from lyric_search import utils
|
||||||
from lyric_search.constructors import LyricsResult
|
from lyric_search.constructors import LyricsResult
|
||||||
from . import common, cache, redis_cache
|
from . import common, cache, redis_cache
|
||||||
|
from constructors import InvalidLRCLibResponseException
|
||||||
|
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
log_level = logging.getLevelName(logger.level)
|
log_level = logging.getLevelName(logger.level)
|
||||||
|
|
||||||
class InvalidResponseException(Exception):
|
|
||||||
"""
|
|
||||||
Invalid Response Exception
|
|
||||||
"""
|
|
||||||
|
|
||||||
class LRCLib:
|
class LRCLib:
|
||||||
"""LRCLib Search Module"""
|
"""LRCLib Search Module"""
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
@ -62,18 +58,18 @@ class LRCLib:
|
|||||||
|
|
||||||
text: Optional[str] = await request.text()
|
text: Optional[str] = await request.text()
|
||||||
if not text:
|
if not text:
|
||||||
raise InvalidResponseException("No search response.")
|
raise InvalidLRCLibResponseException("No search response.")
|
||||||
if len(text) < 100:
|
if len(text) < 100:
|
||||||
raise InvalidResponseException("Search response text was invalid (len < 100 chars.)")
|
raise InvalidLRCLibResponseException("Search response text was invalid (len < 100 chars.)")
|
||||||
|
|
||||||
search_data: Optional[Union[list, dict]] = await request.json()
|
search_data: Optional[Union[list, dict]] = await request.json()
|
||||||
if not isinstance(search_data, list|dict):
|
if not isinstance(search_data, list|dict):
|
||||||
raise InvalidResponseException("No JSON search data.")
|
raise InvalidLRCLibResponseException("No JSON search data.")
|
||||||
|
|
||||||
# logging.info("Search Data:\n%s", search_data)
|
# logging.info("Search Data:\n%s", search_data)
|
||||||
|
|
||||||
if not isinstance(search_data, list):
|
if not isinstance(search_data, list):
|
||||||
raise InvalidResponseException("Invalid JSON.")
|
raise InvalidLRCLibResponseException("Invalid JSON.")
|
||||||
|
|
||||||
if plain:
|
if plain:
|
||||||
possible_matches = [(x, f"{result.get('artistName')} - {result.get('trackName')}")
|
possible_matches = [(x, f"{result.get('artistName')} - {result.get('trackName')}")
|
||||||
@ -92,21 +88,21 @@ class LRCLib:
|
|||||||
best_match_id = best_match[0]
|
best_match_id = best_match[0]
|
||||||
|
|
||||||
if not isinstance(search_data[best_match_id]['artistName'], str):
|
if not isinstance(search_data[best_match_id]['artistName'], str):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find artistName key.\n{search_data}")
|
raise InvalidLRCLibResponseException(f"Invalid JSON: Cannot find artistName key.\n{search_data}")
|
||||||
|
|
||||||
if not isinstance(search_data[best_match_id]['trackName'], str):
|
if not isinstance(search_data[best_match_id]['trackName'], str):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find trackName key.\n{search_data}")
|
raise InvalidLRCLibResponseException(f"Invalid JSON: Cannot find trackName key.\n{search_data}")
|
||||||
|
|
||||||
returned_artist: str = search_data[best_match_id]['artistName']
|
returned_artist: str = search_data[best_match_id]['artistName']
|
||||||
returned_song: str = search_data[best_match_id]['trackName']
|
returned_song: str = search_data[best_match_id]['trackName']
|
||||||
if plain:
|
if plain:
|
||||||
if not isinstance(search_data[best_match_id]['plainLyrics'], str):
|
if not isinstance(search_data[best_match_id]['plainLyrics'], str):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find plainLyrics key.\n{search_data}")
|
raise InvalidLRCLibResponseException(f"Invalid JSON: Cannot find plainLyrics key.\n{search_data}")
|
||||||
returned_lyrics: str = search_data[best_match_id]['plainLyrics']
|
returned_lyrics: str = search_data[best_match_id]['plainLyrics']
|
||||||
returned_lyrics = self.datautils.scrub_lyrics(returned_lyrics)
|
returned_lyrics = self.datautils.scrub_lyrics(returned_lyrics)
|
||||||
else:
|
else:
|
||||||
if not isinstance(search_data[best_match_id]['syncedLyrics'], str):
|
if not isinstance(search_data[best_match_id]['syncedLyrics'], str):
|
||||||
raise InvalidResponseException(f"Invalid JSON: Cannot find syncedLyrics key.\n{search_data}")
|
raise InvalidLRCLibResponseException(f"Invalid JSON: Cannot find syncedLyrics key.\n{search_data}")
|
||||||
returned_lyrics: str = search_data[best_match_id]['syncedLyrics']
|
returned_lyrics: str = search_data[best_match_id]['syncedLyrics']
|
||||||
lrc_obj = self.datautils.create_lrc_object(returned_lyrics)
|
lrc_obj = self.datautils.create_lrc_object(returned_lyrics)
|
||||||
returned_track: str = f"{returned_artist} - {returned_song}"
|
returned_track: str = f"{returned_artist} - {returned_song}"
|
||||||
|
@ -23,3 +23,5 @@ dependencies = [
|
|||||||
"typing-inspect>=0.9.0",
|
"typing-inspect>=0.9.0",
|
||||||
"http3>=0.6.7",
|
"http3>=0.6.7",
|
||||||
]
|
]
|
||||||
|
[tool.mypy]
|
||||||
|
disable_error_code = import-untyped
|
6
utils/constructors.py
Normal file
6
utils/constructors.py
Normal file
@ -0,0 +1,6 @@
|
|||||||
|
"""
|
||||||
|
LastFM
|
||||||
|
"""
|
||||||
|
|
||||||
|
class InvalidLastFMResponseException(Exception):
|
||||||
|
pass
|
@ -4,6 +4,7 @@ from typing import Optional, Union
|
|||||||
import regex
|
import regex
|
||||||
from aiohttp import ClientSession, ClientTimeout
|
from aiohttp import ClientSession, ClientTimeout
|
||||||
from constants import Constants
|
from constants import Constants
|
||||||
|
from .constructors import InvalidLastFMResponseException
|
||||||
|
|
||||||
class LastFM:
|
class LastFM:
|
||||||
"""LastFM Endpoints"""
|
"""LastFM Endpoints"""
|
||||||
@ -182,12 +183,20 @@ class LastFM:
|
|||||||
timeout=ClientTimeout(connect=3, sock_read=8)) as request:
|
timeout=ClientTimeout(connect=3, sock_read=8)) as request:
|
||||||
request.raise_for_status()
|
request.raise_for_status()
|
||||||
data: dict = await request.json()
|
data: dict = await request.json()
|
||||||
|
if not data.get('profile'):
|
||||||
|
raise InvalidLastFMResponseException("Data did not contain 'profile' key.")
|
||||||
|
|
||||||
|
_id: int = data.get('id', None)
|
||||||
|
name: str = data.get('name', None)
|
||||||
|
profile: str = data.get('profile', '')
|
||||||
|
profile = regex.sub(r"(\[(\/{0,})(u|b|i)])", "", profile)
|
||||||
|
members: list = data.get('members', None)
|
||||||
|
|
||||||
ret_obj: dict = {
|
ret_obj: dict = {
|
||||||
'id': data.get('id'),
|
'id': _id,
|
||||||
'name': data.get('name'),
|
'name': name,
|
||||||
'profile': regex.sub(r"(\[(\/{0,})(u|b|i)])", "", data.get('profile')),
|
'profile': profile,
|
||||||
'members': data.get('members'),
|
'members': members,
|
||||||
}
|
}
|
||||||
return ret_obj
|
return ret_obj
|
||||||
except:
|
except:
|
Loading…
x
Reference in New Issue
Block a user