109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
import logging
|
|
import time
|
|
from typing import Optional
|
|
from fastapi import FastAPI
|
|
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.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,
|
|
}
|
|
|
|
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) -> str:
|
|
"""
|
|
Get radio now playing
|
|
Args:
|
|
None
|
|
Returns:
|
|
str: Radio now playing in artist - song format
|
|
"""
|
|
|
|
artistsong: Optional[str] = self.radio.radio_util.now_playing['artistsong']
|
|
if not isinstance(artistsong, str):
|
|
return "N/A - N/A"
|
|
return artistsong
|
|
|
|
|
|
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: str = await self.get_radio_np()
|
|
if not radio_np:
|
|
return JSONResponse(status_code=500, content={
|
|
'err': True,
|
|
'errorText': 'General failure.',
|
|
})
|
|
return JSONResponse(content={
|
|
'now_playing': await self.get_radio_np(),
|
|
}) |