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

rewrite pending; for now, additional support for multi-station

Browse Source
This commit is contained in:
codey
2025-07-19 21:57:21 -04:00
parent 85182b7d8c
commit 9ce16ba923
5 changed files with 76 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
endpoints/constructors.py
View File

@ -211,6 +211,7 @@ class ValidRadioSongRequest(BaseModel):
song: Optional[str] = None
artistsong: Optional[str] = None
alsoSkip: Optional[bool] = False
station: str = "main"
class ValidRadioTypeaheadRequest(BaseModel):

11
endpoints/lyric_search.py
View File

@ -71,10 +71,12 @@ class LyricSearch(FastAPI):
)
for endpoint, handler in self.endpoints.items():
rate_limit: tuple[int, int] = (2, 3) # Default; (Times, Seconds)
rate_limit: tuple[int, int] = (2, 3) # Default; (Times, Seconds)
_schema_include = endpoint in ["lyric/search"]
if endpoint == "typeahead/lyrics": # More permissive rate limiting for typeahead
if (
endpoint == "typeahead/lyrics"
): # More permissive rate limiting for typeahead
rate_limit = (20, 2)
(times, seconds) = rate_limit
@ -83,10 +85,11 @@ class LyricSearch(FastAPI):
handler,
methods=["POST"],
include_in_schema=_schema_include,
dependencies=[Depends(RateLimiter(times=times, seconds=seconds))] if not endpoint == "typeahead/lyrics" else None
dependencies=[Depends(RateLimiter(times=times, seconds=seconds))]
if not endpoint == "typeahead/lyrics"
else None,
)
async def typeahead_handler(self, data: ValidTypeAheadRequest) -> JSONResponse:
"""
Lyric search typeahead handler

8
endpoints/misc.py
View File

@ -115,7 +115,7 @@ class Misc(FastAPI):
with open(fallback_path, "rb") as f:
return Response(content=f.read(), media_type="image/png")
async def get_radio_np(self) -> tuple[str, str, str]:
async def get_radio_np(self, station: str = "main") -> tuple[str, str, str]:
"""
Get radio now playing
Args:
@ -124,7 +124,7 @@ class Misc(FastAPI):
str: Radio now playing in artist - song format
"""
np: dict = self.radio.radio_util.now_playing
np: dict = self.radio.radio_util.now_playing[station]
artistsong: str = "N/A - N/A"
artist = np.get("artist")
song = np.get("song")
@ -194,11 +194,11 @@ class Misc(FastAPI):
)
return JSONResponse(content=found_counts)
async def homepage_radio_widget(self) -> JSONResponse:
async def homepage_radio_widget(self, station: str = "main") -> JSONResponse:
"""
Homepage Radio Widget Handler
"""
radio_np: tuple = await self.get_radio_np()
radio_np: tuple = await self.get_radio_np(station)
if not radio_np:
return JSONResponse(
status_code=500,

22
endpoints/radio.py
View File

@ -76,6 +76,7 @@ class Radio(FastAPI):
Skip to the next track in the queue, or to uuid specified in skipTo if provided
- **key**: API key
- **skipTo**: Optional UUID to skip to
- **station**: default "main"
"""
try:
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
@ -93,9 +94,7 @@ class Radio(FastAPI):
self.radio_util.active_playlist[data.station] = self.radio_util.active_playlist[data.station][
queue_item[0] :
]
# if not self.radio_util.active_playlist:
# self.radio_util.load_playlist()
skip_result: bool = await self.radio_util._ls_skip()
skip_result: bool = await self.radio_util._ls_skip(data.station)
status_code = 200 if skip_result else 500
return JSONResponse(
status_code=status_code,
@ -122,6 +121,7 @@ class Radio(FastAPI):
"""
Reshuffle the play queue
- **key**: API key
- **station**: default "main"
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@ -187,6 +187,7 @@ class Radio(FastAPI):
- **key**: API key
- **uuid**: UUID to shift
- **next**: Play track next? If False, skips to the track
- **station**: default "main"
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@ -204,7 +205,7 @@ class Radio(FastAPI):
self.radio_util.active_playlist[data.station].pop(x)
self.radio_util.active_playlist[data.station].insert(0, item)
if not data.next:
await self.radio_util._ls_skip()
await self.radio_util._ls_skip(data.station)
return JSONResponse(
content={
"ok": True,
@ -218,6 +219,7 @@ class Radio(FastAPI):
Remove an item from the current play queue
- **key**: API key
- **uuid**: UUID of queue item to remove
- **station**: default "main"
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@ -246,6 +248,7 @@ class Radio(FastAPI):
Get album art, optional parameter track_id may be specified.
Otherwise, current track album art will be pulled.
- **track_id**: Optional, if provided, will attempt to retrieve the album art of this track_id. Current track used otherwise.
- **station**: default "main"
"""
try:
if not track_id:
@ -271,6 +274,7 @@ class Radio(FastAPI):
station: Optional[str] = "main") -> JSONResponse:
"""
Get currently playing track info
- **station**: default "main"
"""
ret_obj: dict = {**self.radio_util.now_playing[station]}
ret_obj["station"] = station
@ -293,7 +297,7 @@ class Radio(FastAPI):
(Track will be removed from the queue in the process.)
- **key**: API key
- **skipTo**: Optional UUID to skip to
- **station**: Station (default: "main")
- **station**: default: "main"
"""
logging.info("Radio get next")
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
@ -337,8 +341,7 @@ class Radio(FastAPI):
next["start"] = time_started
next["end"] = time_ends
try:
if data.station == "main":
background_tasks.add_task(self.radio_util.webhook_song_change, next)
background_tasks.add_task(self.radio_util.webhook_song_change, next, data.station)
except Exception as e:
logging.info("radio_get_next Exception: %s", str(e))
traceback.print_exc()
@ -361,6 +364,7 @@ class Radio(FastAPI):
- **song**: Song to search
- **artistsong**: Optional "Artist - Song" pair to search, in place of artist/song
- **alsoSkip**: If True, skips to the track; otherwise, track will be placed next up in queue
- **station**: default "main"
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@ -385,10 +389,10 @@ class Radio(FastAPI):
)
search: bool = self.radio_util.search_db(
artistsong=artistsong, artist=artist, song=song
artistsong=artistsong, artist=artist, song=song, station=data.station
)
if data.alsoSkip:
await self.radio_util._ls_skip()
await self.radio_util._ls_skip(data.station)
return JSONResponse(content={"result": search})
def radio_typeahead(

83
utils/radio_util.py
View File

@ -378,8 +378,13 @@ class RadioUtil:
continue
if station not in self.active_playlist:
self.active_playlist[station] = []
time_start = time.time()
logging.info("[%s] Running query: %s",
time_start, db_query)
db_cursor = db_conn.execute(db_query)
results: list[sqlite3.Row] = db_cursor.fetchall()
time_end = time.time()
logging.info("[%s] Query completed; Time taken: %s", time_end, (time_end - time_start))
self.active_playlist[station] = [
{
"uuid": str(uuid().hex),
@ -402,7 +407,8 @@ class RadioUtil:
station, len(self.active_playlist[station]),
)
random.shuffle(self.active_playlist[station])
if not station == "rock":# REMOVE ME, AFI RELATED
random.shuffle(self.active_playlist[station])
"""Dedupe"""
logging.info("Removing duplicate tracks...")
@ -441,7 +447,7 @@ class RadioUtil:
station, len(self.active_playlist[station]),
)
self.playlists_loaded = True
self.loop.run_until_complete(self._ls_skip())
# self.loop.run_until_complete(self._ls_skip())
except Exception as e:
logging.info("Playlist load failed: %s", str(e))
traceback.print_exc()
@ -521,22 +527,24 @@ class RadioUtil:
return (x, item)
return None
async def _ls_skip(self) -> bool:
async def _ls_skip(self, station: str = "main") -> bool:
"""
Ask LiquidSoap server to skip to the next track
Args:
None
station (str): default "main"
Returns:
bool
"""
try:
async with ClientSession() as session:
async with session.get(
f"{self.ls_uri}/next", timeout=ClientTimeout(connect=2, sock_read=2)
async with session.post(
f"{self.ls_uri}/next",
data=station,
timeout=ClientTimeout(connect=2, sock_read=2)
) as request:
request.raise_for_status()
text: Optional[str] = await request.text()
return text == "OK"
return isinstance(text, str) and text.startswith("OK")
except Exception as e:
logging.debug("Skip failed: %s", str(e))
@ -558,15 +566,17 @@ class RadioUtil:
return None
return response
async def webhook_song_change(self, track: dict) -> None:
async def webhook_song_change(self, track: dict, station: str = "main") -> None:
"""
Handles Song Change Outbounds (Webhooks)
Args:
track (dict)
station (str): default "main"
Returns:
None
"""
try:
return None # disabled temporarily (needs rate limiting)
# First, send track info
"""
TODO:
@ -582,7 +592,7 @@ class RadioUtil:
"username": "serious.FM",
"embeds": [
{
"title": "Now Playing",
"title": f"Now Playing on {station.title()}",
"description": f"## {track['song']}\nby\n## {track['artist']}",
"color": 0x30C56F,
"thumbnail": {
@ -636,36 +646,37 @@ class RadioUtil:
) as request:
request.raise_for_status()
# Next, AI feedback
# Next, AI feedback (for main stream only)
ai_response: Optional[str] = await self.get_ai_song_info(
track["artist"], track["song"]
)
if not ai_response:
return
if station == "main":
ai_response: Optional[str] = await self.get_ai_song_info(
track["artist"], track["song"]
)
if not ai_response:
return
hook_data = {
"username": "GPT",
"embeds": [
{
"title": "AI Feedback",
"color": 0x35D0FF,
"description": ai_response.strip(),
}
],
}
hook_data = {
"username": "GPT",
"embeds": [
{
"title": "AI Feedback",
"color": 0x35D0FF,
"description": ai_response.strip(),
}
],
}
ai_hook: str = self.webhooks["gpt"].get("hook")
async with ClientSession() as session:
async with await session.post(
ai_hook,
json=hook_data,
timeout=ClientTimeout(connect=5, sock_read=5),
headers={
"content-type": "application/json; charset=utf-8",
},
) as request:
request.raise_for_status()
ai_hook: str = self.webhooks["gpt"].get("hook")
async with ClientSession() as session:
async with await session.post(
ai_hook,
json=hook_data,
timeout=ClientTimeout(connect=5, sock_read=5),
headers={
"content-type": "application/json; charset=utf-8",
},
) as request:
request.raise_for_status()
except Exception as e:
logging.info("Webhook error occurred: %s", str(e))
traceback.print_exc()