2025-01-14 11:13:39 -05:00

117 lines
5.2 KiB
Python

#!/usr/bin/env python3.12
# pylint: disable=bare-except, broad-exception-caught, wrong-import-order, wrong-import-position
import sys
sys.path.insert(1,'..')
import traceback
import logging
from aiohttp import ClientTimeout, ClientSession
from bs4 import BeautifulSoup
import html as htm
from . import private
from . import common
from lyric_search_new import utils
from lyric_search_new.constructors import LyricsResult
logger = logging.getLogger()
log_level = logging.getLevelName(logger.level)
class InvalidResponseException(Exception):
"""
InvalidResponseException
"""
class Genius:
"""Genius Search Module"""
def __init__(self):
self.label = "Genius"
self.genius_url = private.GENIUS_URL
self.genius_search_url = f'{self.genius_url}api/search/song?q='
self.headers = common.SCRAPE_HEADERS
self.timeout = ClientTimeout(connect=2, sock_read=4)
self.datautils = utils.DataUtils()
self.matcher = utils.TrackMatcher()
async def search(self, artist: str, song: str):
"""
@artist: the artist to search
@song: the song to search
"""
try:
logging.info("Searching %s - %s on %s",
artist, song, self.label)
artist = artist.strip().lower()
song = song.strip().lower()
search_term = f'{artist}%20{song}'
returned_lyrics = ''
async with ClientSession() as client:
async with client.get(f'{self.genius_search_url}{search_term}',
timeout=self.timeout,
headers=self.headers) as request:
request.raise_for_status()
text = await request.text()
if len(text) < 100:
raise InvalidResponseException("Search response text was invalid (len < 100 chars.)")
search_data = await request.json()
if not isinstance(search_data, dict):
raise InvalidResponseException("Invalid JSON.")
if not isinstance(search_data['response'], dict):
raise InvalidResponseException(f"Invalid JSON: Cannot find response key.\n{search_data}")
if not isinstance(search_data['response']['sections'], list):
raise InvalidResponseException(f"Invalid JSON: Cannot find response->sections key.\n{search_data}")
if not isinstance(search_data['response']['sections'][0]['hits'], list):
raise InvalidResponseException("Invalid JSON: Cannot find response->sections[0]->hits key.")
possible_matches = search_data['response']['sections'][0]['hits']
to_scrape = [
(
returned['result']['path'],
f'{returned['result']['artist_names']} - {returned['result']['title']}',
) for returned in possible_matches
]
searched = f"{artist} - {song}"
best_match = self.matcher.find_best_match(input_track=searched,
candidate_tracks=to_scrape)
((scrape_stub, track), confidence) = best_match
scrape_url = f'{self.genius_url}{scrape_stub[1:]}'
async with client.get(scrape_url,
timeout=self.timeout,
headers=self.headers) as scrape_request:
scrape_request.raise_for_status()
scrape_text = await scrape_request.text()
if len(scrape_text) < 100:
raise InvalidResponseException("Scrape response was invalid (len < 100 chars.)")
html = BeautifulSoup(htm.unescape(scrape_text).replace('<br/>', '\n'), "html.parser")
divs = html.find_all("div", {"data-lyrics-container": "true"})
if not divs:
return
for div in divs:
returned_lyrics += div.get_text()
returned_lyrics = self.datautils.scrub_lyrics(returned_lyrics)
artist = track.split(" - ", maxsplit=1)[0]
song = track.split(" - ", maxsplit=1)[1]
logging.info("Result found on %s", self.label)
return LyricsResult(artist=artist,
song=song,
src=self.label,
lyrics=returned_lyrics,
confidence=confidence)
except:
if log_level == "DEBUG":
traceback.print_exc()
return