import argparse import json import os import re import shlex import shutil import sys import zipfile from pathlib import Path from spotdl import Spotdl from spotdl.download.downloader import AUDIO_PROVIDERS from spotdl.utils.config import DEFAULT_CONFIG FILENAME_TEMPLATE = "{artists} - {title}.{output-ext}" BATCH_SIZE = 4 FALLBACK_PROVIDERS = ["youtube", "bandcamp"] EVENT_PREFIX = "@@SPOTIRIP " _client = None _json_events = False def yt_dlp_args(): runtime = os.environ.get("SPOTIRIP_JS_RUNTIME") if runtime: return "--js-runtimes " + shlex.quote(f"node:{runtime}") for name in ("deno", "node"): path = shutil.which(name) if path: return "--js-runtimes " + shlex.quote(f"{name}:{path}") return None def get_client(output_dir="downloads"): global _client if _client is None: client_id = os.environ.get("SPOTIFY_CLIENT_ID") client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET") _client = Spotdl( client_id=client_id or DEFAULT_CONFIG["client_id"], client_secret=client_secret or DEFAULT_CONFIG["client_secret"], use_official_api=bool(client_id and client_secret), downloader_settings={ "output": str(Path(output_dir) / FILENAME_TEMPLATE), "yt_dlp_args": yt_dlp_args(), **({"ffmpeg": os.environ["SPOTIRIP_FFMPEG"]} if os.environ.get("SPOTIRIP_FFMPEG") else {}), }, ) return _client def emit(event, **data): if _json_events: print(EVENT_PREFIX + json.dumps({"event": event, **data}), flush=True) def safe_filename(name): cleaned = re.sub(r'[\\/:*?"<>|\x00-\x1f]+', " ", name).strip(" .") return re.sub(r"\s+", " ", cleaned)[:150] or "Spotify download" def _retry_with_fallback(client, songs): downloader = client.downloader settings = downloader.settings primary = downloader.audio_providers recovered = [] try: for name in FALLBACK_PROVIDERS: if not songs: break downloader.audio_providers = [ AUDIO_PROVIDERS[name]( output_format=settings["format"], cookie_file=settings["cookie_file"], search_query=settings["search_query"], filter_results=settings["filter_results"], yt_dlp_args=settings["yt_dlp_args"], ) ] for song in songs: song.download_url = None print(f"Retrying {song.name} by {'/'.join(song.artists)} with {name}") results = client.download_songs(songs) recovered += [(song, path) for song, path in results if path] songs = [song for song, path in results if not path] finally: downloader.audio_providers = primary return recovered def _download(urls, output_dir="downloads"): client = get_client(output_dir) songs = client.search(urls if isinstance(urls, list) else [urls]) list_name = songs[0].list_name if songs and songs[0].list_name else None emit("found", total=len(songs), listName=list_name) results = [] for start in range(0, len(songs), BATCH_SIZE): batch = client.download_songs(songs[start : start + BATCH_SIZE]) failed = [song for song, path in batch if not path] if failed: retried = {id(song): path for song, path in _retry_with_fallback(client, failed)} batch = [(song, path or retried.get(id(song))) for song, path in batch] for song, path in batch: results.append((song, path)) if path: print(f"Downloaded {song.name} by {'/'.join(song.artists)} -> {path}") else: print(f"Failed to download {song.name} by {'/'.join(song.artists)}") emit( "progress", done=len(results), total=len(songs), title=song.name, artist="/".join(song.artists), ok=bool(path), ) return results, list_name def zip_results(results, zip_path): with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_STORED) as archive: for _, path in results: if path and Path(path).exists(): archive.write(path, arcname=Path(path).name) return zip_path def download_spotify_song(song_url, output_dir="downloads"): try: results, _ = _download(song_url, output_dir) return results except Exception as e: print(f"Error downloading song: {e}") def download_spotify_playlist(playlist_url, output_dir="downloads", zip_name=None): try: results, list_name = _download(playlist_url, output_dir) if zip_name is not None: name = safe_filename(zip_name or list_name or "Spotify playlist") zip_results(results, Path(output_dir) / f"{name}.zip") return results except Exception as e: print(f"Error downloading playlist: {e}") def main(): global _json_events parser = argparse.ArgumentParser(description="Download Spotify tracks, albums and playlists with spotdl.") parser.add_argument("urls", nargs="*", help="Spotify track, album or playlist URL(s)") parser.add_argument("--urls-stdin", action="store_true", help="Read additional URLs from standard input, one per line") parser.add_argument("--output-dir", default="downloads", help="Folder to save MP3s into") parser.add_argument("--zip", action="store_true", help="Also bundle the MP3s into a zip named after the playlist") parser.add_argument("--zip-name", default="", help="Name for the zip (defaults to the playlist/album name)") parser.add_argument("--json", action="store_true", help="Print machine-readable progress events") args = parser.parse_args() _json_events = args.json if args.urls_stdin: args.urls += [line.strip() for line in sys.stdin if line.strip()] if not args.urls: parser.error("no URLs given") Path(args.output_dir).mkdir(parents=True, exist_ok=True) if not args.json: for url in args.urls: if "/playlist/" in url or "/album/" in url: download_spotify_playlist(url, args.output_dir, args.zip_name if args.zip else None) else: download_spotify_song(url, args.output_dir) return try: results, list_name = _download(args.urls, args.output_dir) saved = [path for _, path in results if path] if not saved: emit("error", message="None of the tracks could be downloaded.") sys.exit(1) if args.zip: name = safe_filename(args.zip_name or list_name or "Spotify playlist") file = zip_results(results, Path(args.output_dir) / f"{name}.zip") else: file = saved[0] emit("complete", file=str(Path(file).resolve()), downloaded=len(saved), failed=len(results) - len(saved)) except Exception as e: emit("error", message=str(e)) sys.exit(1) if __name__ == "__main__": main()