api/endpoints/lrclib.py

import urllib.parse
from fastapi import FastAPI, HTTPException, Depends
from fastapi_throttle import RateLimiter
from fastapi.responses import JSONResponse
from typing import Type, Optional
from sqlalchemy import (
    and_,
    true,
    Column,
    Integer,
    String,
    Float,
    Boolean,
    DateTime,
    ForeignKey,
    UniqueConstraint,
    create_engine,
)
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.orm import sessionmaker
from .constructors import ValidLRCLibRequest
from lyric_search.constructors import LRCLibResult
from lyric_search import notifier
from sqlalchemy.orm import foreign

Base: Type[DeclarativeMeta] = declarative_base()


class Tracks(Base):  # type: ignore
    __tablename__ = "tracks"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    name_lower = Column(String, index=True)
    artist_name = Column(String)
    artist_name_lower = Column(String, index=True)
    album_name = Column(String)
    album_name_lower = Column(String, index=True)
    duration = Column(Float, index=True)
    last_lyrics_id = Column(Integer, ForeignKey("lyrics.id"), index=True)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)

    # Relationships
    lyrics = relationship(
        "Lyrics",
        back_populates="track",
        foreign_keys=[last_lyrics_id],
        primaryjoin="Tracks.id == foreign(Lyrics.track_id)",  # Use string reference for Lyrics
    )

    # Constraints
    __table_args__ = (
        UniqueConstraint(
            "name_lower",
            "artist_name_lower",
            "album_name_lower",
            "duration",
            name="uq_tracks",
        ),
    )


class Lyrics(Base):  # type: ignore
    __tablename__ = "lyrics"

    id = Column(Integer, primary_key=True, autoincrement=True)
    plain_lyrics = Column(String)
    synced_lyrics = Column(String)
    track_id = Column(Integer, ForeignKey("tracks.id"), index=True)
    has_plain_lyrics = Column(Boolean, index=True)
    has_synced_lyrics = Column(Boolean, index=True)
    instrumental = Column(Boolean)
    source = Column(String, index=True)
    created_at = Column(DateTime, index=True)
    updated_at = Column(DateTime)

    # Relationships
    track = relationship(
        "Tracks",
        back_populates="lyrics",
        foreign_keys=[track_id],
        primaryjoin=(Tracks.id == foreign(track_id)),
        remote_side=Tracks.id,
    )


DATABASE_URL: str = "sqlite:////nvme/sqlite_dbs/lrclib.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


"""
TODO:
     - Move retrieval to lyric_search.sources, with separate file for DB Model
"""


class LRCLib(FastAPI):
    """
    LRCLib Cache Search Endpoint
    """

    def __init__(self, app: FastAPI, util, constants) -> None:
        """Initialize LyricSearch endpoints."""
        self.app: FastAPI = app
        self.util = util
        self.constants = constants
        self.declarative_base = declarative_base()
        self.notifier = notifier.DiscordNotifier()

        self.endpoints: dict = {
            "lrclib/search": self.lyric_search_handler,
        }

        for endpoint, handler in self.endpoints.items():
            times: int = 20
            seconds: int = 2
            rate_limit: tuple[int, int] = (2, 3)  # Default; (Times, Seconds)
            (times, seconds) = rate_limit

            app.add_api_route(
                f"/{endpoint}",
                handler,
                methods=["POST"],
                include_in_schema=True,
                dependencies=[Depends(RateLimiter(times=times, seconds=seconds))],
            )

    async def lyric_search_handler(
        self, data: ValidLRCLibRequest, db: Session = Depends(get_db)
    ) -> JSONResponse:
        """
        Search for lyrics.

        Parameters:
        - **data** (ValidLRCLibRequest): Request containing artist, song, and other parameters.

        Returns:
        - **JSONResponse**: LRCLib data or error.
        """
        if not data.artist or not data.song:
            raise HTTPException(detail="Invalid request", status_code=500)

        search_artist: str = urllib.parse.unquote(data.artist).lower()
        search_song: str = urllib.parse.unquote(data.song).lower()
        search_duration: Optional[int] = data.duration

        if not isinstance(search_artist, str) or not isinstance(search_song, str):
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Invalid request",
                },
            )

        query = (
            db.query(
                Tracks.id.label("id"),
                Tracks.artist_name.label("artist"),
                Tracks.name.label("song"),
                Lyrics.plain_lyrics.label("plainLyrics"),
                Lyrics.synced_lyrics.label("syncedLyrics"),
            )
            .join(Lyrics, Tracks.id == Lyrics.track_id)
            .filter(
                and_(
                    Tracks.artist_name_lower == search_artist,
                    Tracks.name == search_song,
                    Tracks.duration == search_duration if search_duration else true(),
                )
            )
        )

        db_result = query.first()
        if not db_result:
            return JSONResponse(
                status_code=404, content={"err": True, "errorText": "No result found."}
            )

        result = LRCLibResult(
            id=db_result.id,
            artist=db_result.artist,
            song=db_result.song,
            plainLyrics=db_result.plainLyrics,
            syncedLyrics=db_result.syncedLyrics,
        )

        return JSONResponse(content=vars(result))