#!/usr/bin/env python3.12 # pylint: disable=bare-except, broad-exception-caught, invalid-name import time from fastapi import FastAPI 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, glob_state, radio): # pylint: disable=super-init-not-called self.app = app self.util = my_util self.constants = constants self.glob_state = glob_state self.radio_pubkey: str = "XC-AJCJS89-AOLOFKZ92921AK-AKASKZJAN178-3D1" self.lyr_cache = LyricsCache.Cache() self.redis_cache = redis_cache.RedisCache() self.redis_client = redis.Redis(password=private.REDIS_PW) self.radio = radio self.endpoints = { "widget/redis": self.homepage_redis_widget, "widget/sqlite": self.homepage_sqlite_widget, "widget/lyrics": self.homepage_lyrics_widget, "widget/radio": self.homepage_radio_widget, } for endpoint, handler in self.endpoints.items(): app.add_api_route(f"/{endpoint}", handler, methods=["GET"], include_in_schema=False) async def get_radio_np(self) -> dict: """ Get radio now playing Uses XC endpoint Args: None Returns: str: Radio now playing in artist - song format """ json_payload = { 'bid': 0, 'cmd': 'radio_metadata', 'key': f'Bearer {self.radio_pubkey}', } headers = { 'content-type': 'application/json; charset=utf-8', } artistsong = self.radio.now_playing['artistsong'] if not isinstance(artistsong, str): return "N/A - N/A" return artistsong async def homepage_redis_widget(self) -> dict: """ /widget/redis Homepage Redis Widget Handler Args: None Returns: dict """ # 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 = index_info.get('num_docs') return { 'responseTime': round(response_time, 7), 'storedKeys': total_keys, 'indexedLyrics': indexed_lyrics, 'sessions': num_ci_keys, } async def homepage_sqlite_widget(self) -> dict: """ /widget/sqlite Homepage SQLite Widget Handler Args: None Returns: dict """ row_count = await self.lyr_cache.sqlite_rowcount() distinct_artists = await self.lyr_cache.sqlite_distinct("artist") lyrics_length = await self.lyr_cache.sqlite_lyrics_length() return { 'storedRows': row_count, 'distinctArtists': distinct_artists, 'lyricsLength': lyrics_length, } async def homepage_lyrics_widget(self) -> dict: """ /widget/lyrics Homepage Lyrics Widget Handler Args: None Returns: dict """ counts = await self.redis_cache.get_found_counts() return counts async def homepage_radio_widget(self) -> dict: """ /widget/radio Homepage Radio Widget Handler Args: None Returns: dict """ return { 'now_playing': await self.get_radio_np(), }