Compare commits
6 Commits
1.0.1
...
a438cda72b
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a438cda72b | ||
|
|
2f1985a42b | ||
|
|
bcf2bbca0d | ||
|
|
457da1724f | ||
|
|
4d7c3b0ef0 | ||
|
|
4fb702d008 |
@@ -3,6 +3,7 @@ spotify:
|
|||||||
client_secret: your_client_secret
|
client_secret: your_client_secret
|
||||||
username: your_spotify_username
|
username: your_spotify_username
|
||||||
redirect_uri: http://localhost:8888/callback
|
redirect_uri: http://localhost:8888/callback
|
||||||
|
open_browser: True # Set to False if using a headless server environment
|
||||||
|
|
||||||
|
|
||||||
# uncomment this block if you want to only sync specific playlist IDs
|
# uncomment this block if you want to only sync specific playlist IDs
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "spotify_to_tidal"
|
name = "spotify_to_tidal"
|
||||||
version = "1.0.1"
|
version = "1.0.4"
|
||||||
requires-python = ">= 3.10"
|
requires-python = ">= 3.10"
|
||||||
|
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -15,11 +15,12 @@ SPOTIFY_SCOPES = 'playlist-read-private, user-library-read'
|
|||||||
|
|
||||||
def open_spotify_session(config) -> spotipy.Spotify:
|
def open_spotify_session(config) -> spotipy.Spotify:
|
||||||
credentials_manager = spotipy.SpotifyOAuth(username=config['username'],
|
credentials_manager = spotipy.SpotifyOAuth(username=config['username'],
|
||||||
scope=SPOTIFY_SCOPES,
|
scope=SPOTIFY_SCOPES,
|
||||||
client_id=config['client_id'],
|
client_id=config['client_id'],
|
||||||
client_secret=config['client_secret'],
|
client_secret=config['client_secret'],
|
||||||
redirect_uri=config['redirect_uri'],
|
redirect_uri=config['redirect_uri'],
|
||||||
requests_timeout=2)
|
requests_timeout=2,
|
||||||
|
open_browser=config.get('open_browser', True))
|
||||||
try:
|
try:
|
||||||
credentials_manager.get_access_token(as_dict=False)
|
credentials_manager.get_access_token(as_dict=False)
|
||||||
except spotipy.SpotifyOauthError:
|
except spotipy.SpotifyOauthError:
|
||||||
|
|||||||
@@ -178,12 +178,14 @@ async def _fetch_all_from_spotify_in_chunks(fetch_function: Callable) -> List[di
|
|||||||
|
|
||||||
async def get_tracks_from_spotify_playlist(spotify_session: spotipy.Spotify, spotify_playlist):
|
async def get_tracks_from_spotify_playlist(spotify_session: spotipy.Spotify, spotify_playlist):
|
||||||
def _get_tracks_from_spotify_playlist(offset: int, playlist_id: str):
|
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)))"
|
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)
|
return spotify_session.playlist_tracks(playlist_id=playlist_id, fields=fields, offset=offset)
|
||||||
|
|
||||||
print(f"Loading tracks from Spotify playlist '{spotify_playlist['name']}'")
|
print(f"Loading tracks from Spotify playlist '{spotify_playlist['name']}'")
|
||||||
return 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"]))
|
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"]))
|
||||||
|
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
|
||||||
|
return list(filter(sanity_filter, filter(track_filter, items)))
|
||||||
|
|
||||||
def populate_track_match_cache(spotify_tracks_: Sequence[t_spotify.SpotifyTrack], tidal_tracks_: Sequence[tidalapi.Track]):
|
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 """
|
""" Populate the track match cache with all the existing tracks in Tidal playlist corresponding to Spotify playlist """
|
||||||
@@ -282,14 +284,18 @@ async def search_new_tracks_on_tidal(tidal_session: tidalapi.Session, spotify_tr
|
|||||||
|
|
||||||
async def sync_playlist(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, spotify_playlist, tidal_playlist: tidalapi.Playlist | None, config: dict):
|
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 """
|
""" sync given playlist to tidal """
|
||||||
# Create a new Tidal playlist if required
|
# Get the tracks from both Spotify and Tidal, creating a new Tidal playlist if necessary
|
||||||
if not tidal_playlist:
|
spotify_tracks = await get_tracks_from_spotify_playlist(spotify_session, spotify_playlist)
|
||||||
|
if len(spotify_tracks) == 0:
|
||||||
|
return # nothing to do
|
||||||
|
if tidal_playlist:
|
||||||
|
old_tidal_tracks = await get_all_playlist_tracks(tidal_playlist)
|
||||||
|
else:
|
||||||
print(f"No playlist found on Tidal corresponding to Spotify playlist: '{spotify_playlist['name']}', creating new playlist")
|
print(f"No playlist found on Tidal corresponding to Spotify playlist: '{spotify_playlist['name']}', creating new playlist")
|
||||||
tidal_playlist = tidal_session.user.create_playlist(spotify_playlist['name'], spotify_playlist['description'])
|
tidal_playlist = tidal_session.user.create_playlist(spotify_playlist['name'], spotify_playlist['description'])
|
||||||
|
old_tidal_tracks = []
|
||||||
|
|
||||||
# Extract the new tracks from the playlist that we haven't already seen before
|
# Extract the new tracks from the playlist that we haven't already seen before
|
||||||
spotify_tracks = await get_tracks_from_spotify_playlist(spotify_session, spotify_playlist)
|
|
||||||
old_tidal_tracks = await get_all_playlist_tracks(tidal_playlist)
|
|
||||||
populate_track_match_cache(spotify_tracks, old_tidal_tracks)
|
populate_track_match_cache(spotify_tracks, old_tidal_tracks)
|
||||||
await search_new_tracks_on_tidal(tidal_session, spotify_tracks, spotify_playlist['name'], config)
|
await search_new_tracks_on_tidal(tidal_session, spotify_tracks, spotify_playlist['name'], config)
|
||||||
new_tidal_track_ids = get_tracks_for_new_tidal_playlist(spotify_tracks)
|
new_tidal_track_ids = get_tracks_for_new_tidal_playlist(spotify_tracks)
|
||||||
@@ -381,7 +387,7 @@ async def get_playlists_from_spotify(spotify_session: spotipy.Spotify, config):
|
|||||||
playlists.extend([p for p in extra_result['items']])
|
playlists.extend([p for p in extra_result['items']])
|
||||||
|
|
||||||
# filter out playlists that don't belong to us or are on the exclude list
|
# filter out playlists that don't belong to us or are on the exclude list
|
||||||
my_playlist_filter = lambda p: p['owner']['id'] == user_id
|
my_playlist_filter = lambda p: p and p['owner']['id'] == user_id
|
||||||
exclude_filter = lambda p: not p['id'] in exclude_list
|
exclude_filter = lambda p: not p['id'] in exclude_list
|
||||||
return list(filter( exclude_filter, filter( my_playlist_filter, playlists )))
|
return list(filter( exclude_filter, filter( my_playlist_filter, playlists )))
|
||||||
|
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ def test_open_spotify_session(mocker):
|
|||||||
"client_id": "test_client_id",
|
"client_id": "test_client_id",
|
||||||
"client_secret": "test_client_secret",
|
"client_secret": "test_client_secret",
|
||||||
"redirect_uri": "http://localhost/",
|
"redirect_uri": "http://localhost/",
|
||||||
|
"open_browser": True,
|
||||||
}
|
}
|
||||||
|
|
||||||
# Create a mock SpotifyOAuth instance
|
# Create a mock SpotifyOAuth instance
|
||||||
@@ -41,6 +42,7 @@ def test_open_spotify_session(mocker):
|
|||||||
client_secret="test_client_secret",
|
client_secret="test_client_secret",
|
||||||
redirect_uri="http://localhost/",
|
redirect_uri="http://localhost/",
|
||||||
requests_timeout=2,
|
requests_timeout=2,
|
||||||
|
open_browser=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Assert that the Spotify instance was created
|
# Assert that the Spotify instance was created
|
||||||
|
|||||||
Reference in New Issue
Block a user