codey/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

misc

Browse Source
This commit is contained in:
codey
2025-04-26 19:47:12 -04:00
parent 6c29c6fede
commit 58ba471b5e
10 changed files with 94 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
endpoints/karma.py
View File

@ -84,7 +84,9 @@ class KarmaDB:
"INSERT INTO karma(keyword, score, last_change) VALUES(?, ?, ?)"
)
friendly_flag: str = "++" if not flag else "--"
audit_message: str = f"{granter} adjusted karma for {keyword} @ {datetime.datetime.now().isoformat()}: {friendly_flag}"
audit_message: str = (
f"{granter} adjusted karma for {keyword} @ {datetime.datetime.now().isoformat()}: {friendly_flag}"
)
audit_query: str = (
"INSERT INTO karma_audit(impacted_keyword, comment) VALUES(?, ?)"
)

2
endpoints/lyric_search.py
View File

@ -115,7 +115,7 @@ class LyricSearch(FastAPI):
if data.src.upper() not in self.acceptable_request_sources:
await self.notifier.send(
f"ERROR @ {__file__.rsplit('/', maxsplit=1)[-1]}",
f"ERROR @ {__file__.rsplit("/", maxsplit=1)[-1]}",
f"Unknown request source: {data.src}",
)
return JSONResponse(

41
endpoints/radio.py
View File

@ -21,11 +21,12 @@ from fastapi.responses import RedirectResponse, JSONResponse
class Radio(FastAPI):
"""Radio Endpoints"""
def __init__(self, app: FastAPI, my_util, constants) -> None:
def __init__(self, app: FastAPI, my_util, constants, loop) -> None:
self.app: FastAPI = app
self.util = my_util
self.constants = constants
self.radio_util = radio_util.RadioUtil(self.constants)
self.loop = loop
self.radio_util = radio_util.RadioUtil(self.constants, self.loop)
self.playlist_loaded: bool = False
self.endpoints: dict = {
@ -57,10 +58,7 @@ class Radio(FastAPI):
async def on_start(self) -> None:
logging.info("radio: Initializing")
with Pool() as pool:
res = pool.apply_async(self.radio_util.load_playlist)
if res:
await self.radio_util._ls_skip()
self.loop.run_in_executor(None, self.radio_util.load_playlist)
async def radio_skip(
self, data: ValidRadioNextRequest, request: Request
@ -86,8 +84,8 @@ class Radio(FastAPI):
self.radio_util.active_playlist = self.radio_util.active_playlist[
queue_item[0] :
]
if not self.radio_util.active_playlist:
await self.radio_util.load_playlist()
# if not self.radio_util.active_playlist:
# self.radio_util.load_playlist()
skip_result: bool = await self.radio_util._ls_skip()
status_code = 200 if skip_result else 500
return JSONResponse(
@ -154,12 +152,15 @@ class Radio(FastAPI):
"duration": item.get("duration"),
}
)
fallback_playlist_len: int = len(
self.radio_util.active_playlist
) # Used if search term is provided
out_json = {
"draw": data.draw,
"recordsTotal": len(queue_full),
"recordsFiltered": (
len(queue_full) if not data.search else len(queue_full)
), # todo: implement search
"recordsTotal": (
len(queue_full) if not data.search else fallback_playlist_len
),
"recordsFiltered": (len(queue_full)),
"items": queue_out,
}
return JSONResponse(content=out_json)
@ -236,7 +237,7 @@ class Radio(FastAPI):
if not track_id:
track_id = self.radio_util.now_playing.get("id")
logging.debug("Seeking album art with trackId: %s", track_id)
album_art: Optional[bytes] = await self.radio_util.get_album_art(
album_art: Optional[bytes] = self.radio_util.get_album_art(
track_id=track_id
)
if not album_art:
@ -312,7 +313,9 @@ class Radio(FastAPI):
if len(self.radio_util.active_playlist) > 1:
self.radio_util.active_playlist.append(next) # Push to end of playlist
else:
await self.radio_util.load_playlist()
with Pool() as pool:
pool.apply_async(self.radio_util.load_playlist())
pool.close()
self.radio_util.now_playing = next
next["start"] = time_started
@ -323,9 +326,9 @@ class Radio(FastAPI):
logging.info("radio_get_next Exception: %s", str(e))
traceback.print_exc()
try:
album_art = await self.radio_util.get_album_art(track_id=next["id"])
album_art = self.radio_util.get_album_art(track_id=next["id"])
if not album_art:
await self.radio_util.cache_album_art(next["id"], next["file_path"])
self.radio_util.cache_album_art(next["id"], next["file_path"])
except Exception as e:
logging.info("radio_get_next Exception: %s", str(e))
traceback.print_exc()
@ -364,7 +367,7 @@ class Radio(FastAPI):
},
)
search: bool = await self.radio_util.search_playlist(
search: bool = self.radio_util.search_playlist(
artistsong=artistsong, artist=artist, song=song
)
if data.alsoSkip:
@ -386,9 +389,7 @@ class Radio(FastAPI):
"errorText": "Invalid request.",
},
)
typeahead: Optional[list[str]] = await self.radio_util.trackdb_typeahead(
data.query
)
typeahead: Optional[list[str]] = self.radio_util.trackdb_typeahead(data.query)
if not typeahead:
return JSONResponse(content=[])
return JSONResponse(content=typeahead)

2
endpoints/rand_msg.py
View File

@ -35,7 +35,7 @@ class RandMsg(FastAPI):
short = data.short
if short:
db_rand_selected: int = 9
db_rand_selected = random.choice([0, 1, 3])
db_rand_selected = random.choice([3])
title_attr: str = "Unknown"
match db_rand_selected:

4
endpoints/transcriptions.py
View File

@ -113,7 +113,9 @@ class Transcriptions(FastAPI):
db_path: Union[str, LiteralString] = os.path.join(
"/usr/local/share", "sqlite_dbs", "sp.db"
)
db_query: str = """SELECT ("S" || Season || "E" || Episode || " " || Title), Character, Line FROM SP_DAT WHERE ID = ?"""
db_query: str = (
"""SELECT ("S" || Season || "E" || Episode || " " || Title), Character, Line FROM SP_DAT WHERE ID = ?"""
)
case 1:
db_path = os.path.join("/usr/local/share", "sqlite_dbs", "futur.db")
db_query = """SELECT ("S" || EP_S || "E" || EP_EP || " " || EP_TITLE || "<br><em>Opener: " || EP_OPENER || "</em>"), EP_LINE_SPEAKER, EP_LINE FROM clean_dialog WHERE EP_ID = ? ORDER BY LINE_ID ASC"""