api/lyric_search/sources/lrclib.py

import time
import logging
from typing import Optional
from sqlalchemy.future import select
from lyric_search import utils
from lyric_search.constructors import LyricsResult
from lyric_search.models import Tracks, Lyrics, AsyncSessionLocal
from . import redis_cache

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


class LRCLib:
    """LRCLib Search Module - Local PostgreSQL Database"""

    def __init__(self) -> None:
        self.label: str = "LRCLib-Cache"
        self.datautils = utils.DataUtils()
        self.matcher = utils.TrackMatcher()
        self.redis_cache = redis_cache.RedisCache()

    async def search(
        self,
        artist: str,
        song: str,
        plain: Optional[bool] = True,
        duration: Optional[int] = None,
    ) -> Optional[LyricsResult]:
        """
        LRCLib Local Database Search
        Args:
            artist (str): the artist to search
            song (str): the song to search
            plain (bool): return plain lyrics (True) or synced lyrics (False)
            duration (int): optional track duration for better matching
        Returns:
            Optional[LyricsResult]: The result, if found - None otherwise.
        """
        try:
            artist = artist.strip().lower()
            song = song.strip().lower()
            time_start: float = time.time()

            logging.info("Searching %s - %s on %s", artist, song, self.label)

            async with AsyncSessionLocal() as db:
                best_match = None
                
                # Try exact match first (fastest)
                result = await db.execute(
                    select(
                        Tracks.artist_name,
                        Tracks.name,
                        Lyrics.plain_lyrics,
                        Lyrics.synced_lyrics,
                    )
                    .join(Lyrics, Tracks.id == Lyrics.track_id)
                    .filter(
                        Tracks.artist_name_lower == artist,
                        Tracks.name_lower == song,
                    )
                    .limit(1)
                )
                best_match = result.first()
                
                # If no exact match, try prefix match (faster than full ILIKE)
                if not best_match:
                    result = await db.execute(
                        select(
                            Tracks.artist_name,
                            Tracks.name,
                            Lyrics.plain_lyrics,
                            Lyrics.synced_lyrics,
                        )
                        .join(Lyrics, Tracks.id == Lyrics.track_id)
                        .filter(
                            Tracks.artist_name_lower.like(f"{artist}%"),
                            Tracks.name_lower.like(f"{song}%"),
                        )
                        .limit(1)
                    )
                    best_match = result.first()
                
                # If still no match, try full ILIKE (slowest)
                if not best_match:
                    result = await db.execute(
                        select(
                            Tracks.artist_name,
                            Tracks.name,
                            Lyrics.plain_lyrics,
                            Lyrics.synced_lyrics,
                        )
                        .join(Lyrics, Tracks.id == Lyrics.track_id)
                        .filter(
                            Tracks.artist_name_lower.ilike(f"%{artist}%"),
                            Tracks.name_lower.ilike(f"%{song}%"),
                        )
                        .limit(1)
                    )
                    best_match = result.first()

                if not best_match:
                    logging.info("No result found on %s", self.label)
                    return None

                returned_artist = best_match.artist_name
                returned_song = best_match.name
                
                if plain:
                    if not best_match.plain_lyrics:
                        logging.info("No plain lyrics available on %s", self.label)
                        return None
                    returned_lyrics = best_match.plain_lyrics
                    returned_lyrics = self.datautils.scrub_lyrics(returned_lyrics)
                    lrc_obj = None
                else:
                    if not best_match.synced_lyrics:
                        logging.info("No synced lyrics available on %s", self.label)
                        return None
                    returned_lyrics = best_match.synced_lyrics
                    lrc_obj = self.datautils.create_lrc_object(returned_lyrics)

                # Calculate match confidence
                input_track = f"{artist} - {song}"
                returned_track = f"{returned_artist} - {returned_song}"
                match_result = self.matcher.find_best_match(
                    input_track=input_track, 
                    candidate_tracks=[(0, returned_track)]
                )
                
                if not match_result:
                    return None
                    
                _matched, confidence = match_result
                
                logging.info("Result found on %s", self.label)
                time_end = time.time()
                time_diff = time_end - time_start
                
                matched = LyricsResult(
                    artist=returned_artist,
                    song=returned_song,
                    src=self.label,
                    lyrics=returned_lyrics if plain else lrc_obj,  # type: ignore
                    confidence=confidence,
                    time=time_diff,
                )
                
                await self.redis_cache.increment_found_count(self.label)
                return matched
                
        except Exception as e:
            logging.error("Exception in %s: %s", self.label, str(e))
            return None