api/lyric_search_new/sources/cache.py

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

import os
import logging
import sys
import traceback
sys.path.insert(1,'..')
sys.path.insert(1,'.')
from typing import Optional
import aiosqlite as sqlite3
from lyric_search_new import utils
from lyric_search_new.constructors import LyricsResult

logger = logging.getLogger()
log_level = logging.getLevelName(logger.level)

class Cache:
    """Cache Search Module"""
    def __init__(self):
        self.cache_db: str = os.path.join("/", "var",
                                     "lib", "singerdbs",
                                     "cached_lyrics.db")
        
        self.cache_pre_query: str = "pragma journal_mode = WAL; pragma synchronous = normal; pragma temp_store = memory; pragma mmap_size = 30000000000;"
        self.sqlite_exts: list[str] = ['/usr/local/lib/python3.11/dist-packages/spellfix1.cpython-311-x86_64-linux-gnu.so']
        self.label: str = "Cache"

    def get_matched(self, sqlite_rows: list[sqlite3.Row], matched_candidate: tuple, confidence: float) -> Optional[LyricsResult]:
        """Get Matched Result"""
        matched_id: int = matched_candidate[0]
        for row in sqlite_rows:
            if row[0] == matched_id:
                (_id, artist, song, lyrics, original_src, _confidence) = row
                return LyricsResult(
                    artist=artist,
                    song=song,
                    lyrics=lyrics,
                    src=f"{original_src} (cached, id: {_id})",
                    confidence=confidence)
        return None

    async def search(self, artist: str, song: str) -> Optional[LyricsResult]:
        """
        @artist: the artist to search
        @song: the song to search
        Returns:
            - LyricsResult corresponding to nearest match found (if found), **None** otherwise
        """
        try:
            artist: str = artist.strip().lower() 
            song: str  = song.strip().lower()
            search_params: Optional[tuple] = None
            random_search: bool = False
            logging.info("Searching %s - %s on %s",
                         artist, song, self.label)            
            async with sqlite3.connect(self.cache_db, timeout=2) as db_conn:
                await db_conn.enable_load_extension(True)
                for ext in self.sqlite_exts:
                    await db_conn.load_extension(ext)
                async with await db_conn.executescript(self.cache_pre_query) as _db_cursor:
                    search_query: str = 'SELECT id, artist, song, lyrics, src, confidence FROM lyrics\
                        WHERE editdist3((artist || " " || song), (? || " " || ?))\
                        <= 410 ORDER BY editdist3((artist || " " || song), ?) ASC LIMIT 10'
                    search_params: tuple = (artist.strip(), song.strip(),
                                    f"{artist.strip()} {song.strip()}")                    
                    if artist == "!" and song == "!":
                        random_search: bool = True
                        search_query: str = 'SELECT id, artist, song, lyrics, src, confidence FROM lyrics ORDER BY RANDOM() LIMIT 1'
                        search_params = None
                    async with await _db_cursor.execute(search_query, search_params) as db_cursor:
                        results: list = await db_cursor.fetchall()
                        result_tracks: list = []
                        for track in results:
                            (_id, _artist, _song, _lyrics, _src, _confidence) = track
                            result_tracks.append((_id, f"{_artist} - {_song}"))
                        input_track: str = f"{artist} - {song}"
                        matcher = utils.TrackMatcher()
                        if not random_search:
                            best_match: tuple|None = matcher.find_best_match(input_track=input_track,
                                                            candidate_tracks=result_tracks)
                        else:
                            best_match = (result_tracks[0], float(1))
                        if not best_match:
                            return None
                        (candidate, confidence) = best_match
                        logging.info("Result found on %s", self.label)
                        return self.get_matched(sqlite_rows=results,
                                                matched_candidate=candidate,
                                                confidence=confidence)
        except:
            if log_level == "DEBUG":
                traceback.print_exc()
            return