updated tidal client playlist parsing
This commit is contained in:
nathan
parent
f4bdb76c72
commit
76760bdaaf
1 changed files with 118 additions and 35 deletions
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in a new issue