diff --git a/.gitignore b/.gitignore index aea6eda..18df7ca 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,4 @@ dmypy.json # Visual Studio Code .vscode/ +example_config.yml diff --git a/src/spotify_to_tidal/sync.py b/src/spotify_to_tidal/sync.py index 4d0c482..1fc2427 100755 --- a/src/spotify_to_tidal/sync.py +++ b/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) diff --git a/src/spotify_to_tidal/type/spotify.py b/src/spotify_to_tidal/type/spotify.py index 3970ad5..77e918c 100644 --- a/src/spotify_to_tidal/type/spotify.py +++ b/src/spotify_to_tidal/type/spotify.py @@ -19,13 +19,13 @@ 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 @@ -33,7 +33,7 @@ class SpotifyArtist(TypedDict): 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"] diff --git a/tests/unit/test_sync.py b/tests/unit/test_sync.py new file mode 100644 index 0000000..adc40cf --- /dev/null +++ b/tests/unit/test_sync.py @@ -0,0 +1,188 @@ +from typing import Literal +from unittest.mock import MagicMock +import asyncio + +from spotify_to_tidal.sync import isrc_match +from spotify_to_tidal.sync import _fetch_all_from_spotify_in_chunks +from spotify_to_tidal.sync import get_tracks_from_spotify_playlist +from spotify_to_tidal.sync import _get_tracks_from_spotify_favorites + + +def _make_tidal_track(isrc: str = 'USAT21234567') -> MagicMock: + track = MagicMock() + track.isrc = isrc + return track + + +def _make_spotify_track( + isrc: str = 'USAT21234567', + name: str = 'Test Track', + external_ids: bool = True, +) -> dict: + """Build a minimal Spotify track dict.""" + track = { + 'name': name, + 'id': 'spotify123', + 'type': 'track', + 'duration_ms': 210000, + 'track_number': 1, + 'artists': [{'name': 'Test Artist'}], + 'album': {'name': 'Test Album', 'artists': [{'name': 'Test Artist'}]}, + } + if external_ids: + track['external_ids'] = {'isrc': isrc} + return track + +def _make_spotify_api_response(items, next_url=None, total=None, limit=50) -> dict: + """Build a response dict matching Spotify's playlist items endpoint.""" + return { + 'items': items, + 'next': next_url, + 'total': total or len(items), + 'limit': limit, + } + + +def _make_playlist_item(track_name: str = 'Test Track', item_type: Literal['track','episode'] = 'track') -> dict: + """Build a single playlist item.""" + return { + 'item': { + 'name': track_name, + 'id': 'spotify123', + 'type': item_type, + 'duration_ms': 210000, + 'track_number': 1, + 'artists': [{'name': 'Test Artist'}], + 'album': {'name': 'Test Album', 'artists': [{'name': 'Test Artist'}]}, + }, + 'type': item_type, + } + + +def _make_saved_track_item(track_name: str = 'Test Track') -> dict: + """Build a single saved track item matching current_user_saved_tracks response format.""" + return { + 'track': { + 'name': track_name, + 'id': 'spotify123', + 'type': 'track', + 'duration_ms': 210000, + 'track_number': 1, + 'artists': [{'name': 'Test Artist'}], + 'album': {'name': 'Test Album', 'artists': [{'name': 'Test Artist'}]}, + }, + } + + + +class TestIsrcMatch: + def test_matches_when_isrc_present_and_equal(self) -> None: + tidal = _make_tidal_track(isrc='USAT21234567') + spotify = _make_spotify_track(isrc='USAT21234567') + assert isrc_match(tidal, spotify) is True + + def test_no_match_when_isrc_differs(self) -> None: + tidal = _make_tidal_track(isrc='USAT21234567') + spotify = _make_spotify_track(isrc='GBDCA0000001') + assert isrc_match(tidal, spotify) is False + + def test_returns_false_when_external_ids_missing(self) -> None: + tidal = _make_tidal_track() + spotify = _make_spotify_track(external_ids=False) + assert isrc_match(tidal, spotify) is False + + def test_returns_false_when_isrc_key_missing(self) -> None: + tidal = _make_tidal_track() + spotify = _make_spotify_track() + spotify['external_ids'] = {} + assert isrc_match(tidal, spotify) is False + + + +class TestFetchAllFromSpotifyInChunks: + def test_single_page(self) -> None: + items = [_make_playlist_item('Track 1'), _make_playlist_item('Track 2')] + response = _make_spotify_api_response(items) + + def fetch_fn(offset: int) -> dict: + return response + + result = asyncio.run(_fetch_all_from_spotify_in_chunks(fetch_fn)) + assert len(result) == 1 + assert result[0] is response + + def test_multiple_pages(self) -> None: + page1_items = [_make_playlist_item('Track 1')] + page2_items = [_make_playlist_item('Track 2')] + page1 = _make_spotify_api_response( + page1_items, next_url='https://spotify.example.com/next', total=2, limit=1 + ) + page2 = _make_spotify_api_response(page2_items, total=2, limit=1) + + def fetch_fn(offset: int) -> dict: + return page1 if offset == 0 else page2 + + result = asyncio.run(_fetch_all_from_spotify_in_chunks(fetch_fn)) + assert len(result) == 2 + assert result[0] is page1 + assert result[1] is page2 + + +class TestGetTracksFromSpotifyPlaylist: + def test_filters_out_episodes(self) -> None: + items = [ + _make_playlist_item('Track 1', item_type='track'), + _make_playlist_item('Episode 1', item_type='episode'), + ] + response = _make_spotify_api_response(items) + mock_session = MagicMock() + mock_session.playlist_items.return_value = response + spotify_playlist = {'name': 'My Playlist', 'id': 'playlist123'} + + result = asyncio.run( + get_tracks_from_spotify_playlist(mock_session, spotify_playlist) + ) + assert len(result) == 1 + assert result[0]['name'] == 'Track 1' + + +class TestGetTracksFromSpotifyFavorites: + def test_extracts_tracks_from_saved_tracks_response(self) -> None: + items = [ + _make_saved_track_item('Fav 1'), + _make_saved_track_item('Fav 2'), + ] + response = _make_spotify_api_response(items) + mock_session = MagicMock() + mock_session.current_user_saved_tracks.return_value = response + + result = asyncio.run(_get_tracks_from_spotify_favorites(mock_session)) + assert len(result) == 2 + assert result[0]['name'] == 'Fav 2' + assert result[1]['name'] == 'Fav 1' + + def test_filters_out_none_tracks(self) -> None: + items = [_make_saved_track_item('Fav 1'), {'track': None}] + response = _make_spotify_api_response(items) + mock_session = MagicMock() + mock_session.current_user_saved_tracks.return_value = response + + result = asyncio.run(_get_tracks_from_spotify_favorites(mock_session)) + assert len(result) == 1 + assert result[0]['name'] == 'Fav 1' + + def test_multiple_pages(self) -> None: + page1 = _make_spotify_api_response( + [_make_saved_track_item('Fav 1')], + next_url='https://spotify.example.com/next', total=2, limit=1, + ) + page2 = _make_spotify_api_response( + [_make_saved_track_item('Fav 2')], total=2, limit=1, + ) + mock_session = MagicMock() + mock_session.current_user_saved_tracks.side_effect = lambda offset: page1 if offset == 0 else page2 + + result = asyncio.run(_get_tracks_from_spotify_favorites(mock_session)) + assert len(result) == 2 + assert result[0]['name'] == 'Fav 2' + assert result[1]['name'] == 'Fav 1'