#!/usr/bin/env python3.12 # pylint: disable=bare-except, broad-exception-raised, broad-exception-caught import logging import os import urllib.parse import regex import aiosqlite as sqlite3 from fastapi import FastAPI, HTTPException from typing import LiteralString, Optional, Callable from regex import Pattern from .constructors import ValidTypeAheadRequest, ValidLyricRequest from lyric_search.constructors import LyricsResult from lyric_search.sources import aggregate from lyric_search import notifier class CacheUtils: """Lyrics Cache DB Utils""" def __init__(self) -> None: self.lyrics_db_path: LiteralString = os.path.join("/usr/local/share", "sqlite_dbs", "cached_lyrics.db") async def check_typeahead(self, s: str, pre_query: str | None = None) -> list[dict]: """Check s against artists stored - for typeahead""" async with sqlite3.connect(self.lyrics_db_path, timeout=2) as db_conn: db_conn.row_factory = sqlite3.Row if not pre_query: query: str = "SELECT distinct(artist) FROM lyrics WHERE artist LIKE ? LIMIT 15" query_params: tuple = (f"%{s}%",) else: query = "SELECT distinct(song) FROM lyrics WHERE artist LIKE ? AND song LIKE ? LIMIT 15" query_params = (f"%{pre_query}%", f"%{s}%",) async with await db_conn.execute(query, query_params) as db_cursor: return await db_cursor.fetchall() class LyricSearch(FastAPI): """Lyric Search Endpoint""" def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called self.app = app self.util = util self.constants = constants self.cache_utils = CacheUtils() self.notifier = notifier.DiscordNotifier() self.endpoints: dict = { "typeahead/artist": self.artist_typeahead_handler, "typeahead/song": self.song_typeahead_handler, "lyric_search": self.lyric_search_handler, # Preserving old endpoint path temporarily "lyric/search": self.lyric_search_handler, } self.acceptable_request_sources: list = [ "WEB", "WEB-RADIO", "IRC-MS", "IRC-FS", "IRC-KALI", "DISC-ACES", "DISC-HAVOC", "IRC-SHARED", "LIMNORIA-SHARED", ] self.lrc_regex: Pattern = regex.compile(r'\[([0-9]{2}:[0-9]{2})\.[0-9]{1,3}\](\s(.*)){0,}') for endpoint, handler in self.endpoints.items(): _schema_include = endpoint in ["lyric/search"] app.add_api_route(f"/{endpoint}", handler, methods=["POST"], include_in_schema=_schema_include) async def artist_typeahead_handler(self, data: ValidTypeAheadRequest) -> list[str]|dict: """Artist Type Ahead Handler""" if not isinstance(data.query, str) or len(data.query) < 2: return { 'err': True, 'errorText': 'Invalid request', } query: str = data.query typeahead_result: list[dict] = await self.cache_utils.check_typeahead(query) typeahead_list: list[str] = [str(r.get('artist')) for r in typeahead_result] return typeahead_list async def song_typeahead_handler(self, data: ValidTypeAheadRequest) -> list[str]|dict: """Song Type Ahead Handler""" if not isinstance(data.pre_query, str)\ or not isinstance(data.query, str|None): return { 'err': True, 'errorText': 'Invalid request', } pre_query: str = data.pre_query query: str = data.query typeahead_result: list[dict] = await self.cache_utils.check_typeahead(query, pre_query) typeahead_list: list[str] = [str(r.get('song')) for r in typeahead_result] return typeahead_list async def lyric_search_handler(self, data: ValidLyricRequest) -> dict: """ 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 - **excluded_sources**: sources to exclude [optional, default: none] """ if (not data.a or not data.s) and not data.t or not data.src: raise HTTPException(detail="Invalid request", status_code=500) if data.src.upper() not in self.acceptable_request_sources: await self.notifier.send(f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}", f"Unknown request source: {data.src}") return { 'err': True, 'errorText': f'Unknown request source: {data.src}', } if not data.t: search_artist: Optional[str] = data.a search_song: Optional[str] = data.s else: t_split = data.t.split(" - ", maxsplit=1) search_artist = t_split[0] search_song = t_split[1] if search_artist and search_song: search_artist = str(self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())) search_song = str(self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())) search_artist = urllib.parse.unquote(search_artist) search_song = urllib.parse.unquote(search_song) if not isinstance(search_artist, str) or not isinstance(search_song, str): return { 'err': True, 'errorText': 'Invalid request', } excluded_sources: Optional[list] = data.excluded_sources aggregate_search = aggregate.Aggregate(exclude_methods=excluded_sources) plain_lyrics: bool = not data.lrc result: Optional[LyricsResult|dict] = await aggregate_search.search(search_artist, search_song, plain_lyrics) if not result: return { 'err': True, 'errorText': 'Sources exhausted, lyrics not located.', } result = vars(result) if data.sub and not data.lrc: seeked_found_line: Optional[int] = None lyric_lines: list[str] = result['lyrics'].strip().split(" / ") for i, line in enumerate(lyric_lines): line = regex.sub(r'\u2064', '', line.strip()) if data.sub.strip().lower() in line.strip().lower(): seeked_found_line = i logging.debug("Found %s at %s, match for %s!", line, seeked_found_line, data.sub) # REMOVEME: DEBUG break if not seeked_found_line: return { 'failed_seek': True, } result['lyrics'] = " / ".join(lyric_lines[seeked_found_line:]) result['confidence'] = int(result.get('confidence', 0)) result['time'] = f'{float(result['time']):.4f}' if plain_lyrics: result['lyrics'] = regex.sub(r'(\s/\s|\n)', '
', result['lyrics']).strip() else: # Swap lyrics key for 'lrc' result['lrc'] = result['lyrics'] result.pop('lyrics') if "cache" in result['src']: result['from_cache'] = True if not data.extra: result.pop('src') return result