Feat/sync artists albums

example_config.yml

@@ -19,6 +19,8 @@ spotify:
 #   - when true:  favorites will be synced by default (overriden when any command line arg provided)
 #   - when false: favorites can only be synced manually via --sync-favorites argument
 sync_favorites_default: true
+sync_artists_default: true
+sync_albums_default: true
 
 # increasing these parameters should increase the search speed, while decreasing reduces likelihood of 429 errors
 max_concurrency: 10 # max concurrent connections at any given time

readme.md

@@ -36,6 +36,18 @@ or sync just your 'Liked Songs' with:
 spotify_to_tidal --sync-favorites
 ```
 
+or sync just your saved albums with:
+
+```bash
+spotify_to_tidal --sync-albums
+```
+
+or sync just your saved artists with:
+
+```bash
+spotify_to_tidal --sync-artists
+```
+
 See example_config.yml for more configuration options, and `spotify_to_tidal --help` for more options.
 
 ---

src/spotify_to_tidal/__main__.py

@@ -10,8 +10,14 @@ def main():
     parser.add_argument('--config', default='config.yml', help='location of the config file')
     parser.add_argument('--uri', help='synchronize a specific URI instead of the one in the config')
     parser.add_argument('--sync-favorites', action=argparse.BooleanOptionalAction, help='synchronize the favorites')
+    parser.add_argument('--sync-artists', action=argparse.BooleanOptionalAction, help='synchronize the artists')
+    parser.add_argument('--sync-albums', action=argparse.BooleanOptionalAction, help='synchronize the albums')
     args = parser.parse_args()
 
+    sync_favorites = False
+    sync_artists = False
+    sync_albums = False
+
     with open(args.config, 'r') as f:
         config = yaml.safe_load(f)
     print("Opening Spotify session")
@@ -27,20 +33,39 @@ def main():
         tidal_playlist = _sync.pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists)
         _sync.sync_playlists_wrapper(spotify_session, tidal_session, [tidal_playlist], config)
         sync_favorites = args.sync_favorites # only sync favorites if command line argument explicitly passed
+        sync_artists = args.sync_artists # only sync artists if command line argument explicitly passed
+        sync_albums = args.sync_albums # only sync albums if command line argument explicitly passed
     elif args.sync_favorites:
         sync_favorites = True # sync only the favorites
+    elif args.sync_artists:
+        sync_artists = True # sync only the artists
+    elif args.sync_albums:
+        sync_albums = True # sync only the albums
     elif config.get('sync_playlists', None):
         # if the config contains a sync_playlists list of mappings then use that
         _sync.sync_playlists_wrapper(spotify_session, tidal_session, _sync.get_playlists_from_config(spotify_session, tidal_session, config), config)
         sync_favorites = args.sync_favorites is None and config.get('sync_favorites_default', True)
+        sync_artists = args.sync_artists is None and config.get('sync_artists_default', False)
+        sync_albums = args.sync_albums is None and config.get('sync_albums_default', False)
     else:
         # otherwise sync all the user playlists in the Spotify account and favorites unless explicitly disabled
         _sync.sync_playlists_wrapper(spotify_session, tidal_session, _sync.get_user_playlist_mappings(spotify_session, tidal_session, config), config)
         sync_favorites = args.sync_favorites is None and config.get('sync_favorites_default', True)
+        sync_artists = args.sync_artists is None and config.get('sync_artists_default', False)
+        sync_albums = args.sync_albums is None and config.get('sync_albums_default', False)
 
+    # Sync favorites
     if sync_favorites:
         _sync.sync_favorites_wrapper(spotify_session, tidal_session, config)
 
+    # Sync artists
+    if sync_artists:
+        _sync.sync_artists_wrapper(spotify_session, tidal_session, config)
+
+    # Sync albums
+    if sync_albums:
+        _sync.sync_albums_wrapper(spotify_session, tidal_session, config)
+
 if __name__ == '__main__':
     main()
     sys.exit(0)

src/spotify_to_tidal/auth.py

@@ -11,7 +11,7 @@
     'open_tidal_session'
 ]
 
-SPOTIFY_SCOPES = 'playlist-read-private, user-library-read'
+SPOTIFY_SCOPES = 'playlist-read-private, user-library-read, user-follow-read'
 
 def open_spotify_session(config) -> spotipy.Spotify:
     credentials_manager = spotipy.SpotifyOAuth(username=config['username'],

src/spotify_to_tidal/sync.py

@@ -158,20 +158,22 @@ async def repeat_on_request_error(function, *args, remaining=5, **kwargs):
         return await repeat_on_request_error(function, *args, remaining=remaining-1, **kwargs)
 
 
-async def _fetch_all_from_spotify_in_chunks(fetch_function: Callable) -> List[dict]:
+async def _fetch_all_from_spotify_in_chunks(fetch_function: Callable, item_key: str = "track") -> List[dict]:
     output = []
     results = fetch_function(0)
-    output.extend([item['track'] for item in results['items'] if item['track'] is not None])
+
+    # Get all the items from the first chunk
+    output.extend([item[item_key] for item in results['items'] if item.get(item_key) is not None])
 
-    # Get all the remaining tracks in parallel
+    # Get all the remaining chunks 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(
             *[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])
+            output.extend([item[item_key] for item in extra_result['items'] if item.get(item_key) is not None])
 
     return output
 
@@ -342,6 +344,129 @@ def get_new_tidal_favorites() -> List[int]:
     else:
         print("No new tracks to add to Tidal favorites")
 
+async def sync_artists(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config: dict):
+    """Synchronize user-followed artists from Spotify to Tidal."""
+    print("Loading followed artists from Spotify")
+
+    async def get_all_followed_artists() -> List[dict]:
+        followed_artists = []
+        after = None
+
+        while True:
+            response = await repeat_on_request_error(
+                lambda: asyncio.to_thread(spotify_session.current_user_followed_artists, after=after)
+            )
+            artists = response['artists']['items']
+            followed_artists.extend(artists)
+
+            if not response['artists']['cursors'].get('after'):
+                break
+
+            after = response['artists']['cursors']['after']
+
+        return followed_artists
+
+    async def search_artist_on_tidal(artist_name: str, tidal_session: tidalapi.Session):
+        """Search for an artist on Tidal by name."""
+        try:
+            # Search for the artist on Tidal
+            result = tidal_session.search(artist_name, models=[tidalapi.artist.Artist])
+
+            # Check if the search result contains artists
+            if result and 'artists' in result and isinstance(result['artists'], list) and len(result['artists']) > 0:
+                # Return the first artist from the search results
+                return result['artists'][0]
+        except Exception as e:
+            print(f"Error searching for artist '{artist_name}' on Tidal: {e}")
+        return None
+
+    # Fetch all followed artists from Spotify
+    artists = await get_all_followed_artists()
+    if not artists:
+        print("No artists followed on Spotify.")
+        return
+
+    print(f"Found {len(artists)} artists followed on Spotify.")
+
+    # Load existing followed artists from Tidal
+    print("Loading existing followed artists from Tidal")
+    tidal_artists = tidal_session.user.favorites.artists()
+    tidal_artist_names = set([normalize(artist.name.lower()) for artist in tidal_artists])
+
+    # Filter new artists that are not already followed on Tidal
+    new_artists = [artist for artist in artists if normalize(artist['name'].lower()) not in tidal_artist_names]
+
+    if not new_artists:
+        print("All followed artists are already in Tidal.")
+        return
+
+    # Add new artists to Tidal
+    print(f"Searching and adding {len(new_artists)} new artists to Tidal.")
+    for artist in tqdm(new_artists, desc="Adding new artists to Tidal"):
+        try:
+            # Search for the artist on Tidal
+            tidal_artist = await search_artist_on_tidal(artist['name'], tidal_session)
+            if tidal_artist:
+                # Add the artist to Tidal favorites
+                tidal_session.user.favorites.add_artist(tidal_artist.id)
+            else:
+                print(f"Artist '{artist['name']}' not found on Tidal.")
+        except Exception as e:
+            print(f"Failed to add artist '{artist['name']}' to Tidal: {e}")
+
+    print("Artist synchronization complete.")
+
+async def sync_albums(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config: dict):
+    """Synchronize user-saved albums from Spotify to Tidal."""
+    print("Loading saved albums from Spotify")
+
+    # Get all saved albums from Spotify
+    def _get_saved_albums(offset=0):
+        return spotify_session.current_user_saved_albums(offset=offset)
+
+    # Fetch all saved albums from Spotify
+    results = await repeat_on_request_error(_fetch_all_from_spotify_in_chunks, _get_saved_albums, item_key="album")
+    albums = [item for item in results]
+
+    print(f"Found {len(albums)} albums saved on Spotify")
+
+    # Get existing saved albums from Tidal
+    tidal_albums = tidal_session.user.favorites.albums()
+    tidal_album_names = set([normalize(album.name.lower()) for album in tidal_albums])
+
+    # Filter new albums to add to Tidal
+    new_albums = [album for album in albums if normalize(album['name'].lower()) not in tidal_album_names]
+
+    if not new_albums:
+        print("All saved albums are already in Tidal.")
+        return
+
+    # Function to search for an album on Tidal
+    async def search_album_on_tidal(album_name: str, artist_name: str):
+        try:
+            query = f"{album_name} {artist_name}"
+            result = tidal_session.search(query, models=[tidalapi.album.Album])
+            if result and 'albums' in result and result['albums']:
+                return result['albums'][0]  # Return the first match
+        except Exception as e:
+            print(f"Error searching for album '{album_name}' on Tidal: {e}")
+        return None
+
+    # Add new albums to Tidal
+    for album in tqdm(new_albums, desc="Adding new albums to Tidal"):
+        try:
+            artist_name = album['artists'][0]['name'] if album['artists'] else ""
+            tidal_album = await search_album_on_tidal(album['name'], artist_name)
+            if tidal_album:
+                tidal_session.user.favorites.add_album(tidal_album.id)
+            else:
+                print(f"Album '{album['name']}' not found on Tidal.")
+        except Exception as e:
+            print(f"Failed to add album '{album['name']}' to Tidal: {e}")
+
+    print("Album synchronization complete.")
+
+
 def sync_playlists_wrapper(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, playlists, config: dict):
   for spotify_playlist, tidal_playlist in playlists:
     # sync the spotify playlist to tidal
@@ -350,6 +475,12 @@ def sync_playlists_wrapper(spotify_session: spotipy.Spotify, tidal_session: tida
 def sync_favorites_wrapper(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config):
     asyncio.run(main=sync_favorites(spotify_session=spotify_session, tidal_session=tidal_session, config=config))
 
+def sync_artists_wrapper(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config: dict):
+    asyncio.run(sync_artists(spotify_session=spotify_session, tidal_session=tidal_session, config=config))
+
+def sync_albums_wrapper(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config: dict):
+    asyncio.run(sync_albums(spotify_session=spotify_session, tidal_session=tidal_session, config=config))
+
 def get_tidal_playlists_wrapper(tidal_session: tidalapi.Session) -> Mapping[str, tidalapi.Playlist]:
     tidal_playlists = asyncio.run(get_all_playlists(tidal_session.user))
     return {playlist.name: playlist for playlist in tidal_playlists}