api/lyric_search_new/sources/genius.py

#!/usr/bin/env python3.12
# pylint: disable=bare-except, broad-exception-caught, wrong-import-position

import sys
sys.path.insert(1,'..')
import traceback
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

class InvalidResponseException(Exception):
    """
    """

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:
            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]
                        return LyricsResult(artist=artist,
                                            song=song,
                                            src=self.label,
                                            lyrics=returned_lyrics,
                                            confidence=confidence)

        except:
            traceback.print_exc()
            return