Aurora-SpotiRip/spotify_downloader.py

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()