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

misc/basic genre filtering

Browse Source
This commit is contained in:
codey 2025-04-22 15:31:26 -04:00
parent 40fa51af36
commit f18a9da4a0
1 changed files with 65 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
utils/radio_util.py
View File

@ -48,7 +48,14 @@ class RadioUtil:
self.album_art_db_path: str = os.path.join(
"/usr/local/share", "sqlite_dbs", "track_album_art.db"
)
self.active_playlist_name = "default" # not used
self.playback_genres: list[str] = [
"post-hardcore",
"post hardcore",
"metalcore",
"deathcore",
"edm",
"electronic",
]
self.active_playlist: list[dict] = []
self.now_playing: dict = {
"artist": "N/A",
@ -181,23 +188,30 @@ class RadioUtil:
"""
try:
async with sqlite3.connect(self.artist_genre_db_path, timeout=2) as _db:
query: str = "INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
query: str = (
"INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
)
params: tuple[str, str] = (artist, genre)
res = await _db.execute_insert(query, params)
if res:
logging.debug("Query executed successfully for %s/%s, committing",
artist, genre)
logging.debug(
"Query executed successfully for %s/%s, committing",
artist,
genre,
)
await _db.commit()
return True
logging.debug("Failed to store artist/genre pair: %s/%s (res: %s)",
artist, genre, res)
logging.debug(
"Failed to store artist/genre pair: %s/%s (res: %s)", artist, genre, res
)
return False
except Exception as e:
logging.info("Failed to store artist/genre pair: %s/%s (%s)",
artist, genre, str(e))
logging.info(
"Failed to store artist/genre pair: %s/%s (%s)", artist, genre, str(e)
)
traceback.print_exc()
return False
async def add_genres(self, pairs: list[dict[str, str]]) -> bool:
"""
(BATCH) Add artist/genre pairings to DB
@ -213,20 +227,33 @@ class RadioUtil:
for pair in pairs:
try:
artist, genre = pair
query: str = "INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
query: str = (
"INSERT OR IGNORE INTO artist_genre (artist, genre) VALUES(?, ?)"
)
params: tuple[str, str] = (artist, genre)
res = await _db.execute_insert(query, params)
if res:
logging.debug("add_genres: Query executed successfully for %s/%s",
artist, genre)
logging.debug(
"add_genres: Query executed successfully for %s/%s",
artist,
genre,
)
added_rows += 1
else:
logging.debug("Failed to store artist/genre pair: %s/%s (res: %s)",
artist, genre, res)
logging.debug(
"Failed to store artist/genre pair: %s/%s (res: %s)",
artist,
genre,
res,
)
except Exception as e:
logging.info("Failed to store artist/genre pair: %s/%s (%s)",
artist, genre, str(e))
continue
logging.info(
"Failed to store artist/genre pair: %s/%s (%s)",
artist,
genre,
str(e),
)
continue
if added_rows:
logging.info("add_genres: Committing %s rows", added_rows)
await _db.commit()
@ -234,11 +261,9 @@ class RadioUtil:
logging.info("add_genres: Failed (No rows added)")
return False
except Exception as e:
logging.info("Failed to store artist/genre pairs: %s",
str(e))
logging.info("Failed to store artist/genre pairs: %s", str(e))
traceback.print_exc()
return False
async def get_genre(self, artist: str) -> str:
"""
@ -250,16 +275,17 @@ class RadioUtil:
"""
try:
artist = artist.strip()
query: str = "SELECT genre FROM artist_genre WHERE artist LIKE ?"
query: str = "SELECT genre FROM artist_genre WHERE artist LIKE ? COLLATE NOCASE"
params: tuple[str] = (f"%%{artist}%%",)
async with sqlite3.connect(self.artist_genre_db_path, timeout=2) as _db:
_db.row_factory = sqlite3.Row
async with await _db.execute(query, params) as _cursor:
res = await _cursor.fetchone()
if not res:
raise RadioException(
f"Could not locate {artist} in artist_genre_map db."
)
return "Not Found" # Exception suppressed
# raise RadioException(
# f"Could not locate {artist} in artist_genre_map db."
# )
return res["genre"]
except Exception as e:
logging.info("Failed to look up genre for artist: %s (%s)", artist, str(e))
@ -331,6 +357,21 @@ class RadioUtil:
"Populated active playlists with %s items",
len(self.active_playlist),
)
if self.playback_genres:
new_playlist: list[dict] = []
logging.info("Limiting playback genres")
for x, item in enumerate(self.active_playlist):
item_genres: str = item.get("genre", "").strip().lower()
for genre in self.playback_genres:
genre = genre.strip().lower()
if genre in item_genres:
new_playlist.append(item)
continue
self.active_playlist = new_playlist
logging.info(
"%s items remain for playback after filtering",
len(self.active_playlist),
)
except Exception as e:
logging.info("Playlist load failed: %s", str(e))
traceback.print_exc()