From 76760bdaaf358ba6fa20ec65ee9a694f9b91b6d9 Mon Sep 17 00:00:00 2001 From: nathan Date: Wed, 14 Jan 2026 19:38:04 +0000 Subject: [PATCH] updated tidal client playlist parsing --- core/tidal_client.py | 153 +++++++++++++++++++++++++++++++++---------- 1 file changed, 118 insertions(+), 35 deletions(-) diff --git a/core/tidal_client.py b/core/tidal_client.py index 052b87e2..71d01649 100644 --- a/core/tidal_client.py +++ b/core/tidal_client.py @@ -1,5 +1,6 @@ import requests import time +import re import threading from typing import Dict, List, Optional, Any from functools import wraps @@ -163,7 +164,7 @@ class TidalClient: logger.info("Saved Tidal tokens") except Exception as e: logger.error(f"Error saving Tidal tokens: {e}") - + def _parse_json_api_track(self, track_data: Dict[str, Any], artist_details_map: Dict[str, Any] = None) -> Optional[Track]: """Parse a track from a JSON:API 'included' object with artist details.""" try: @@ -550,7 +551,7 @@ class TidalClient: endpoint = f"{self.base_url}/playlists" params = { 'countryCode': 'US', - 'filter[r.owners.id]': user_id + 'filter[owners.id]': user_id } headers = self.session.headers.copy() @@ -688,6 +689,7 @@ class TidalClient: @rate_limited def get_playlist(self, playlist_id: str) -> Optional[Playlist]: """Get playlist details including tracks""" + try: if not self._ensure_valid_token(): logger.error("Not authenticated with Tidal") @@ -708,67 +710,65 @@ class TidalClient: # Get playlist tracks with pagination to handle large playlists tracks = [] - offset = 0 - limit = 100 + cursor = "" total_fetched = 0 while True: - logger.info(f"Fetching tracks for playlist {playlist_id}: offset={offset}, limit={limit}") - + track_ids = [] + logger.info(f"Fetching tracks for playlist {playlist_id}: cursor={cursor}") + params = {"countryCode": "US"} + if cursor: + params["page[cursor]"] = cursor tracks_response = self.session.get( - f"{self.base_url}/playlists/{playlist_id}/items", - params={'countryCode': 'US', 'limit': limit, 'offset': offset}, + f"{self.base_url}/playlists/{playlist_id}/relationships/items", + params=params, timeout=10 ) if tracks_response.status_code != 200: - logger.error(f"Failed to get Tidal playlist tracks at offset {offset}: {tracks_response.status_code} - {tracks_response.text}") + logger.error(f"Failed to get Tidal playlist tracks at cursor {cursor}: {tracks_response.status_code} - {tracks_response.text}") break tracks_data = tracks_response.json() - if 'items' not in tracks_data: - logger.warning(f"No items found in playlist {playlist_id} response at offset {offset}") - break - - items = tracks_data['items'] - if len(items) == 0: - logger.info(f"No more tracks found at offset {offset}, stopping pagination") + if not tracks_data.get("data"): + logger.warning(f"No items found in playlist {playlist_id} response at cursor {cursor}") break # Process this batch of tracks batch_count = 0 - for item in items: - # Handle different item structures - track_data = item - if 'item' in item and item.get('type') == 'track': - track_data = item['item'] - elif 'resource' in item: - track_data = item['resource'] - - track = self._parse_track_data(track_data) - if track: - tracks.append(track) + for item in tracks_data.get("data",[]): + # we can do a batch call with the tracks to get the artist data + if item.get("type"): + track_ids.append(item.get("id")) batch_count += 1 + total_fetched += batch_count logger.info(f"Fetched {batch_count} tracks in this batch, {total_fetched} total so far") - # If we got fewer items than the limit, we've reached the end - if len(items) < limit: - logger.info(f"Received {len(items)} items (less than limit {limit}), pagination complete") - break + # now we have a page of tracks we can hydrate some of the data + + tracks.extend(self._get_track_playlist(track_ids)) # Move to next page - offset += limit + cursor = tracks_data.get("links", {}).get("meta", {}).get("nextCursor") + + # Tidal uses cursor based pagination so if no next cursor exists we can finish + if not cursor: + break + # now we have a list of track ids but not hydrated track data + + playlist_attrs = playlist_data.get("attributes", {}) + playlist = Playlist( id=playlist_data.get('id', playlist_id), - name=playlist_data.get('title', 'Unknown Playlist'), - description=playlist_data.get('description', ''), + name=playlist_attrs.get('name', 'Unknown Playlist'), + description=playlist_attrs.get('description', ''), tracks=tracks, external_urls={'tidal': f"https://tidal.com/browse/playlist/{playlist_id}"}, - public=not playlist_data.get('publicPlaylist', True) # Tidal uses 'publicPlaylist' field + public = playlist_attrs.get('accessType', '') == "PUBLIC" ) logger.info(f"Retrieved Tidal playlist '{playlist.name}' with {len(tracks)} tracks") @@ -778,6 +778,89 @@ class TidalClient: logger.error(f"Error getting Tidal playlist {playlist_id}: {e}") return None + + def parse_duration(self, duration: str) -> int: + """Convert ISO-8601 duration string (like 'PT3M36S') to milliseconds. + Only supports minutes and seconds. + """ + if not duration.startswith("PT"): + return 0 + + minutes = 0 + seconds = 0 + + # Extract minutes and seconds using regex + m = re.search(r"(\d+)M", duration) + s = re.search(r"(\d+)S", duration) + + if m: + minutes = int(m.group(1)) + if s: + seconds = int(s.group(1)) + + return (minutes * 60 + seconds) * 1000 + + def _get_playlist_track_data(self, track_ids: list[str]) -> list[Track]: + try: + params = {"countryCode": "US", "include":"artists,albums", "filter[id]": ",".join(track_ids)} + resp = self.session.get( + f"{self.base_url}/tracks", + params=params, + timeout=10 + ) + + if resp.status_code != 200: + logger.error(f"Failed to get Tidal playlist tracks data: {resp.status_code} - {resp.text}") + + tracks_data = resp.json() + + albumCache = Dict[str,str] = {} + artistCache = Dict[str,str] = {} + + # first scan through the included albums and artists so we can link ids to names + + for item in tracks_data.get("included"): + item_id = item.get("id") + if item.get("type") == "albums": + albumCache.setdefault(item_id, item.get("attributes", {}).get("title", "Unknown Album")) + if item.get("type") == "artists": + artistCache.setdefault(item_id, item.get("attributes", {}).get("name", "Unknown Artist")) + + # now go through the tracks and hydrate with the artist and album data + hydratedTracks: list[Track] = [] + for trackData in tracks_data.get("data"): + attrs = trackData.get("attributes", {}) + track_id = trackData.get("id") + relateds = trackData.get("relationships") + + album_data = relateds.get("albums", {}).get("data", []) + album_id = album_data[0].get("id") if album_data else None + album = albumCache.get(album_id, "Unknown Album") + + + artists = [ + artistCache.get(a.get("id"), "Unknown Artist") + for a in relateds.get("artists", {}).get("data", []) + if a.get("id") + ] + + hydratedTracks.append(Track( + id=str(track_id), + name=attrs.get('title', 'Unknown Track'), + artists=artists, + album=album, + duration_ms=self.parse_duration(attrs.get('duration')), + external_urls={'tidal': f"https://tidal.com/browse/track/{track_id}"}, + explicit=attrs.get('explicit', False) + )) + + except Exception as e: + logger.error(f"Error getting playlist tracks: {e}") + return [] + + return hydratedTracks + + def _parse_track_data(self, item: Dict[str, Any]) -> Optional[Track]: """Parse Tidal track data into Track object""" try: