From f18a9da4a0c5349cf71abac13ab19cdac9a94b57 Mon Sep 17 00:00:00 2001 From: codey Date: Tue, 22 Apr 2025 15:31:26 -0400 Subject: [PATCH] misc/basic genre filtering --- utils/radio_util.py | 89 +++++++++++++++++++++++++++++++++------------ 1 file changed, 65 insertions(+), 24 deletions(-) diff --git a/utils/radio_util.py b/utils/radio_util.py index 9473e14..f002685 100644 --- a/utils/radio_util.py +++ b/utils/radio_util.py @@ -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()