typeahead

import importlib

import traceback

import logging

import os

import urllib.parse

import regex

import aiohttp

import aiosqlite as sqlite3

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel 

from lyric_search_new.sources import aggregate 

                    }

            }

class ValidTypeAheadRequest(BaseModel):

    """

    - **query**: query string

    """

    pre_query: str|None = None

    query: str

class ValidLyricSearchLogRequest(BaseModel):

    """

    webradio: bool = False

class CacheUtils:

    """Lyrics Cache DB Utils"""

    def __init__(self):

        self.lyrics_db_path = os.path.join("/", "var", "lib",

                                        "singerdbs", "cached_lyrics.db")    

    async def check_typeahead(self, s: str, pre_query: str | None = None):

        """Check s against artists stored - for typeahead"""

        async with sqlite3.connect(self.lyrics_db_path,

                                   timeout=2) as db_conn:

            db_conn.row_factory = lambda c, r: dict([(col[0], r[idx]) for idx, col in enumerate(c.description)])

            if not pre_query:

                query = "SELECT distinct(artist) FROM lyrics WHERE artist LIKE ? LIMIT 15"

                query_params = (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 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, glob_state): # pylint: disable=super-init-not-called

        self.util = util

        self.constants = constants

        self.glob_state = glob_state

        self.cache_utils = CacheUtils()

        self.lyrics_engine = importlib.import_module("lyrics_engine").LyricsEngine()

        self.endpoint_name = "lyric_search"

        self.endpoint2_name = "lyric_cache_list"

        self.endpoints = {

            "typeahead/artist": self.artist_typeahead_handler,

            "typeahead/song": self.song_typeahead_handler,

            "lyric_search": self.lyric_search_handler,

            "lyric_cache_list": self.lyric_cache_list_handler,

            "lyric_search_history": self.lyric_search_log_handler,

            'data': await self.lyrics_engine.listCacheEntries()

        }

    async def artist_typeahead_handler(self, data: ValidTypeAheadRequest):

        """Artist Type  Ahead Handler"""

        if not isinstance(data.query, str) or len(data.query) < 2:

            return {

                'err': True,

                'errorText': 'Invalid request',

            }

        query = data.query

        typeahead_result = await self.cache_utils.check_typeahead(query)

        typeahead_list = [str(r.get('artist')) for r in typeahead_result]

        return typeahead_list

    async def song_typeahead_handler(self, data: ValidTypeAheadRequest):

        """Song Type Ahead Handler"""

        if not isinstance(data.pre_query, str) or len(data.pre_query) < 2\

            or not isinstance(data.query, str) or len(data.query) < 2:

            return {

                'err': True,

                'errorText': 'Invalid request',

            }

        pre_query = data.pre_query

        query = data.query

        typeahead_result = await self.cache_utils.check_typeahead(query, pre_query)

        typeahead_list = [str(r.get('song')) for r in typeahead_result]

        return typeahead_list

    async def lyric_search_log_handler(self, data: ValidLyricSearchLogRequest):

        """Lyric Search Log Handler"""

        include_radio = data.webradio