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, })