- Changed API key validation from if not _key in self.constants.API_KEYS to if _key not in self.constants.API_KEYS for better readability.

Enhance RadioUtil playlist handling and deduplication

- Added checks to ensure playlists are initialized and not empty.
- Improved deduplication logic to prevent modifying the original playlist during iteration.
- Added logging for duplicate removal and playlist population.

Add cover art handling in rip_background.py

- Implemented functionality to attach album art if provided in metadata.
- Added error handling for cover art download failures.

Introduce unique filename handling in rip_background.py

- Added `ensure_unique_filename_in_dir` function to prevent overwriting files with the same name.

Refactor SRUtil for improved error handling and metadata fetching

- Introduced `MetadataFetchError` for better error management during metadata retrieval.
- Implemented `_safe_api_call` for resilient API calls with retry logic.
- Enhanced `get_artists_by_name` to optionally group results by artist name.
- Updated various methods to utilize the new error handling and retry mechanisms.
This commit is contained in:
2025-09-22 11:08:48 -04:00
parent e1194475b3
commit c2044711fb
9 changed files with 1466 additions and 354 deletions

View File

@@ -35,6 +35,7 @@ class RadioUtil:
self.gpt = gpt.GPT(self.constants)
self.ls_uri: str = self.constants.LS_URI
self.redis_client = redis.Redis(password=private.REDIS_PW)
self.DEDUPE_PLAYLISTS: bool = True
self.sqlite_exts: list[str] = [
"/home/kyle/api/solibs/spellfix1.cpython-311-x86_64-linux-gnu.so"
]
@@ -392,41 +393,69 @@ class RadioUtil:
for playlist in self.playlists:
playlist_redis_key: str = f"playlist:{playlist}"
_playlist = await self.redis_client.json().get(playlist_redis_key) # type: ignore
# Ensure we always have a list to work with
if not _playlist:
logging.warning("No playlist found in redis for %s, skipping", playlist)
self.active_playlist[playlist] = []
continue
# Make sure playlist key exists
if playlist not in self.active_playlist.keys():
self.active_playlist[playlist] = []
random.shuffle(_playlist)
self.active_playlist[playlist] = [
{
"uuid": str(uuid().hex),
"id": r["id"],
"artist": double_space.sub(" ", r["artist"]).strip(),
"song": double_space.sub(" ", r["song"]).strip(),
"album": double_space.sub(" ", r["album"]).strip(),
"genre": r["genre"] if r["genre"] else "Not Found",
"artistsong": double_space.sub(
" ", r["artistdashsong"]
).strip(),
"file_path": r["file_path"],
"duration": r["duration"],
} for r in _playlist
if r not in self.active_playlist[playlist]
]
# Shuffle a copy so we don't mutate the underlying redis object
try:
shuffled = list(_playlist)
random.shuffle(shuffled)
except Exception:
shuffled = _playlist
# Build a fresh list rather than modifying in-place (prevents duplication)
built: list[dict] = []
for r in shuffled:
try:
item = {
"uuid": str(uuid().hex),
"id": r.get("id"),
"artist": double_space.sub(" ", (r.get("artist") or "")).strip(),
"song": double_space.sub(" ", (r.get("song") or "")).strip(),
"album": double_space.sub(" ", (r.get("album") or "")).strip(),
"genre": r.get("genre") if r.get("genre") else "Not Found",
"artistsong": double_space.sub(" ", (r.get("artistdashsong") or "")).strip(),
"file_path": r.get("file_path"),
"duration": r.get("duration"),
}
built.append(item)
except Exception:
logging.debug("Skipping malformed playlist entry for %s: %s", playlist, r)
self.active_playlist[playlist] = built
logging.info(
"Populated playlist: %s with %s items",
playlist, len(self.active_playlist[playlist]),
)
"""Dedupe"""
logging.info("Removing duplicate tracks...")
dedupe_processed = []
for item in self.active_playlist[playlist]:
artistsongabc: str = non_alnum.sub("", item.get("artistsong", ""))
if not artistsongabc:
logging.info("Missing artistsong: %s", item)
continue
if artistsongabc in dedupe_processed:
self.active_playlist[playlist].remove(item)
dedupe_processed.append(artistsongabc)
if self.DEDUPE_PLAYLISTS:
logging.info("Removing duplicate tracks (by file_path only)...")
dedupe_processed: set[str] = set()
deduped_list: list[dict] = []
for item in self.active_playlist[playlist]:
fp = item.get("file_path")
if not fp:
# If no file_path available, skip the item (can't dedupe reliably)
logging.info("Skipping item without file_path during dedupe: %s", item)
continue
key = fp
if key in dedupe_processed:
continue
dedupe_processed.add(key)
deduped_list.append(item)
self.active_playlist[playlist] = deduped_list
else:
logging.warning("Dupe removal disabled")
logging.info(
"Duplicates for playlist: %s removed. New playlist size: %s",