#!/usr/bin/env python3.12 import importlib import urllib.parse import regex from fastapi import FastAPI, HTTPException from pydantic import BaseModel 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] - **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 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 } } 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] - **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 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 } } 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 } self.acceptable_request_sources = [ "WEB", "WEB-RADIO", "IRC-MS", "IRC-FS", "IRC-KALI", "DISC-ACES", "DISC-HAVOC", "IRC-SHARED" ] 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): 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 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] - **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 """ 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: 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)) search_worker = await self.lyrics_engine.lyrics_worker(searching=search_object, recipient='anyone') 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", "
", " ".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') }