Fix/spotify api feb 2026

.gitignore

@@ -34,3 +34,4 @@ dmypy.json
 
 # Visual Studio Code
 .vscode/
+example_config.yml

src/spotify_to_tidal/sync.py

@@ -4,7 +4,7 @@
 from .cache import failure_cache, track_match_cache
 import datetime
 from difflib import SequenceMatcher
-from functools import partial
+from itertools import chain
 from typing import Callable, List, Sequence, Set, Mapping
 import math
 import requests
@@ -17,7 +17,6 @@
 from tqdm import tqdm
 import traceback
 import unicodedata
-import math
 
 from .type import spotify as t_spotify
 
@@ -29,8 +28,8 @@ def simple(input_string: str) -> str:
     return input_string.split('-')[0].strip().split('(')[0].strip().split('[')[0].strip()
 
 def isrc_match(tidal_track: tidalapi.Track, spotify_track) -> bool:
-    if "isrc" in spotify_track["external_ids"]:
-        return tidal_track.isrc == spotify_track["external_ids"]["isrc"]
+    """Match by ISRC if external_ids is present. Note: external_ids has been removed
+    from Spotify API as of February 2026."""
     return False
 
 def duration_match(tidal_track: tidalapi.Track, spotify_track, tolerance=2) -> bool:
@@ -159,37 +158,67 @@ async def repeat_on_request_error(function, *args, remaining=5, **kwargs):
 
 
 async def _fetch_all_from_spotify_in_chunks(fetch_function: Callable) -> List[dict]:
-    output = []
-    results = fetch_function(0)
-    output.extend([item['track'] for item in results['items'] if item['track'] is not None])
+    """Make paginated requests to Spotify and return a list of all responses."""
+    responses = [fetch_function(0)]
 
-    # Get all the remaining tracks in parallel
-    if results['next']:
-        offsets = [results['limit'] * n for n in range(1, math.ceil(results['total'] / results['limit']))]
-        extra_results = await atqdm.gather(
+    # Get all the remaining pages in parallel
+    if responses[0]['next']:
+        limit: int = responses[0]['limit']
+        total: int = responses[0]['total']
+        num_pages = math.ceil(total/limit)
+
+        offsets = [limit * n for n in range(1, num_pages)]
+        responses += await atqdm.gather(
             *[asyncio.to_thread(fetch_function, offset) for offset in offsets],
             desc="Fetching additional data chunks"
         )
-        for extra_result in extra_results:
-            output.extend([item['track'] for item in extra_result['items'] if item['track'] is not None])
 
-    return output
+    return responses
 
 
 async def get_tracks_from_spotify_playlist(spotify_session: spotipy.Spotify, spotify_playlist):
     def _get_tracks_from_spotify_playlist(offset: int, playlist_id: str):
-        fields = "next,total,limit,items(track(name,album(name,artists),artists,track_number,duration_ms,id,external_ids(isrc))),type"
-        return spotify_session.playlist_tracks(playlist_id=playlist_id, fields=fields, offset=offset)
+        # Use new API endpoint /playlists/{id}/items instead of deprecated /playlists/{id}/tracks
+        # The /playlists/{id}/tracks endpoint was removed in February 2026
+        url = f"https://api.spotify.com/v1/playlists/{playlist_id}/items"
+        # Get access token - ensure we get it as a string, not a dict
+        token_info = spotify_session.auth_manager.get_access_token(as_dict=True)
+        if isinstance(token_info, dict):
+            access_token = token_info.get('access_token')
+        else:
+            access_token = token_info
+        headers = {"Authorization": f"Bearer {access_token}"}
+        params = {
+            "limit": 100,
+            "offset": offset,
+            "fields": "next,total,limit,items(item(name,album(name,artists),artists,track_number,duration_ms,id)),type"
+        }
+        response = requests.get(url, headers=headers, params=params)
+        response.raise_for_status()
+        return response.json()
 
     print(f"Loading tracks from Spotify playlist '{spotify_playlist['name']}'")
-    items = await repeat_on_request_error( _fetch_all_from_spotify_in_chunks, lambda offset: _get_tracks_from_spotify_playlist(offset=offset, playlist_id=spotify_playlist["id"]))
+
+    responses = await repeat_on_request_error(
+        _fetch_all_from_spotify_in_chunks,
+        lambda offset: _get_tracks_from_spotify_playlist(offset=offset, playlist_id=spotify_playlist["id"]),
+    )
+
+    # Check if the response contains items (may not contain items for playlists not owned by user)
+    if not responses or 'items' not in responses[0]:
+        print(f"Warning: No items found in playlist. This may be a playlist you don't have access to.")
+        return []
+
+    items = chain.from_iterable(r.get('items', []) for r in responses)
+    items_data = [item['item'] for item in items if 'item' in item and item['item'] is not None]
+
     track_filter = lambda item: item.get('type', 'track') == 'track' # type may be 'episode' also
     sanity_filter = lambda item: ('album' in item
                                   and 'name' in item['album']
                                   and 'artists' in item['album']
                                   and len(item['album']['artists']) > 0
                                   and item['album']['artists'][0]['name'] is not None)
-    return list(filter(sanity_filter, filter(track_filter, items)))
+    return list(filter(sanity_filter, filter(track_filter, items_data)))
 
 def populate_track_match_cache(spotify_tracks_: Sequence[t_spotify.SpotifyTrack], tidal_tracks_: Sequence[tidalapi.Track]):
     """ Populate the track match cache with all the existing tracks in Tidal playlist corresponding to Spotify playlist """
@@ -239,7 +268,7 @@ def get_tracks_for_new_tidal_playlist(spotify_tracks: Sequence[t_spotify.Spotify
             if tidal_id in seen_tracks:
                 track_name = spotify_track['name']
                 artist_names = ', '.join([artist['name'] for artist in spotify_track['artists']])
-                print(f'Duplicate found: Track "{track_name}" by {artist_names} will be ignored') 
+                print(f'Duplicate found: Track "{track_name}" by {artist_names} will be ignored')
             else:
                 output.append(tidal_id)
                 seen_tracks.add(tidal_id)
@@ -287,7 +316,7 @@ async def _run_rate_limiter(semaphore):
         for song in song404:
             file.write(f"{song}\n")
 
-            
+
 async def sync_playlist(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, spotify_playlist, tidal_playlist: tidalapi.Playlist | None, config: dict):
     """ sync given playlist to tidal """
     # Get the tracks from both Spotify and Tidal, creating a new Tidal playlist if necessary
@@ -318,14 +347,15 @@ async def sync_playlist(spotify_session: spotipy.Spotify, tidal_session: tidalap
         clear_tidal_playlist(tidal_playlist)
         add_multiple_tracks_to_playlist(tidal_playlist, new_tidal_track_ids)
 
+async def _get_tracks_from_spotify_favorites(spotify_session: spotipy.Spotify) -> List[dict]:
+    _get_favorite_tracks = lambda offset: spotify_session.current_user_saved_tracks(offset=offset)
+    pages = await repeat_on_request_error(_fetch_all_from_spotify_in_chunks, _get_favorite_tracks)
+    tracks = [item['track'] for page in pages for item in page['items'] if item['track'] is not None]
+    tracks.reverse()
+    return tracks
+
 async def sync_favorites(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config: dict):
     """ sync user favorites to tidal """
-    async def get_tracks_from_spotify_favorites() -> List[dict]:
-        _get_favorite_tracks = lambda offset: spotify_session.current_user_saved_tracks(offset=offset)    
-        tracks = await repeat_on_request_error( _fetch_all_from_spotify_in_chunks, _get_favorite_tracks)
-        tracks.reverse()
-        return tracks
-
     def get_new_tidal_favorites() -> List[int]:
         existing_favorite_ids = set([track.id for track in old_tidal_tracks])
         new_ids = []
@@ -336,7 +366,7 @@ def get_new_tidal_favorites() -> List[int]:
         return new_ids
 
     print("Loading favorite tracks from Spotify")
-    spotify_tracks = await get_tracks_from_spotify_favorites()
+    spotify_tracks = await _get_tracks_from_spotify_favorites(spotify_session)
     print("Loading existing favorite tracks from Tidal")
     old_tidal_tracks = await get_all_favorites(tidal_session.user.favorites, order='DATE')
     populate_track_match_cache(spotify_tracks, old_tidal_tracks)

src/spotify_to_tidal/type/spotify.py

@@ -19,21 +19,21 @@ class SpotifyFollower(TypedDict):
 
 class SpotifyArtist(TypedDict):
     external_urls: Mapping[str, str]
-    followers: SpotifyFollower
-    genres: List[str]
+    followers: Optional[SpotifyFollower]  # Removed from API in Feb 2026
+    genres: Optional[List[str]]  # Removed from API in Feb 2026
     href: str
     id: str
     images: List[SpotifyImage]
     name: str
-    popularity: int
+    popularity: Optional[int]  # Removed from API in Feb 2026
     type: str
     uri: str
 
 
 class SpotifyAlbum(TypedDict):
     album_type: Literal["album", "single", "compilation"]
     total_tracks: int
-    available_markets: List[str]
+    available_markets: Optional[List[str]]  # Removed from API in Feb 2026
     external_urls: Dict[str, str]
     href: str
     id: str
@@ -50,19 +50,19 @@ class SpotifyAlbum(TypedDict):
 class SpotifyTrack(TypedDict):
     album: SpotifyAlbum
     artists: List[SpotifyArtist]
-    available_markets: List[str]
+    available_markets: Optional[List[str]]  # Removed from API in Feb 2026
     disc_number: int
     duration_ms: int
     explicit: bool
-    external_ids: Dict[str, str]
+    external_ids: Optional[Dict[str, str]]  # Removed from API in Feb 2026
     external_urls: Dict[str, str]
     href: str
     id: str
     is_playable: bool
-    linked_from: Dict
+    linked_from: Optional[Dict]  # Removed from API in Feb 2026
     restrictions: Optional[Dict[Literal["reason"], str]]
     name: str
-    popularity: int
+    popularity: Optional[int]  # Removed from API in Feb 2026
     preview_url: str
     track_number: int
     type: Literal["track"]