api/endpoints/lyric_search.py

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

import importlib
import traceback
import logging
import urllib.parse
import regex
import aiohttp

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel 
from lyric_search_new.sources import aggregate 


class ValidLyricRequest(BaseModel):
    """
    - **a**: artist
    - **s**: song
    - **t**: track (artist and song combined) [used only if a & s are not used]
    - **extra**: include extra details in response [optional, default: false]
    - **lrc**: Request LRCs?
    - **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional]
    - **src**: the script/utility which initiated the request
    """

    a: str | None = None
    s: str | None = None 
    t: str | None = None
    sub: str | None = None
    extra: bool | None = False
    lrc: bool | None = False
    src: str

    class Config: # pylint: disable=missing-class-docstring too-few-public-methods
        schema_extra = {
                "example": {
                        "a": "eminem",
                        "s": "rap god",
                        "src": "WEB",
                        "extra": True,
                        "lrc": False,
                    }
            }

        
class ValidLyricSearchLogRequest(BaseModel):
    """
    - **webradio**: whether or not to include requests generated automatically by the radio page on codey.lol, defaults to False
    """

    webradio: bool = False

class LyricSearch(FastAPI):
    """Lyric Search Endpoint"""
    def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
        self.app = app
        self.util = util
        self.constants = constants
        self.glob_state = glob_state
        self.lyrics_engine = importlib.import_module("lyrics_engine").LyricsEngine()

        self.endpoint_name = "lyric_search"
        self.endpoint2_name = "lyric_cache_list"

        self.endpoints = {
            "lyric_search": self.lyric_search_handler,
            "lyric_cache_list": self.lyric_cache_list_handler,
            "lyric_search_history": self.lyric_search_log_handler,
            "lyric_search_test": self.new_test,
        }

        self.acceptable_request_sources = [
            "WEB",
            "WEB-RADIO",
            "IRC-MS",
            "IRC-FS",
            "IRC-KALI",
            "DISC-ACES",
            "DISC-HAVOC",
            "IRC-SHARED"
        ]

        self.lrc_regex = regex.compile(r'\[([0-9]{2}:[0-9]{2})\.[0-9]{1,3}\](\s(.*)){0,}')

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(f"/{endpoint}/", handler, methods=["POST"])

    async def lyric_cache_list_handler(self):
        """
        Get currently cached lyrics entries
        """
        return {
            'err': False,
            'data': await self.lyrics_engine.listCacheEntries()
        }
    
    async def lyric_search_log_handler(self, data: ValidLyricSearchLogRequest):
        """Lyric Search Log Handler"""
        include_radio = data.webradio
        await self.glob_state.increment_counter('lyrichistory_requests')
        last_10k_sings = await self.lyrics_engine.getHistory(limit=10000, webradio=include_radio)
        return {
            'err': False,
            'history': last_10k_sings
        }
    
    async def new_test(self, data: ValidLyricRequest):
        """
        Search for lyrics (testing)

        - **a**: artist
        - **s**: song
        - **t**: track (artist and song combined) [used only if a & s are not used] [unused]
        - **extra**: include extra details in response [optional, default: false] [unused]
        - **lrc**: Request LRCs? [unused]
        - **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none] [unused]
        - **src**: the script/utility which initiated the request [unused]
        """

        if not data.a or not data.s:
            raise HTTPException(detail="Invalid request", status_code=500)
        
        aggregate_search = aggregate.Aggregate()
        result = await aggregate_search.search(data.a, data.s)
        return result.dict()
        
        

    async def lyric_search_handler(self, data: ValidLyricRequest):
        """
        Search for lyrics

        - **a**: artist
        - **s**: song
        - **t**: track (artist and song combined) [used only if a & s are not used]
        - **extra**: include extra details in response [optional, default: false]
        - **lrc**: Request LRCs?
        - **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none]
        - **src**: the script/utility which initiated the request
        """

        lrc = data.lrc
        src = data.src.upper()
        if not src in self.acceptable_request_sources:
            raise HTTPException(detail="Invalid request source", status_code=403)

        await self.glob_state.increment_counter('lyric_requests')           

        search_artist = data.a
        search_song = data.s
        search_text = data.t
        add_extras = data.extra
        sub_search = data.sub
        search_object = None

        random_song_requested = (search_artist == "!" and search_song == "!")
        query_valid = (
            not(search_artist is None) and
        not(search_song is None) and
        len(search_artist) >= 1 and
        len(search_song) >= 1 and
        len(search_artist) + len(search_song) >= 3
        )

        if not random_song_requested and (not search_text and not query_valid):
            return {
                "err": True,
                "errorText": "Invalid parameters"
                }
        if search_artist and search_song:
            search_artist = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())
            search_song = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())
            search_artist = urllib.parse.unquote(search_artist)
            search_song = urllib.parse.unquote(search_song)

        if search_text is None:
            # pylint: disable=consider-using-f-string
            search_object = self.lyrics_engine.create_query_object("%s : %s" % (search_artist, search_song)) 
            if sub_search:
                sub_search = regex.sub(r'\s{2,}', ' ', sub_search.strip())
                search_object = self.lyrics_engine.create_query_object("%s : %s : %s" % (search_artist, search_song, sub_search))
        else:
            search_object = self.lyrics_engine.create_query_object(str(search_text))

        if lrc:
            search_worker = await self.lyrics_engine.grabFromSpotify(searching=search_object,
                                                                     lrc=True)
            
            spotify_lyrics_unsynced = True
            if search_worker and search_worker.get('l'):
                for line in search_worker.get('l'):
                    if line.get('timeTag') and line.get('timeTag') != "00:00.00":
                        spotify_lyrics_unsynced = False
            if not search_worker or spotify_lyrics_unsynced:
                # Try LRCLib before failing out
                try:
                    lrclib_api_url = "https://lrclib.net/api/get"
                    sane_artist = urllib.parse.quote_plus(search_artist)
                    sane_track = urllib.parse.quote_plus(search_song)
                    async with aiohttp.ClientSession() as session:
                        async with session.get(f"{lrclib_api_url}?artist_name={sane_artist}&track_name={sane_track}") as request:
                            request.raise_for_status()
                            response_json = await request.json()
                            if not "syncedLyrics" in response_json:
                                raise BaseException("LRCLib Fallback Failed")
                            lrc_content = response_json.get('syncedLyrics')
                            returned_artist = response_json.get('artistName')
                            returned_song = response_json.get('trackName')
                            logging.debug("Synced Lyrics [LRCLib]: %s",
                                          lrc_content)
                            lrc_content_out = []
                            for line in lrc_content.split("\n"):
                                _timetag = None
                                _words = None
                                if not line.strip():
                                    continue
                                reg_helper = regex.findall(self.lrc_regex, line.strip())
                                if not reg_helper:
                                    continue
                                reg_helper = reg_helper[0]
                                logging.debug("Reg helper: %s for line: %s; len: %s",
                                              reg_helper, line, len(reg_helper))
                                _timetag = reg_helper[0]
                                if not reg_helper[1].strip():
                                    _words = "♪"
                                else:
                                    _words = reg_helper[1]
                                lrc_content_out.append({
                                    "timeTag": _timetag,
                                    "words": _words,
                                })

                            return {
                                'err': False,
                                'artist': returned_artist,
                                'song': returned_song,
                                'combo_lev': "N/A",
                                'lrc': lrc_content_out,
                                'from_cache': False,
                                'src': 'Alt LRC SRC',
                                'reqn': await self.glob_state.get_counter('lyric_requests'),
                            }
                except:
                    traceback.print_exc()
                    return {
                        'err': True,
                        'errorText': 'Search failed!',
                    }                            
                            

                return {
                    'err': True,
                    'errorText': 'Search failed!',
                }
        if lrc:
            return {
                'err': False,
                'artist': search_worker['artist'],
                'song': search_worker['song'],
                'combo_lev': search_worker['combo_lev'],
                'lrc': search_worker['l'],
                'from_cache': False,
                'src': search_worker['method'],
                'reqn': await self.glob_state.get_counter('lyric_requests'),
            }
        
        search_worker = await self.lyrics_engine.lyrics_worker(searching=search_object)
            

        if not search_worker or not 'l' in search_worker.keys():
            await self.glob_state.increment_counter('failedlyric_requests')
            return {
                'err': True,
                'errorText': 'Sources exhausted, lyrics not located.'
            }
        

        await self.lyrics_engine.storeHistEntry(artist=search_worker.get('artist'),
                                                song=search_worker.get('song'),
                                                retr_method=search_worker.get('method'),
                                                request_src=src.strip())

        return {
            'err': False,
            'artist': search_worker['artist'],
            'song': search_worker['song'],
            'combo_lev': f'{search_worker['combo_lev']:.2f}',
            'lyrics': regex.sub(r"\s/\s", "<br>", " ".join(search_worker['l'])),
            'from_cache': search_worker['method'].strip().lower().startswith("local cache"),
            'src': search_worker['method'] if add_extras else None,
            'reqn': await self.glob_state.get_counter('lyric_requests')
        }