From 4e0c81071b6175f2515fc4175abc5e42b0e86560 Mon Sep 17 00:00:00 2001 From: joshrmcdaniel <80354972+joshrmcdaniel@users.noreply.github.com> Date: Sun, 12 May 2024 10:20:49 -0500 Subject: [PATCH 1/4] move to src --- auth.py => src/spotify_to_tidal/auth.py | 9 +++++++-- sync.py => src/spotify_to_tidal/sync.py | 0 2 files changed, 7 insertions(+), 2 deletions(-) rename auth.py => src/spotify_to_tidal/auth.py (91%) rename sync.py => src/spotify_to_tidal/sync.py (100%) diff --git a/auth.py b/src/spotify_to_tidal/auth.py similarity index 91% rename from auth.py rename to src/spotify_to_tidal/auth.py index bda3eb1..2bc2bba 100644 --- a/auth.py +++ b/src/spotify_to_tidal/auth.py @@ -6,7 +6,12 @@ import tidalapi import webbrowser import yaml -def open_spotify_session(config): +__all__ = [ + 'open_spotify_session', + 'open_tidal_session' +] + +def open_spotify_session(config) -> spotipy.Spotify: credentials_manager = spotipy.SpotifyOAuth(username=config['username'], scope='playlist-read-private', client_id=config['client_id'], @@ -19,7 +24,7 @@ def open_spotify_session(config): return spotipy.Spotify(oauth_manager=credentials_manager) -def open_tidal_session(config = None): +def open_tidal_session(config = None) -> tidalapi.Session: try: with open('.session.yml', 'r') as session_file: previous_session = yaml.safe_load(session_file) diff --git a/sync.py b/src/spotify_to_tidal/sync.py similarity index 100% rename from sync.py rename to src/spotify_to_tidal/sync.py From 6aaf72bdd1c5dd21a814cfbe5497100dff332059 Mon Sep 17 00:00:00 2001 From: joshrmcdaniel <80354972+joshrmcdaniel@users.noreply.github.com> Date: Sun, 12 May 2024 10:21:47 -0500 Subject: [PATCH 2/4] type hint, move --- src/spotify_to_tidal/__main__.py | 35 +++++++++++++++++++++++ src/spotify_to_tidal/sync.py | 49 +++++++++----------------------- 2 files changed, 48 insertions(+), 36 deletions(-) create mode 100644 src/spotify_to_tidal/__main__.py diff --git a/src/spotify_to_tidal/__main__.py b/src/spotify_to_tidal/__main__.py new file mode 100644 index 0000000..d4abebc --- /dev/null +++ b/src/spotify_to_tidal/__main__.py @@ -0,0 +1,35 @@ +import yaml +import argparse +import sys + +from . import sync as _sync +from . import auth as _auth + +def main(): + parser = argparse.ArgumentParser() + 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') + args = parser.parse_args() + + with open(args.config, 'r') as f: + config = yaml.safe_load(f) + spotify_session = _auth.open_spotify_session(config['spotify']) + tidal_session = _auth.open_tidal_session() + if not tidal_session.check_login(): + sys.exit("Could not connect to Tidal") + if args.uri: + # if a playlist ID is explicitly provided as a command line argument then use that + spotify_playlist = spotify_session.playlist(args.uri) + tidal_playlists = _sync.get_tidal_playlists_dict(tidal_session) + tidal_playlist = _sync.pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists) + _sync.sync_list(spotify_session, tidal_session, [tidal_playlist], config) + elif config.get('sync_playlists', None): + # if the config contains a sync_playlists list of mappings then use that + _sync.sync_list(spotify_session, tidal_session, _sync.get_playlists_from_config(config), config) + else: + # otherwise just use the user playlists in the Spotify account + _sync.sync_list(spotify_session, tidal_session, _sync.get_user_playlist_mappings(spotify_session, tidal_session, config), config) + +if __name__ == '__main__': + main() + sys.exit(0) \ No newline at end of file diff --git a/src/spotify_to_tidal/sync.py b/src/spotify_to_tidal/sync.py index ea0790e..7d51928 100755 --- a/src/spotify_to_tidal/sync.py +++ b/src/spotify_to_tidal/sync.py @@ -1,8 +1,9 @@ #!/usr/bin/env python3 import argparse -from auth import open_tidal_session, open_spotify_session +from .auth import open_tidal_session, open_spotify_session from functools import partial +from typing import Sequence, Set from multiprocessing import Pool import requests import sys @@ -15,24 +16,25 @@ import traceback import unicodedata import yaml + def normalize(s): return unicodedata.normalize('NFD', s).encode('ascii', 'ignore').decode('ascii') -def simple(input_string): +def simple(input_string: str) -> str: # only take the first part of a string before any hyphens or brackets to account for different versions return input_string.split('-')[0].strip().split('(')[0].strip().split('[')[0].strip() -def isrc_match(tidal_track, spotify_track): +def isrc_match(tidal_track: tidalapi.Track, spotify_track): if "isrc" in spotify_track["external_ids"]: return tidal_track.isrc == spotify_track["external_ids"]["isrc"] return False -def duration_match(tidal_track, spotify_track, tolerance=2): +def duration_match(tidal_track: tidalapi.Track, spotify_track, tolerance=2) -> float: # the duration of the two tracks must be the same to within 2 seconds return abs(tidal_track.duration - spotify_track['duration_ms']/1000) < tolerance -def name_match(tidal_track, spotify_track): - def exclusion_rule(pattern, tidal_track, spotify_track): +def name_match(tidal_track, spotify_track) -> bool: + def exclusion_rule(pattern: str, tidal_track: tidalapi.Track, spotify_track): spotify_has_pattern = pattern in spotify_track['name'].lower() tidal_has_pattern = pattern in tidal_track.name.lower() or (not tidal_track.version is None and (pattern in tidal_track.version.lower())) return spotify_has_pattern != tidal_has_pattern @@ -47,8 +49,8 @@ def name_match(tidal_track, spotify_track): simple_spotify_track = simple(spotify_track['name'].lower()).split('feat.')[0].strip() return simple_spotify_track in tidal_track.name.lower() or normalize(simple_spotify_track) in normalize(tidal_track.name.lower()) -def artist_match(tidal_track, spotify_track): - def split_artist_name(artist): +def artist_match(tidal_track: tidalapi.Track, spotify_track) -> Set[str]: + def split_artist_name(artist: str) -> Sequence[str]: if '&' in artist: return artist.split('&') elif ',' in artist: @@ -56,7 +58,7 @@ def artist_match(tidal_track, spotify_track): else: return [artist] - def get_tidal_artists(tidal_track, do_normalize=False): + def get_tidal_artists(tidal_track: tidalapi.Track, do_normalize=False) -> Set[str]: result = [] for artist in tidal_track.artists: if do_normalize: @@ -66,7 +68,7 @@ def artist_match(tidal_track, spotify_track): result.extend(split_artist_name(artist_name)) return set([simple(x.strip().lower()) for x in result]) - def get_spotify_artists(spotify_track, do_normalize=False): + def get_spotify_artists(spotify_track, do_normalize=False) -> Set[str]: result = [] for artist in spotify_track['artists']: if do_normalize: @@ -284,29 +286,4 @@ def get_playlists_from_spotify(spotify_session, config): def get_playlists_from_config(config): # get the list of playlist sync mappings from the configuration file - return [(item['spotify_id'], item['tidal_id']) for item in config['sync_playlists']] - -if __name__ == '__main__': - parser = argparse.ArgumentParser() - 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') - args = parser.parse_args() - - with open(args.config, 'r') as f: - config = yaml.safe_load(f) - spotify_session = open_spotify_session(config['spotify']) - tidal_session = open_tidal_session() - if not tidal_session.check_login(): - sys.exit("Could not connect to Tidal") - if args.uri: - # if a playlist ID is explicitly provided as a command line argument then use that - spotify_playlist = spotify_session.playlist(args.uri) - tidal_playlists = get_tidal_playlists_dict(tidal_session) - tidal_playlist = pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists) - sync_list(spotify_session, tidal_session, [tidal_playlist], config) - elif config.get('sync_playlists', None): - # if the config contains a sync_playlists list of mappings then use that - sync_list(spotify_session, tidal_session, get_playlists_from_config(config), config) - else: - # otherwise just use the user playlists in the Spotify account - sync_list(spotify_session, tidal_session, get_user_playlist_mappings(spotify_session, tidal_session, config), config) + return [(item['spotify_id'], item['tidal_id']) for item in config['sync_playlists']] \ No newline at end of file From 76f502f2bcdeb1f327aca738ae4ed6ab1d4b92df Mon Sep 17 00:00:00 2001 From: joshrmcdaniel <80354972+joshrmcdaniel@users.noreply.github.com> Date: Sun, 12 May 2024 10:46:36 -0500 Subject: [PATCH 3/4] types --- src/spotify_to_tidal/sync.py | 44 ++++++------ .../spotify_to_tidal/tidalapi_patch.py | 0 src/spotify_to_tidal/type/__init__.py | 26 +++++++ src/spotify_to_tidal/type/spotify.py | 69 +++++++++++++++++++ 4 files changed, 116 insertions(+), 23 deletions(-) rename tidalapi_patch.py => src/spotify_to_tidal/tidalapi_patch.py (100%) create mode 100644 src/spotify_to_tidal/type/__init__.py create mode 100644 src/spotify_to_tidal/type/spotify.py diff --git a/src/spotify_to_tidal/sync.py b/src/spotify_to_tidal/sync.py index 7d51928..f18f996 100755 --- a/src/spotify_to_tidal/sync.py +++ b/src/spotify_to_tidal/sync.py @@ -1,30 +1,29 @@ #!/usr/bin/env python3 -import argparse -from .auth import open_tidal_session, open_spotify_session from functools import partial -from typing import Sequence, Set +from typing import Sequence, Set, Mapping from multiprocessing import Pool import requests import sys import spotipy import tidalapi -from tidalapi_patch import set_tidal_playlist +from .tidalapi_patch import set_tidal_playlist import time from tqdm import tqdm import traceback import unicodedata -import yaml + +from .type import spotify as t_spotify -def normalize(s): +def normalize(s) -> str: return unicodedata.normalize('NFD', s).encode('ascii', 'ignore').decode('ascii') def simple(input_string: str) -> str: # only take the first part of a string before any hyphens or brackets to account for different versions return input_string.split('-')[0].strip().split('(')[0].strip().split('[')[0].strip() -def isrc_match(tidal_track: tidalapi.Track, spotify_track): +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"] return False @@ -34,7 +33,7 @@ def duration_match(tidal_track: tidalapi.Track, spotify_track, tolerance=2) -> f return abs(tidal_track.duration - spotify_track['duration_ms']/1000) < tolerance def name_match(tidal_track, spotify_track) -> bool: - def exclusion_rule(pattern: str, tidal_track: tidalapi.Track, spotify_track): + def exclusion_rule(pattern: str, tidal_track: tidalapi.Track, spotify_track: t_spotify.SpotifyTrack): spotify_has_pattern = pattern in spotify_track['name'].lower() tidal_has_pattern = pattern in tidal_track.name.lower() or (not tidal_track.version is None and (pattern in tidal_track.version.lower())) return spotify_has_pattern != tidal_has_pattern @@ -68,7 +67,7 @@ def artist_match(tidal_track: tidalapi.Track, spotify_track) -> Set[str]: result.extend(split_artist_name(artist_name)) return set([simple(x.strip().lower()) for x in result]) - def get_spotify_artists(spotify_track, do_normalize=False) -> Set[str]: + def get_spotify_artists(spotify_track: t_spotify.SpotifyTrack, do_normalize=False) -> Set[str]: result = [] for artist in spotify_track['artists']: if do_normalize: @@ -83,7 +82,7 @@ def artist_match(tidal_track: tidalapi.Track, spotify_track) -> Set[str]: return True return get_tidal_artists(tidal_track, True).intersection(get_spotify_artists(spotify_track, True)) != set() -def match(tidal_track, spotify_track): +def match(tidal_track, spotify_track) -> bool: return isrc_match(tidal_track, spotify_track) or ( duration_match(tidal_track, spotify_track) and name_match(tidal_track, spotify_track) @@ -91,7 +90,7 @@ def match(tidal_track, spotify_track): ) -def tidal_search(spotify_track_and_cache, tidal_session): +def tidal_search(spotify_track_and_cache, tidal_session: tidalapi.Session) -> tidalapi.Track | None: spotify_track, cached_tidal_track = spotify_track_and_cache if cached_tidal_track: return cached_tidal_track # search for album name and first album artist @@ -108,7 +107,7 @@ def tidal_search(spotify_track_and_cache, tidal_session): if match(track, spotify_track): return track -def get_tidal_playlists_dict(tidal_session): +def get_tidal_playlists_dict(tidal_session: tidalapi.Session) -> Mapping[str, tidalapi.Playlist]: # a dictionary of name --> playlist tidal_playlists = tidal_session.user.playlists() output = {} @@ -152,7 +151,7 @@ def call_async_with_progress(function, values, description, num_processes, **kwa results[index] = result return results -def get_tracks_from_spotify_playlist(spotify_session, spotify_playlist): +def get_tracks_from_spotify_playlist(spotify_session: spotipy.Spotify, spotify_playlist): output = [] results = spotify_session.playlist_tracks( spotify_playlist["id"], @@ -167,10 +166,10 @@ def get_tracks_from_spotify_playlist(spotify_session, spotify_playlist): return output class TidalPlaylistCache: - def __init__(self, playlist): + def __init__(self, playlist: tidalapi.Playlist): self._data = playlist.tracks() - def _search(self, spotify_track): + def _search(self, spotify_track: t_spotify.SpotifyTrack): ''' check if the given spotify track was already in the tidal playlist.''' results = [] for tidal_track in self._data: @@ -178,7 +177,7 @@ class TidalPlaylistCache: return tidal_track return None - def search(self, spotify_session, spotify_playlist): + def search(self, spotify_session: spotipy.Spotify, spotify_playlist): ''' Add the cached tidal track where applicable to a list of spotify tracks ''' results = [] cache_hits = 0 @@ -193,7 +192,7 @@ class TidalPlaylistCache: results.append( (track, None) ) return (results, cache_hits) -def tidal_playlist_is_dirty(playlist, new_track_ids): +def tidal_playlist_is_dirty(playlist: tidalapi.Playlist, new_track_ids: Sequence[str]) -> bool: old_tracks = playlist.tracks() if len(old_tracks) != len(new_track_ids): return True @@ -202,13 +201,12 @@ def tidal_playlist_is_dirty(playlist, new_track_ids): return True return False -def sync_playlist(spotify_session, tidal_session, spotify_id, tidal_id, config): +def sync_playlist(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, spotify_id: str, tidal_id: int, config): try: spotify_playlist = spotify_session.playlist(spotify_id) except spotipy.SpotifyException as e: print("Error getting Spotify playlist " + spotify_id) print(e) - results.append(None) return if tidal_id: # if a Tidal playlist was specified then look it up @@ -243,7 +241,7 @@ def sync_playlist(spotify_session, tidal_session, spotify_id, tidal_id, config): else: print("No changes to write to Tidal playlist") -def sync_list(spotify_session, tidal_session, playlists, config): +def sync_list(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, playlists: Mapping[str, tidalapi.Playlist], config): results = [] for spotify_id, tidal_id in playlists: # sync the spotify playlist to tidal @@ -251,7 +249,7 @@ def sync_list(spotify_session, tidal_session, playlists, config): results.append(tidal_id) return results -def pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists): +def pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists: Mapping[str, tidalapi.Playlist]): if spotify_playlist['name'] in tidal_playlists: # if there's an existing tidal playlist with the name of the current playlist then use that tidal_playlist = tidal_playlists[spotify_playlist['name']] @@ -260,7 +258,7 @@ def pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists): return (spotify_playlist['id'], None) -def get_user_playlist_mappings(spotify_session, tidal_session, config): +def get_user_playlist_mappings(spotify_session: spotipy.Spotify, tidal_session: tidalapi.Session, config): results = [] spotify_playlists = get_playlists_from_spotify(spotify_session, config) tidal_playlists = get_tidal_playlists_dict(tidal_session) @@ -268,7 +266,7 @@ def get_user_playlist_mappings(spotify_session, tidal_session, config): results.append( pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists) ) return results -def get_playlists_from_spotify(spotify_session, config): +def get_playlists_from_spotify(spotify_session: spotipy.Spotify, config): # get all the user playlists from the Spotify account playlists = [] spotify_results = spotify_session.user_playlists(config['spotify']['username']) diff --git a/tidalapi_patch.py b/src/spotify_to_tidal/tidalapi_patch.py similarity index 100% rename from tidalapi_patch.py rename to src/spotify_to_tidal/tidalapi_patch.py diff --git a/src/spotify_to_tidal/type/__init__.py b/src/spotify_to_tidal/type/__init__.py new file mode 100644 index 0000000..58a4108 --- /dev/null +++ b/src/spotify_to_tidal/type/__init__.py @@ -0,0 +1,26 @@ +from .config import SpotifyConfig, TidalConfig, PlaylistConfig, SyncConfig +from .playlist import TidalPlaylist +from .spotify import SpotifyTrack + +from spotipy import Spotify +from tidalapi import Session, Track + +TidalID = str +SpotifyID = str +TidalSession = Session +TidalTrack = Track +SpotifySession = Spotify + +__all__ = [ + "SpotifyConfig", + "TidalConfig", + "PlaylistConfig", + "SyncConfig", + "TidalPlaylist", + "TidalID", + "SpotifyID", + "SpotifySession", + "TidalSession", + "TidalTrack", + "SpotifyTrack", +] diff --git a/src/spotify_to_tidal/type/spotify.py b/src/spotify_to_tidal/type/spotify.py new file mode 100644 index 0000000..3970ad5 --- /dev/null +++ b/src/spotify_to_tidal/type/spotify.py @@ -0,0 +1,69 @@ +from spotipy import Spotify +from typing import TypedDict, List, Dict, Mapping, Literal, Optional + + +class SpotifyImage(TypedDict): + url: str + height: int + width: int + + +class SpotifyFollower(TypedDict): + href: str + total: int + + +SpotifyID = str +SpotifySession = Spotify + + +class SpotifyArtist(TypedDict): + external_urls: Mapping[str, str] + followers: SpotifyFollower + genres: List[str] + href: str + id: str + images: List[SpotifyImage] + name: str + popularity: int + type: str + uri: str + + +class SpotifyAlbum(TypedDict): + album_type: Literal["album", "single", "compilation"] + total_tracks: int + available_markets: List[str] + external_urls: Dict[str, str] + href: str + id: str + images: List[SpotifyImage] + name: str + release_date: str + release_date_precision: Literal["year", "month", "day"] + restrictions: Optional[Dict[Literal["reason"], str]] + type: Literal["album"] + uri: str + artists: List[SpotifyArtist] + + +class SpotifyTrack(TypedDict): + album: SpotifyAlbum + artists: List[SpotifyArtist] + available_markets: List[str] + disc_number: int + duration_ms: int + explicit: bool + external_ids: Dict[str, str] + external_urls: Dict[str, str] + href: str + id: str + is_playable: bool + linked_from: Dict + restrictions: Optional[Dict[Literal["reason"], str]] + name: str + popularity: int + preview_url: str + track_number: int + type: Literal["track"] + uri: str From 9e3285686ef229dcc5128aaa2419b023bdc3b9bf Mon Sep 17 00:00:00 2001 From: joshrmcdaniel <80354972+joshrmcdaniel@users.noreply.github.com> Date: Sun, 12 May 2024 10:46:57 -0500 Subject: [PATCH 4/4] toml --- pyproject.toml | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 pyproject.toml diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f4ad425 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools >= 61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "spotify_to_tidal" +version = "0.0.1-dev" +requires-python = ">= 3.10" + +dependencies = [ + "spotipy~=2.21", + "tidalapi~=0.7", + "pyyaml~=6.0", + "tqdm~=4.64", +] + + +[tools.setuptools.packages."spotify_to_tidal"] +where = "src" +include = "spotify_to_tidal*" + +[project.scripts] +spotify_to_tidal = "spotify_to_tidal.__main__:main"