api/endpoints/misc.py

import logging
import time
import os
from typing import Optional, Annotated
from fastapi import FastAPI, Request, UploadFile, Response, HTTPException, Form
from fastapi.responses import JSONResponse
import redis.asyncio as redis
from lyric_search.sources import private, cache as LyricsCache, redis_cache


class Misc(FastAPI):
    """
    Misc Endpoints
    """

    def __init__(self, app: FastAPI, my_util, constants, radio) -> None:
        self.app: FastAPI = app
        self.util = my_util
        self.constants = constants
        self.lyr_cache = LyricsCache.Cache()
        self.redis_cache = redis_cache.RedisCache()
        self.redis_client = redis.Redis(password=private.REDIS_PW)
        self.radio = radio
        self.activity_image: Optional[bytes] = None
        self.endpoints: dict = {
            "widget/redis": self.homepage_redis_widget,
            "widget/sqlite": self.homepage_sqlite_widget,
            "widget/lyrics": self.homepage_lyrics_widget,
            "widget/radio": self.homepage_radio_widget,
            "misc/get_activity_image": self.get_activity_image,
        }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(
                f"/{endpoint}", handler, methods=["GET"], include_in_schema=True
            )

        app.add_api_route(
            "/misc/upload_activity_image", self.upload_activity_image, methods=["POST"]
        )

    async def upload_activity_image(
        self, image: UploadFile, key: Annotated[str, Form()], request: Request
    ) -> Response:
        if (
            not key
            or not isinstance(key, str)
            or not self.util.check_key(path=request.url.path, req_type=2, key=key)
        ):
            raise HTTPException(status_code=403, detail="Unauthorized")
        if not image:
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Invalid request",
                },
            )
        self.activity_image = await image.read()
        return JSONResponse(
            content={
                "success": True,
            }
        )

    async def get_activity_image(self, request: Request) -> Response:
        if isinstance(self.activity_image, bytes):
            return Response(content=self.activity_image, media_type="image/png")

        # Fallback
        fallback_path = os.path.join(
            "/var/www/codey.lol/public", "images", "plex_placeholder.png"
        )

        with open(fallback_path, "rb") as f:
            return Response(content=f.read(), media_type="image/png")

    async def get_radio_np(self) -> tuple[str, str, str]:
        """
        Get radio now playing
        Args:
            None
        Returns:
            str: Radio now playing in artist - song format
        """

        np: dict = self.radio.radio_util.now_playing
        artistsong: str = np.get("artistsong", "N/A - N/A")
        album: str = np.get("album", "N/A")
        genre: str = np.get("genre", "N/A")
        return (artistsong, album, genre)

    async def homepage_redis_widget(self) -> JSONResponse:
        """
        Homepage Redis Widget Handler
        """
        # Measure response time w/ test lyric search
        time_start: float = time.time()  # Start time for response_time
        test_lyrics_result = await self.redis_client.ft().search(
            "@artist: test @song: test"
        )
        time_end: float = time.time()
        # End response time test
        total_keys = await self.redis_client.dbsize()
        response_time: float = time_end - time_start
        (_, ci_keys) = await self.redis_client.scan(
            cursor=0, match="ci_session*", count=10000000
        )
        num_ci_keys = len(ci_keys)
        index_info = await self.redis_client.ft().info()
        indexed_lyrics: int = index_info.get("num_docs")
        return JSONResponse(
            content={
                "responseTime": round(response_time, 7),
                "storedKeys": total_keys,
                "indexedLyrics": indexed_lyrics,
                "sessions": num_ci_keys,
            }
        )

    async def homepage_sqlite_widget(self) -> JSONResponse:
        """
        Homepage SQLite Widget Handler
        """
        row_count: int = await self.lyr_cache.sqlite_rowcount()
        distinct_artists: int = await self.lyr_cache.sqlite_distinct("artist")
        lyrics_length: int = await self.lyr_cache.sqlite_lyrics_length()
        return JSONResponse(
            content={
                "storedRows": row_count,
                "distinctArtists": distinct_artists,
                "lyricsLength": lyrics_length,
            }
        )

    async def homepage_lyrics_widget(self) -> JSONResponse:
        """
        Homepage Lyrics Widget Handler
        """
        found_counts: Optional[dict] = await self.redis_cache.get_found_counts()
        if not isinstance(found_counts, dict):
            logging.info(
                "DEBUG: Type of found counts from redis: %s\nContents: %s",
                type(found_counts),
                found_counts,
            )
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "General failure.",
                },
            )
        return JSONResponse(content=found_counts)

    async def homepage_radio_widget(self) -> JSONResponse:
        """
        Homepage Radio Widget Handler
        """
        radio_np: tuple = await self.get_radio_np()
        if not radio_np:
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "General failure.",
                },
            )
        (artistsong, album, genre) = radio_np
        return JSONResponse(
            content={
                "now_playing": artistsong,
                "album": album,
                "genre": genre,
            }
        )