diff --git a/auth.py b/auth.py deleted file mode 100644 index bda3eb1..0000000 --- a/auth.py +++ /dev/null @@ -1,56 +0,0 @@ -#!/usr/bin/env python3 - -import sys -import spotipy -import tidalapi -import webbrowser -import yaml - -def open_spotify_session(config): - credentials_manager = spotipy.SpotifyOAuth(username=config['username'], - scope='playlist-read-private', - client_id=config['client_id'], - client_secret=config['client_secret'], - redirect_uri=config['redirect_uri']) - try: - credentials_manager.get_access_token(as_dict=False) - except spotipy.SpotifyOauthError: - sys.exit("Error opening Spotify sesion; could not get token for username: ".format(config['username'])) - - return spotipy.Spotify(oauth_manager=credentials_manager) - -def open_tidal_session(config = None): - try: - with open('.session.yml', 'r') as session_file: - previous_session = yaml.safe_load(session_file) - except OSError: - previous_session = None - - if config: - session = tidalapi.Session(config=config) - else: - session = tidalapi.Session() - if previous_session: - try: - if session.load_oauth_session(token_type= previous_session['token_type'], - access_token=previous_session['access_token'], - refresh_token=previous_session['refresh_token'] ): - return session - except Exception as e: - print("Error loading previous Tidal Session: \n" + str(e) ) - - login, future = session.login_oauth() - print('Login with the webbrowser: ' + login.verification_uri_complete) - url = login.verification_uri_complete - if not url.startswith('https://'): - url = 'https://' + url - webbrowser.open(url) - future.result() - with open('.session.yml', 'w') as f: - yaml.dump( {'session_id': session.session_id, - 'token_type': session.token_type, - 'access_token': session.access_token, - 'refresh_token': session.refresh_token}, f ) - return session - - diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f5cb219 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools >= 61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "spotify_to_tidal" +version = "0.1.0" +requires-python = ">= 3.10" + +dependencies = [ + "spotipy==2.21.0", + "tidalapi==0.7.3", + "pyyaml==6.0", + "tqdm==4.64.1", +] + + +[tools.setuptools.packages."spotify_to_tidal"] +where = "src" +include = "spotify_to_tidal*" + +[project.scripts] +spotify_to_tidal = "spotify_to_tidal.__main__:main" + diff --git a/readme.md b/readme.md index baba328..b39e900 100644 --- a/readme.md +++ b/readme.md @@ -5,7 +5,8 @@ Installation Clone this git repository and then run: ```bash -python3 -m pip install -r requirements.txt +python3 -m pip install -e . +spotify_to_tidal -h ``` Setup @@ -21,6 +22,8 @@ Usage To synchronize all of your Spotify playlists with your Tidal account run the following ```bash +pip install -e . +spotify_to_tidal python3 sync.py ``` @@ -29,7 +32,7 @@ This will take a long time because the Tidal API is really slow. You can also just synchronize a specific playlist by doing the following: ```bash -python3 sync.py --uri 1ABCDEqsABCD6EaABCDa0a +spotify_to_tidal --uri 1ABCDEqsABCD6EaABCDa0a ``` See example_config.yml for more configuration options, and `sync.py --help` for more options. diff --git a/requirements.txt b/requirements.txt index b7c7718..a08a901 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,4 @@ requests>=2.28.1 # for tidalapi tidalapi==0.7.2 pyyaml==6.0 tqdm==4.64.1 +cachetools==5.3.2 \ No newline at end of file diff --git a/src/spotify_to_tidal/__init__.py b/src/spotify_to_tidal/__init__.py new file mode 100644 index 0000000..c559c04 --- /dev/null +++ b/src/spotify_to_tidal/__init__.py @@ -0,0 +1,42 @@ +from .auth import open_tidal_session, open_spotify_session +from .parse import ( + get_tidal_playlists_dict, + playlist_id_tuple, + create_playlist_id_tuple, + +) +from .sync import sync_list + +from .type import ( + SpotifyConfig, + TidalConfig, + PlaylistConfig, + SyncConfig, + TidalPlaylist, + TidalID, + SpotifyID, + SpotifySession, + TidalSession, + TidalTrack, + SpotifyTrack, +) + +__all__ = [ + "open_tidal_session", + "open_spotify_session", + "get_tidal_playlists_dict", + "playlist_id_tuple", + "create_playlist_id_tuple", + "sync_list" + "SpotifyConfig", + "TidalConfig", + "PlaylistConfig", + "SyncConfig", + "TidalPlaylist", + "TidalID", + "SpotifyID", + "SpotifySession", + "TidalSession", + "TidalTrack", + "SpotifyTrack", +] \ No newline at end of file diff --git a/src/spotify_to_tidal/__main__.py b/src/spotify_to_tidal/__main__.py new file mode 100644 index 0000000..8915a04 --- /dev/null +++ b/src/spotify_to_tidal/__main__.py @@ -0,0 +1,174 @@ +import logging +import argparse +import sys +import yaml +from pathlib import Path +from .filters import Filter429 +from . import ( + open_tidal_session, open_spotify_session, + get_tidal_playlists_dict, + playlist_id_tuple, + sync_list, + create_playlist_id_tuple, + SyncConfig, SpotifyConfig, +) + +from typing import NoReturn + + +def setup_args() -> argparse.ArgumentParser: + synopsis = """ +Syncs spotify playlists to Tidal. Can specify a config yaml or specify Spotify Oauth values on the command line. + +""" + parser = argparse.ArgumentParser(description=synopsis) + + parser.add_argument( + "-c", + "--config", + type=Path, + help="Location of the spotify to tidal config file", + dest="config", + ) + parser.add_argument( + "-i", "--id", type=str, help="Spotify client ID", dest="client_id" + ) + parser.add_argument( + "-s", "--secret", type=str, help="Spotify client secret", dest="client_secret" + ) + parser.add_argument( + "-U", + "--username", + type=str, + help="Spotify username of the Oauth creds", + dest="spotify_uname", + default=None, + ) + parser.add_argument( + "-u", + "--uri", + help="Synchronize specific URI(s) instead of the one in the config", + nargs='*', + dest="uri", + ) + parser.add_argument( + "-v", + help="verbosity level, increases per v (max of 3) specified", + default=0, + action="count", + dest="verbosity", + ) + parser.add_argument( + "-a", + "--all", + help="Sync all spotify playlists to Tidal. This will not honor the config's exclude list.", + action="store_true", + default=False, + dest="all_playlists", + ) + parser.add_argument( + "-e", + "--exclude", + help="ID of Spotify playlists to exclude", + nargs="*", + dest="exclude_ids", + type=set, + ) + + return parser + + +logger = None + + +def setup_logging(verbosity: int) -> None: + log_map = [logging.WARNING, logging.INFO, logging.DEBUG] + strm_hndl = logging.StreamHandler(sys.stdout) + logging.root.addFilter(Filter429("tidalapi")) + logging.root.addFilter(Filter429("spotipy")) + global logger + fmt = logging.Formatter( + "[%(asctime)s] %(levelname)s %(module)s:%(funcName)s:%(lineno)d - %(message)s" + ) + strm_hndl.setFormatter(fmt) + logger = logging.getLogger(__package__) + logger.setLevel(log_map[min(verbosity, 2)]) + logger.addHandler(strm_hndl) + logger.debug("Initialized logging.") + + +def parse_args(parser: argparse.ArgumentParser) -> argparse.Namespace | NoReturn: + args = parser.parse_args() + setup_logging(args.verbosity) + if args.config is None: + logging.debug("No config specified, checking other args.") + if args.client_id is None: + raise RuntimeError( + "No config specified and Spotify client ID not specified." + ) + if args.client_secret is None: + raise RuntimeError( + "No config specified and Spotify secret ID not specified." + ) + if args.spotify_uname is None: + raise RuntimeError( + "No config specified and Spotify username not specified." + ) + if args.config and any( + x is not None for x in (args.client_id, args.client_secret, args.spotify_uname) + ): + raise RuntimeError( + "Config specfied with config attributes. Only specify a config or all attributes." + ) + if args.exclude_ids is None: + args.exclude_ids = set() + # TODO: more validation? + return args + + +def main(): + parser = setup_args() + args = parse_args(parser) + if args.config: + with open(args.config, "r") as f: + config: SyncConfig = yaml.safe_load(f) + args.exclude_ids.update(*config.get("excluded_playlists", [])) + spotify_cfg: SpotifyConfig = config.get("spotify", {}) + args.spotify_uname = spotify_cfg.get("username") + args.client_secret = spotify_cfg.get("client_secret") + args.client_id = spotify_cfg.get("client_id") + args.username = spotify_cfg.get("username") + redirect_uri = spotify_cfg.get("redirect_uri", "http://localhost:8888/callback") + else: + args.config = {} + redirect_uri = "http://localhost:8888/callback" + spotify_session = open_spotify_session( + username=args.username, + client_id=args.client_id, + client_secret=args.client_secret, + redirect_uri=redirect_uri, + ) + tidal_session = open_tidal_session() + tidal_playlists = get_tidal_playlists_dict(tidal_session) + id_tuples = [] + if args.uri: + # if a playlist ID is explicitly provided as a command line argument then use that + for uri in args.uri: + spotify_playlist = spotify_session.playlist(uri) + id_tuples.append(playlist_id_tuple(spotify_playlist, tidal_playlists)) + elif args.config and (x := config.get("sync_playlists", [])): + for ids in x: + id_tuples.append((ids["spotify_id"], ids.get("tidal_id"))) + elif args.all_playlists or config.get("sync_playlists", None): + id_tuples = create_playlist_id_tuple( + spotify_session, tidal_session, args.exclude_ids + ) + sync_list(spotify_session, tidal_session, id_map, config) + logger.info("Syncing %d playlists", len(id_tuples)) + + sync_list( + spotify_session, + tidal_session, + id_tuples, + config, + ) diff --git a/src/spotify_to_tidal/auth.py b/src/spotify_to_tidal/auth.py new file mode 100644 index 0000000..5b97345 --- /dev/null +++ b/src/spotify_to_tidal/auth.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +import logging +import sys + +from .type import * +from typing import Union, NoReturn, Optional +import spotipy +import tidalapi +import webbrowser +import yaml + +logger = logging.getLogger(__name__) + + +def open_spotify_session( + *, username: str, client_id: str, client_secret: str, redirect_uri: str +) -> Union[spotipy.Spotify, NoReturn]: + credentials_manager = spotipy.SpotifyOAuth( + username=username, + scope="playlist-read-private", + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + ) + try: + credentials_manager.get_access_token(as_dict=False) + except spotipy.SpotifyOauthError: + logger.critical( + "Error opening Spotify sesion; could not get token for username: %s", + username, + ) + sys.exit(1) + + return spotipy.Spotify(oauth_manager=credentials_manager) + + +def open_tidal_session(config: Optional[tidalapi.Config] = None) -> tidalapi.Session: + try: + with open(".session.yml", "r") as session_file: + previous_session: TidalConfig = yaml.safe_load(session_file) + except OSError: + previous_session = None + + if config: + session = tidalapi.Session(config=config) + else: + session = tidalapi.Session() + if previous_session: + try: + if session.load_oauth_session( + token_type=previous_session["token_type"], + access_token=previous_session["access_token"], + refresh_token=previous_session["refresh_token"], + ): + return session + except Exception as e: + logger.warn("Error loading previous Tidal Session") + logger.debug(e) + + if not session.check_login(): + logging.critical("Could not connect to Tidal") + sys.exit(1) + + login, future = session.login_oauth() + print("Login with the webbrowser: " + login.verification_uri_complete) + url = login.verification_uri_complete + if not url.startswith("https://"): + url = "https://" + url + webbrowser.open(url) + future.result() + + if not session.check_login(): + logging.critical("Could not connect to Tidal") + sys.exit(1) + with open(".session.yml", "w") as f: + yaml.dump( + { + "session_id": session.session_id, + "token_type": session.token_type, + "access_token": session.access_token, + "refresh_token": session.refresh_token, + }, + f, + ) + return session diff --git a/src/spotify_to_tidal/filters.py b/src/spotify_to_tidal/filters.py new file mode 100644 index 0000000..c3475ae --- /dev/null +++ b/src/spotify_to_tidal/filters.py @@ -0,0 +1,119 @@ +import unicodedata +from .type import * +from typing import List +import logging +from itertools import chain + +pkg_logger = logging.getLogger(__package__) +logger = logging.getLogger(__name__) +for hndl in pkg_logger.handlers: + logger.addHandler(hndl) + + +def normalize(s: str) -> 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: TidalTrack, spotify_track: SpotifyTrack) -> bool: + if "isrc" in spotify_track["external_ids"]: + return tidal_track.isrc == spotify_track["external_ids"]["isrc"] + return False + + +def duration_match( + tidal_track: TidalTrack, spotify_track: SpotifyTrack, tolerance=2 +) -> bool: + # 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: TidalTrack, spotify_track: SpotifyTrack) -> bool: + def exclusion_rule(pattern: str): + spotify_has_pattern = pattern in spotify_track["name"].casefold() + tidal_has_pattern = pattern in tidal_track.name.casefold() or ( + not tidal_track.version is None + and (pattern in tidal_track.version.casefold()) + ) + return spotify_has_pattern or tidal_has_pattern + + # handle some edge cases + if exclusion_rule("instrumental"): + return False + if exclusion_rule("acapella"): + return False + if exclusion_rule("remix"): + return False + + # the simplified version of the Spotify track name must be a substring of the Tidal track name + # Try with both un-normalized and then normalized + simple_spotify_track = simple(spotify_track["name"].casefold()) + if "feat." in simple_spotify_track: + kw = ".feat" + elif "ft." in simple_spotify_track: + kw = ".ft" + else: + kw = "" + if kw != "": + simple_spotify_track = simple_spotify_track.split(kw)[0].strip() + return simple_spotify_track in tidal_track.name.casefold() or normalize( + simple_spotify_track + ) in normalize(tidal_track.name.casefold()) + + +def artist_match(tidal_track: TidalTrack, spotify_track: SpotifyTrack) -> bool: + def split_artist_name(artist: str) -> List[str]: + if "&" in artist: + return artist.split("&") + elif "," in artist: + return artist.split(",") + else: + return [artist] + + def split_and_clean(artist: str) -> List[str]: + return map(lambda x: x.strip().casefold(), split_artist_name(artist)) + + spotify_artists = set( + chain.from_iterable( + map(lambda x: split_and_clean(x["name"]), spotify_track["artists"]) + ) + ) + spotify_artists_normalized = set(map(normalize, spotify_artists)) + tidal_artists = chain(map(lambda x: x.name, tidal_track.artists)) + tidal_artists = set(chain.from_iterable(map(split_and_clean, tidal_artists))) + tidal_artists_normalized = set(map(normalize, tidal_artists)) + + return tidal_artists.intersection( + spotify_artists + ) or tidal_artists_normalized.intersection(spotify_artists_normalized) + + +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) + and artist_match(tidal_track, spotify_track) + ) + + +class FilterOtherPkgs(logging.Filter): + def filter(self, record: logging.LogRecord): + return ( + record.module.split(".")[0] != __package__ + and record.levelno <= logging.DEBUG + ) + + +class Filter429(logging.Filter): + def filter(self, record: logging.LogRecord): + return ( + record.getMessage() == "HTTP error on 429" + or record.getMessage() == "HTTP error on 412" + or not record.module.startswith(__package__) + ) diff --git a/src/spotify_to_tidal/parse.py b/src/spotify_to_tidal/parse.py new file mode 100644 index 0000000..991c58e --- /dev/null +++ b/src/spotify_to_tidal/parse.py @@ -0,0 +1,80 @@ +from typing import Any, Dict, List, Tuple, Callable, Optional, Set +from .type import * + + +def get_tidal_playlists_dict(tidal_session: TidalSession) -> Dict[str, TidalPlaylist]: + # a dictionary of name --> playlist + tidal_playlists = tidal_session.user.playlists() + output = {} + for playlist in tidal_playlists: + output[playlist.name] = playlist + return output + + +def playlist_id_tuple( + spotify_playlist: Dict[str, Any], tidal_playlists: Dict[str, TidalPlaylist] +) -> Tuple[SpotifyID, TidalID | None]: + if spotify_playlist["name"] in tidal_playlists: + # if there's an existing tidal playlist with the name of the current playlist then use that + tdl_playlist_id = tidal_playlists[spotify_playlist["name"]].id + else: + tdl_playlist_id = None + return (spotify_playlist["id"], tdl_playlist_id) + + +def create_playlist_id_tuple( + spotify_session: SpotifySession, + tidal_session: TidalSession, + exclude: Optional[List[str]] = None, + uname: Optional[str] = None, +) -> List[Tuple[SpotifyID, TidalID | None]]: + spotify_playlists = get_spotify_playlists( + spotify_session, exclude=exclude, uname=uname + ) + tidal_playlists = get_tidal_playlists_dict(tidal_session) + + return [ + playlist_id_tuple(spotify_playlist, tidal_playlists) + for spotify_playlist in spotify_playlists + ] + + +def get_spotify_playlists( + spotify_session: SpotifySession, + exclude: Optional[Set[str]] = None, + uname: Optional[str] = None, +): + # get all the user playlists from the Spotify account + playlists = [] + if exclude is None: + exclude = [] + if uname is None: + spotify_results = spotify_session.current_user_playlists() + uname = spotify_session.current_user()["id"] + else: + spotify_results = spotify_session.user_playlists(uname) + print(exclude) + exclude_list = set(map(str.split(":")[-1], exclude)) + + def condition(playlist: Dict[str, Any]) -> bool: + return playlist["owner"]["id"] == uname and playlist["id"] not in exclude_list + + while True: + playlists.extend(filter(condition, spotify_results["items"])) + # move to the next page of results if there are still playlists remaining + if spotify_results["next"]: + spotify_results = spotify_session.next(spotify_results) + else: + break + return playlists + + +def get_playlists_from_config( + config: SyncConfig, +) -> List[Tuple[SpotifyID, TidalID]]: + # get the list of playlist sync mappings from the configuration file + return list( + map( + lambda x: (x["spotify_id"], x["tidal_id"]), config.get("sync_playlists", []) + ) + ) diff --git a/src/spotify_to_tidal/sync.py b/src/spotify_to_tidal/sync.py new file mode 100755 index 0000000..cf52bb2 --- /dev/null +++ b/src/spotify_to_tidal/sync.py @@ -0,0 +1,356 @@ +#!/usr/bin/env python3 + +import logging +import sys +import time +import traceback + +from asyncio import Future +from functools import partial +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any, List, Tuple, Callable, Optional, Set + +import requests +import spotipy +import tidalapi +import tidalapi.playlist + +from cachetools import cached, TTLCache +from tqdm import tqdm + +from .type import * +from .tidalapi_patch import set_tidal_playlist +from .filters import * + + +logger = logging.getLogger(__package__) + + +@cached(TTLCache(maxsize=1024, ttl=600)) +def _query_tidal(tidal_session: TidalSession, q: str, model): + return tidal_session.search(q, models=[model]) + + +@cached(TTLCache(maxsize=1024, ttl=600)) +def query_tidal_album(tidal_session: TidalSession, artist: str, album: str): + artist = simple(artist.casefold()) + album = simple(album.casefold()) + return _query_tidal(tidal_session, artist+" "+album, tidalapi.Album) + + +@cached(TTLCache(maxsize=1024, ttl=600)) +def query_tidal_track(tidal_session: TidalSession, artist: str, track: str): + artist = simple(artist.casefold()) + track = simple(track.casefold()) + return _query_tidal(tidal_session, artist+" "+track, tidalapi.Track) + + +def tidal_search( + spotify_track_and_cache: Tuple[SpotifyTrack, TidalTrack | None], + tidal_session: TidalSession, +) -> TidalTrack: + spotify_track, cached_tidal_track = spotify_track_and_cache + if cached_tidal_track: + logger.debug("Found %s in cache.", spotify_track["name"]) + logger.handlers[0].flush() + return cached_tidal_track + # search for album name and first album artist + if ( + "album" in spotify_track + and "artists" in spotify_track["album"] + and len(spotify_track["album"]["artists"]) + ): + for artist in spotify_track["album"]["artists"]: + artist_name = artist["name"] + album_name = spotify_track["album"]["name"] + album_result = query_tidal_album(tidal_session, artist_name, album_name) + + logger.debug( + "Looking for album %s in Tidal" % spotify_track["album"]["name"] + ) + logger.handlers[0].flush() + for album in album_result["albums"]: + album_tracks = album.tracks() + if len(album_tracks) >= spotify_track["track_number"]: + track = album_tracks[spotify_track["track_number"] - 1] + if match(track, spotify_track): + return track + + # if that fails then search for track name and first artist + logger.info( + "Did not find track %s in any artist albums, running general search." + % spotify_track["name"] + ) + logger.handlers[0].flush() + logger.debug("Searching spotify for %s", spotify_track["name"]) + logger.handlers[0].flush() + spotify_track_name = spotify_track["name"] + for artist in spotify_track["artists"]: + query_tidal_track(tidal_session, artist["name"], spotify_track_name) + artist_name = artist["name"] + search_res = query_tidal_track(tidal_session, artist=artist_name, track=spotify_track_name) + res: TidalTrack | None = next( + (x for x in search_res["tracks"] if match(x, spotify_track)), None + ) + if res: + logger.info("Found song %s in Tidal!", spotify_track["name"]) + logger.handlers[0].flush() + return res + logger.info("Could not find song %s" % spotify_track["name"]) + logger.handlers[0].flush() + return res + + +def repeat_on_request_error(function: Callable, *args, **kwargs): + # utility to repeat calling the function up to 5 times if an exception is thrown + logging.root.addFilter(Filter429('tidalapi')) + sleep_schedule = { + 5: 1, + 4: 10, + 3: 30, + 2: 60, + 1: 90, + } # sleep variable length of time depending on retry number + for rem_attempts in range(5, 0, -1): + try: + return function(*args, **kwargs) + except requests.exceptions.RequestException as e: + if e.response.status_code != 429 or rem_attempts < 1: + logger.critical("%s could not be recovered", e) + logger.debug("Exception info: %s", traceback.format_exception(e)) + raise e + logger.info( + "429 occured on %s, retrying %d more times", + e.request.path_url, + rem_attempts, + ) + if e.response is not None: + logger.debug("Response message: %s", e.response.text) + logger.debug("Response headers: %s", e.response.headers) + time.sleep(sleep_schedule.get(rem_attempts, 1)) + + logger.critical("Aborting sync") + logger.critical("The following arguments were provided\n\n: %s", str(args)) + logger.exception(traceback.format_exc()) + sys.exit(1) + + +def _enumerate_wrapper(value_tuple: Tuple, function: Callable, **kwargs) -> List: + # just a wrapper which accepts a tuple from enumerate and returns the index back as the first argument + index, value = value_tuple + logging.root.addFilter(Filter429('tidalapi')) + return (index, repeat_on_request_error(function, value, **kwargs)) + + +def call_async_with_progress(function, values, description, num_processes, **kwargs): + results = len(values) * [None] + cached = 0 + # Add the known results to the existing return var + cached_results = filter(lambda x: x[1][1] is not None, enumerate(values)) + for idx, res in cached_results: + results[idx] = res[1] + cached += 1 + # Only search for non-cached songs + to_search = filter(lambda x: x[1][1] is None, enumerate(values)) + with ThreadPoolExecutor(max_workers=num_processes) as process_pool: + try: + futures: List[Future] = [] + for s in to_search: + futures.append(process_pool.submit(_enumerate_wrapper, value_tuple=s, function=function, **kwargs)) + for index, result in tqdm( + map(lambda x: x.result(), as_completed(futures)), + total=len(futures), + desc=description, + unit="req", + ): + results[index] = result + except KeyboardInterrupt: + logger.critical("KeyboardInterrupt received. Killing pool.") + process_pool.shutdown(cancel_futures=True) + exit(1) + return results + + +def get_tracks_from_spotify_playlist( + spotify_session: SpotifySession, spotify_playlist +) -> List[SpotifyTrack]: + output = [] + results: List[SpotifyTrack] = spotify_session.playlist_tracks( + spotify_playlist["id"], + fields="next,items(track(name,album(name,artists),artists,track_number,duration_ms,id,external_ids(isrc)))", + ) + while True: + output.extend([r["track"] for r in results["items"] if r["track"] is not None]) + # move to the next page of results if there are still tracks remaining in the playlist + if results["next"]: + results = spotify_session.next(results) + else: + return output + + +class TidalPlaylistCache: + _existing: Any | None = None + _data: Set[TidalTrack] = set() + __slots__ = ('_existing', '_data') + + def __new__(cls, playlist: TidalPlaylist): + if cls._existing is None: + cls._existing = super().__new__(cls) + cls._data = set() + + cls._data.update(playlist.tracks()) + return cls._existing + + def _search(self, spotify_track: SpotifyTrack) -> TidalTrack | None: + """check if the given spotify track was already in the tidal playlist.""" + return next(filter(lambda x: match(x, spotify_track), self._data), None) + + def search( + self, spotify_session: spotipy.Spotify, spotify_playlist + ) -> Tuple[List[Tuple[SpotifyTrack, TidalTrack | None]], int]: + """Add the cached tidal track where applicable to a list of spotify tracks""" + results = [] + cache_hits = 0 + spotify_tracks = get_tracks_from_spotify_playlist( + spotify_session, spotify_playlist + ) + for track in spotify_tracks: + cached_track = self._search(track) + if cached_track is not None: + cache_hits += 1 + results.append((track, cached_track)) + return (results, cache_hits) + + +def tidal_playlist_is_dirty( + playlist: TidalPlaylist, new_track_ids: List[TidalID] +) -> bool: + old_tracks = playlist.tracks() + if len(old_tracks) != len(new_track_ids): + return True + for i in range(len(old_tracks)): + if old_tracks[i].id != new_track_ids[i]: + return True + return False + + +def sync_playlist( + spotify_session: SpotifySession, + tidal_session: TidalSession, + spotify_id: SpotifyID, + config, + tidal_id: Optional[TidalID] = None, +): + try: + spotify_playlist = spotify_session.playlist(spotify_id) + except spotipy.SpotifyException as e: + logger.error("Error getting Spotify playlist %s", spotify_id) + logger.exception(e) + return + if tidal_id: + # if a Tidal playlist was specified then look it up + try: + tidal_playlist = tidal_session.playlist(tidal_id) + except Exception as e: + logger.warning("Error getting Tidal playlist %s", tidal_id) + logger.debug(e) + return + else: + # create a new Tidal playlist if required + logger.warn( + "No playlist found on Tidal corresponding to Spotify playlist: %s. Creating new playlist", + spotify_playlist["name"], + ) + tidal_playlist: TidalPlaylist = tidal_session.user.create_playlist( + spotify_playlist["name"], spotify_playlist["description"] + ) + tidal_track_ids = [] + spotify_tracks, cache_hits = TidalPlaylistCache(tidal_playlist).search( + spotify_session, spotify_playlist + ) + if cache_hits == len(spotify_tracks): + logger.warn( + "No new tracks to search in Spotify playlist '%s'", spotify_playlist["name"] + ) + return + + task_description = ( + "Searching Tidal for {}/{} tracks in Spotify playlist '{}'".format( + len(spotify_tracks) - cache_hits, + len(spotify_tracks), + spotify_playlist["name"], + ) + ) + tidal_tracks = call_async_with_progress( + tidal_search, + spotify_tracks, + task_description, + config.get("subprocesses", 25), + tidal_session=tidal_session, + ) + + missing_tracks = 0 + for index, tidal_track in enumerate(tidal_tracks): + spotify_track = spotify_tracks[index][0] + if tidal_track: + tidal_track_ids.append(tidal_track.id) + else: + missing_tracks += 1 + color = ("\033[91m", "\033[0m") + logger.info( + color[0] + "Could not find track %s: %s - %s" + color[1], + spotify_track["id"], + ",".join(map(lambda x: x["name"], spotify_track["artists"])), + spotify_track["name"], + ) + logger.warn("Could not find %d tracks in Tidal", missing_tracks) + if tidal_playlist_is_dirty(tidal_playlist, tidal_track_ids): + set_tidal_playlist(tidal_playlist, tidal_track_ids) + print("Synced playlist.") + else: + print("No changes to write to Tidal playlist") + + +def update_tidal_playlist( + playlist: TidalPlaylist, track_ids: List[TidalID], *, chunk_size: int = 20 +) -> None: + offset = 0 + with tqdm( + desc="Erasing existing tracks from Tidal playlist", total=playlist.num_tracks + ) as progress: + while playlist.num_tracks: + indices = range(offset, min(playlist.num_tracks, chunk_size + offset)) + try: + print(playlist._etag) + playlist.remove_by_indices(indices) + offset += chunk_size + except: + logger.info("412 hit, sleeping for a bit.") + time.sleep(0.5) + continue + progress.update(len(indices)) + with tqdm( + desc="Adding new tracks to Tidal playlist", total=len(track_ids) + ) as progress: + offset = 0 + while offset < len(track_ids): + count = min(chunk_size, len(track_ids) - offset) + playlist.add(track_ids[offset : offset + chunk_size]) + offset += count + progress.update(count) + + +def sync_list( + spotify_session: spotipy.Spotify, + tidal_session: tidalapi.Session, + playlists: List[PlaylistConfig], + config: SyncConfig, +) -> List[TidalID]: + results = [] + for spotify_id, tidal_id in playlists: + # sync the spotify playlist to tidal + repeat_on_request_error( + sync_playlist, spotify_session, tidal_session, spotify_id, config, tidal_id + ) + results.append(tidal_id) + return results diff --git a/src/spotify_to_tidal/tidalapi_patch.py b/src/spotify_to_tidal/tidalapi_patch.py new file mode 100644 index 0000000..d972a3a --- /dev/null +++ b/src/spotify_to_tidal/tidalapi_patch.py @@ -0,0 +1,44 @@ +from tqdm import tqdm +from .type import TidalPlaylist, TidalID +from typing import List + + +def _remove_indices_from_playlist(playlist: TidalPlaylist, indices) -> None: + headers = {"If-None-Match": playlist._etag} + index_string = ",".join(map(str, indices)) + playlist.requests.request( + "DELETE", + (playlist._base_url + "/items/%s") % (playlist.id, index_string), + headers=headers, + ) + playlist._reparse() + + +def clear_tidal_playlist(playlist: TidalPlaylist, chunk_size=20) -> None: + with tqdm( + desc="Erasing existing tracks from Tidal playlist", total=playlist.num_tracks, unit='req' + ) as progress: + while playlist.num_tracks: + indices = range(min(playlist.num_tracks, chunk_size)) + _remove_indices_from_playlist(playlist, indices) + progress.update(len(indices)) + + +def add_multiple_tracks_to_playlist( + playlist: TidalPlaylist, track_ids: List[TidalID], chunk_size=20 +) -> None: + offset = 0 + with tqdm( + desc="Adding new tracks to Tidal playlist", total=len(track_ids), unit='req' + ) as progress: + while offset < len(track_ids): + count = min(chunk_size, len(track_ids) - offset) + playlist.add(track_ids[offset : offset + chunk_size]) + offset += count + progress.update(count) + + +def set_tidal_playlist(playlist: TidalPlaylist, track_ids: List[TidalID]) -> None: + if playlist.num_tracks != 0: + clear_tidal_playlist(playlist) + add_multiple_tracks_to_playlist(playlist, track_ids) diff --git a/src/spotify_to_tidal/type/__init__.py b/src/spotify_to_tidal/type/__init__.py new file mode 100644 index 0000000..b84fe04 --- /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/config.py b/src/spotify_to_tidal/type/config.py new file mode 100644 index 0000000..f7f3236 --- /dev/null +++ b/src/spotify_to_tidal/type/config.py @@ -0,0 +1,26 @@ +from typing import TypedDict, Literal, List, Optional + + +class SpotifyConfig(TypedDict): + client_id: str + client_secret: str + username: str + redirect_url: str + + +class TidalConfig(TypedDict): + access_token: str + refresh_token: str + session_id: str + token_type: Literal["Bearer"] + + +class PlaylistConfig(TypedDict): + spotify_id: str + tidal_id: str + + +class SyncConfig(TypedDict): + spotify: SpotifyConfig + sync_playlists: Optional[List[PlaylistConfig]] + excluded_playlists: Optional[List[str]] diff --git a/src/spotify_to_tidal/type/playlist.py b/src/spotify_to_tidal/type/playlist.py new file mode 100644 index 0000000..bc024af --- /dev/null +++ b/src/spotify_to_tidal/type/playlist.py @@ -0,0 +1,5 @@ +from tidalapi.playlist import Playlist, UserPlaylist +from typing import Union + + +TidalPlaylist = Union[Playlist, UserPlaylist] diff --git a/src/spotify_to_tidal/type/spotify.py b/src/spotify_to_tidal/type/spotify.py new file mode 100644 index 0000000..1a1c0e1 --- /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 diff --git a/sync.py b/sync.py deleted file mode 100755 index ea0790e..0000000 --- a/sync.py +++ /dev/null @@ -1,312 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -from auth import open_tidal_session, open_spotify_session -from functools import partial -from multiprocessing import Pool -import requests -import sys -import spotipy -import tidalapi -from tidalapi_patch import set_tidal_playlist -import time -from tqdm import tqdm -import traceback -import unicodedata -import yaml - -def normalize(s): - return unicodedata.normalize('NFD', s).encode('ascii', 'ignore').decode('ascii') - -def simple(input_string): - # 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): - 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): - # 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): - 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 - - # handle some edge cases - if exclusion_rule("instrumental", tidal_track, spotify_track): return False - if exclusion_rule("acapella", tidal_track, spotify_track): return False - if exclusion_rule("remix", tidal_track, spotify_track): return False - - # the simplified version of the Spotify track name must be a substring of the Tidal track name - # Try with both un-normalized and then normalized - 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): - if '&' in artist: - return artist.split('&') - elif ',' in artist: - return artist.split(',') - else: - return [artist] - - def get_tidal_artists(tidal_track, do_normalize=False): - result = [] - for artist in tidal_track.artists: - if do_normalize: - artist_name = normalize(artist.name) - else: - artist_name = artist.name - 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): - result = [] - for artist in spotify_track['artists']: - if do_normalize: - artist_name = normalize(artist['name']) - else: - artist_name = artist['name'] - result.extend(split_artist_name(artist_name)) - return set([simple(x.strip().lower()) for x in result]) - # There must be at least one overlapping artist between the Tidal and Spotify track - # Try with both un-normalized and then normalized - if get_tidal_artists(tidal_track).intersection(get_spotify_artists(spotify_track)) != set(): - return True - return get_tidal_artists(tidal_track, True).intersection(get_spotify_artists(spotify_track, True)) != set() - -def match(tidal_track, spotify_track): - return isrc_match(tidal_track, spotify_track) or ( - duration_match(tidal_track, spotify_track) - and name_match(tidal_track, spotify_track) - and artist_match(tidal_track, spotify_track) - ) - - -def tidal_search(spotify_track_and_cache, tidal_session): - 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 - if 'album' in spotify_track and 'artists' in spotify_track['album'] and len(spotify_track['album']['artists']): - album_result = tidal_session.search(simple(spotify_track['album']['name']) + " " + simple(spotify_track['album']['artists'][0]['name']), models=[tidalapi.album.Album]) - for album in album_result['albums']: - album_tracks = album.tracks() - if len(album_tracks) >= spotify_track['track_number']: - track = album_tracks[spotify_track['track_number'] - 1] - if match(track, spotify_track): - return track - # if that fails then search for track name and first artist - for track in tidal_session.search(simple(spotify_track['name']) + ' ' + simple(spotify_track['artists'][0]['name']), models=[tidalapi.media.Track])['tracks']: - if match(track, spotify_track): - return track - -def get_tidal_playlists_dict(tidal_session): - # a dictionary of name --> playlist - tidal_playlists = tidal_session.user.playlists() - output = {} - for playlist in tidal_playlists: - output[playlist.name] = playlist - return output - -def repeat_on_request_error(function, *args, remaining=5, **kwargs): - # utility to repeat calling the function up to 5 times if an exception is thrown - try: - return function(*args, **kwargs) - except requests.exceptions.RequestException as e: - if remaining: - print(f"{str(e)} occurred, retrying {remaining} times") - else: - print(f"{str(e)} could not be recovered") - - if not e.response is None: - print(f"Response message: {e.response.text}") - print(f"Response headers: {e.response.headers}") - - if not remaining: - print("Aborting sync") - print(f"The following arguments were provided:\n\n {str(args)}") - print(traceback.format_exc()) - sys.exit(1) - sleep_schedule = {5: 1, 4:10, 3:60, 2:5*60, 1:10*60} # sleep variable length of time depending on retry number - time.sleep(sleep_schedule.get(remaining, 1)) - return repeat_on_request_error(function, *args, remaining=remaining-1, **kwargs) - -def _enumerate_wrapper(value_tuple, function, **kwargs): - # just a wrapper which accepts a tuple from enumerate and returns the index back as the first argument - index, value = value_tuple - return (index, repeat_on_request_error(function, value, **kwargs)) - -def call_async_with_progress(function, values, description, num_processes, **kwargs): - results = len(values)*[None] - with Pool(processes=num_processes) as process_pool: - for index, result in tqdm(process_pool.imap_unordered(partial(_enumerate_wrapper, function=function, **kwargs), - enumerate(values)), total=len(values), desc=description): - results[index] = result - return results - -def get_tracks_from_spotify_playlist(spotify_session, spotify_playlist): - output = [] - results = spotify_session.playlist_tracks( - spotify_playlist["id"], - fields="next,items(track(name,album(name,artists),artists,track_number,duration_ms,id,external_ids(isrc)))", - ) - while True: - output.extend([r['track'] for r in results['items'] if r['track'] is not None]) - # move to the next page of results if there are still tracks remaining in the playlist - if results['next']: - results = spotify_session.next(results) - else: - return output - -class TidalPlaylistCache: - def __init__(self, playlist): - self._data = playlist.tracks() - - def _search(self, spotify_track): - ''' check if the given spotify track was already in the tidal playlist.''' - results = [] - for tidal_track in self._data: - if match(tidal_track, spotify_track): - return tidal_track - return None - - def search(self, spotify_session, spotify_playlist): - ''' Add the cached tidal track where applicable to a list of spotify tracks ''' - results = [] - cache_hits = 0 - work_to_do = False - spotify_tracks = get_tracks_from_spotify_playlist(spotify_session, spotify_playlist) - for track in spotify_tracks: - cached_track = self._search(track) - if cached_track: - results.append( (track, cached_track) ) - cache_hits += 1 - else: - results.append( (track, None) ) - return (results, cache_hits) - -def tidal_playlist_is_dirty(playlist, new_track_ids): - old_tracks = playlist.tracks() - if len(old_tracks) != len(new_track_ids): - return True - for i in range(len(old_tracks)): - if old_tracks[i].id != new_track_ids[i]: - return True - return False - -def sync_playlist(spotify_session, tidal_session, spotify_id, tidal_id, 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 - try: - tidal_playlist = tidal_session.playlist(tidal_id) - except Exception as e: - print("Error getting Tidal playlist " + tidal_id) - print(e) - return - else: - # create a new Tidal playlist if required - 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_track_ids = [] - spotify_tracks, cache_hits = TidalPlaylistCache(tidal_playlist).search(spotify_session, spotify_playlist) - if cache_hits == len(spotify_tracks): - print("No new tracks to search in Spotify playlist '{}'".format(spotify_playlist['name'])) - return - - task_description = "Searching Tidal for {}/{} tracks in Spotify playlist '{}'".format(len(spotify_tracks) - cache_hits, len(spotify_tracks), spotify_playlist['name']) - tidal_tracks = call_async_with_progress(tidal_search, spotify_tracks, task_description, config.get('subprocesses', 50), tidal_session=tidal_session) - for index, tidal_track in enumerate(tidal_tracks): - spotify_track = spotify_tracks[index][0] - if tidal_track: - tidal_track_ids.append(tidal_track.id) - else: - color = ('\033[91m', '\033[0m') - print(color[0] + "Could not find track {}: {} - {}".format(spotify_track['id'], ",".join([a['name'] for a in spotify_track['artists']]), spotify_track['name']) + color[1]) - - if tidal_playlist_is_dirty(tidal_playlist, tidal_track_ids): - set_tidal_playlist(tidal_playlist, tidal_track_ids) - else: - print("No changes to write to Tidal playlist") - -def sync_list(spotify_session, tidal_session, playlists, config): - results = [] - for spotify_id, tidal_id in playlists: - # sync the spotify playlist to tidal - repeat_on_request_error(sync_playlist, spotify_session, tidal_session, spotify_id, tidal_id, config) - results.append(tidal_id) - return results - -def pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists): - 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']] - return (spotify_playlist['id'], tidal_playlist.id) - else: - return (spotify_playlist['id'], None) - - -def get_user_playlist_mappings(spotify_session, tidal_session, config): - results = [] - spotify_playlists = get_playlists_from_spotify(spotify_session, config) - tidal_playlists = get_tidal_playlists_dict(tidal_session) - for spotify_playlist in spotify_playlists: - results.append( pick_tidal_playlist_for_spotify_playlist(spotify_playlist, tidal_playlists) ) - return results - -def get_playlists_from_spotify(spotify_session, config): - # get all the user playlists from the Spotify account - playlists = [] - spotify_results = spotify_session.user_playlists(config['spotify']['username']) - exclude_list = set([x.split(':')[-1] for x in config.get('excluded_playlists', [])]) - while True: - for spotify_playlist in spotify_results['items']: - if spotify_playlist['owner']['id'] == config['spotify']['username'] and not spotify_playlist['id'] in exclude_list: - playlists.append(spotify_playlist) - # move to the next page of results if there are still playlists remaining - if spotify_results['next']: - spotify_results = spotify_session.next(spotify_results) - else: - break - return playlists - -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) diff --git a/tidalapi_patch.py b/tidalapi_patch.py deleted file mode 100644 index 24712cd..0000000 --- a/tidalapi_patch.py +++ /dev/null @@ -1,27 +0,0 @@ -from tqdm import tqdm - -def _remove_indices_from_playlist(playlist, indices): - headers = {'If-None-Match': playlist._etag} - index_string = ",".join(map(str, indices)) - playlist.requests.request('DELETE', (playlist._base_url + '/items/%s') % (playlist.id, index_string), headers=headers) - playlist._reparse() - -def clear_tidal_playlist(playlist, chunk_size=20): - with tqdm(desc="Erasing existing tracks from Tidal playlist", total=playlist.num_tracks) as progress: - while playlist.num_tracks: - indices = range(min(playlist.num_tracks, chunk_size)) - _remove_indices_from_playlist(playlist, indices) - progress.update(len(indices)) - -def add_multiple_tracks_to_playlist(playlist, track_ids, chunk_size=20): - offset = 0 - with tqdm(desc="Adding new tracks to Tidal playlist", total=len(track_ids)) as progress: - while offset < len(track_ids): - count = min(chunk_size, len(track_ids) - offset) - playlist.add(track_ids[offset:offset+chunk_size]) - offset += count - progress.update(count) - -def set_tidal_playlist(playlist, track_ids): - clear_tidal_playlist(playlist) - add_multiple_tracks_to_playlist(playlist, track_ids)