#!/usr/bin/env python3.12 # pylint: disable=bare-except, broad-exception-caught, invalid-name import logging import traceback import time from fastapi import FastAPI from aiohttp import ClientSession, ClientTimeout import redis.asyncio as redis from redis.commands.search.query import Query from redis.commands.search.indexDefinition import IndexDefinition, IndexType from redis.commands.search.field import TextField, TagField from redis.commands.json.path import Path 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): # 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.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', } #TODO: change URL below to dynamically populate based on listener async with ClientSession() as session: async with await session.post("http://127.0.0.1:52111/xc", json=json_payload, headers=headers, timeout=ClientTimeout(connect=3, sock_read=2)) as request: request.raise_for_status() request_json = await request.json() request_json = request_json.get('response') np_artist = request_json.get('artist') np_song = request_json.get('title') if not isinstance(np_artist, str)\ or not isinstance(np_song, str): return "N/A - N/A" return f"{np_artist} - {np_song}" 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(), }