api/endpoints/misc.py

#!/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_new.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"])
            
    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(),
        }