api/endpoints/misc.py

import logging
import time
import os
import json
import random
from typing import Any, Optional, Annotated
from fastapi import FastAPI, Request, UploadFile, Response, HTTPException, Form, Depends
from fastapi_throttle import RateLimiter
from fastapi.responses import JSONResponse
import redis
from rq import Queue
from rq.registry import (
    StartedJobRegistry,
    FinishedJobRegistry,
    FailedJobRegistry,
    DeferredJobRegistry,
)
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:
        """Initialize Misc endpoints."""
        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: Any = redis.Redis(password=private.REDIS_PW)
        self.radio = radio
        self.activity_image: Optional[bytes] = None
        self.nos_json_path: str = os.path.join(
            "/", "usr", "local", "share", "naas", "reasons.json"
        )
        self.nos: list[str] = []
        self.last_5_nos: list[str] = []
        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,
            "widget/rq": self.homepage_rq_widget,
            "misc/get_activity_image": self.get_activity_image,
            "misc/no": self.no,
        }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(
                f"/{endpoint}",
                handler,
                methods=["GET"],
                include_in_schema=True,
                dependencies=[Depends(RateLimiter(times=10, seconds=2))],
            )

        app.add_api_route(
            "/misc/upload_activity_image",
            self.upload_activity_image,
            methods=["POST"],
            dependencies=[Depends(RateLimiter(times=10, seconds=2))],
        )

        logging.debug("Loading NaaS reasons")

        with open(self.nos_json_path, "r", encoding="utf-8") as f:
            self.nos = json.loads(f.read())

        logging.debug("Loaded %s reasons", len(self.nos))

    def get_no(self) -> str:
        try:
            no = random.choice(self.nos)
            if no in self.last_5_nos:
                return self.get_no()  # recurse
            self.last_5_nos.append(no)
            if len(self.last_5_nos) >= 5:
                self.last_5_nos.pop(0)
            return no
        except Exception as e:
            logging.debug("Exception: %s", str(e))
            return "No."

    async def no(self) -> JSONResponse:
        """
        Get a random 'no' reason.

        Returns:
        - **JSONResponse**: Contains a random 'no' reason.
        """

        return JSONResponse(content={"no": self.get_no()})

    async def upload_activity_image(
        self, image: UploadFile, key: Annotated[str, Form()], request: Request
    ) -> Response:
        """
        Upload activity image.

        Parameters:
        - **image** (UploadFile): The uploaded image file.
        - **key** (str): The API key for authentication.
        - **request** (Request): The HTTP request object.

        Returns:
        - **Response**: Indicates success or failure of the upload.
        """
        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:
        """
        Get the activity image.

        Parameters:
        - **request** (Request): The HTTP request object.

        Returns:
        - **Response**: The activity image or a fallback image.
        """
        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, station: str = "main") -> tuple[str, str, str]:
        """
        Get radio now playing info.

        Args:
            station: Station name.

        Returns:
            Tuple of (artistsong, album, genre).
        """

        np: dict = self.radio.radio_util.now_playing[station]
        artistsong: str = "N/A - N/A"
        artist = np.get("artist")
        song = np.get("song")
        if artist and song:
            artistsong = f"{artist} - {song}"

        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:
        """
        Get Redis stats for homepage widget.

        Returns:
        - **JSONResponse**: Contains Redis stats.
        """
        # Measure response time w/ test lyric search
        time_start: float = time.time()  # Start time for response_time
        test_lyrics_result = self.redis_client.ft().search(  # noqa: F841
            "@artist: test @song: test"
        )
        time_end: float = time.time()
        # End response time test
        total_keys = self.redis_client.dbsize()
        response_time: float = time_end - time_start
        index_info = 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": -1,
            }
        )

    async def homepage_rq_widget(self) -> JSONResponse:
        """
        Get RQ job stats for homepage widget.

        Returns:
        - **JSONResponse**: Contains RQ job stats.
        """
        queue_name = "dls"
        queue = Queue(queue_name, self.redis_client)
        queued = queue.count
        started = StartedJobRegistry(queue_name, connection=self.redis_client).count
        failed = FailedJobRegistry(queue_name, connection=self.redis_client).count
        finished = FinishedJobRegistry(queue_name, connection=self.redis_client).count
        deferred = DeferredJobRegistry(queue_name, connection=self.redis_client).count

        return JSONResponse(
            content={
                queue_name: {
                    "queued": queued,
                    "started": started,
                    "failed": failed,
                    "finished": finished,
                    "deferred": deferred,
                }
            }
        )

    async def homepage_sqlite_widget(self) -> JSONResponse:
        """
        Get SQLite stats for homepage widget.

        Returns:
        - **JSONResponse**: Contains SQLite stats.
        """
        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:
        """
        Get lyrics stats for homepage widget.

        Returns:
        - **JSONResponse**: Contains lyrics stats.
        """
        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, station: str = "main") -> JSONResponse:
        """
        Get radio now playing for homepage widget.

        Parameters:
        - **station** (str): Station name. Defaults to "main".

        Returns:
        - **JSONResponse**: Contains now playing info.
        """
        radio_np: tuple = await self.get_radio_np(station)
        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,
            }
        )