spotify.py

from concurrent.futures import ThreadPoolExecutor
import time
import requests
import os
import re
import logging

PLAYLIST_URL = "https://api.spotifydown.com/trackList/playlist/{}"
TRACK_URL = "https://api.spotifydown.com/download/{}"
DOWNLOAD_PATH = r"C:\Users\USERNAME\Desktop\tracklist-1"

HEADERS = {
    "accept": "*/*",
    "accept-language": "en-US,en;q=0.9",
    "cache-control": "no-cache",
    "origin": "https://spotifydown.com",
    "pragma": "no-cache",
    "priority": "u=1, i",
    "referer": "https://spotifydown.com/",
    "sec-ch-ua": '"Google Chrome";v="125", "Chromium";v="125", "Not.A/Brand";v="24"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": "Windows",
    "sec-fetch-dest": "empty",
    "sec-fetch-mode": "cors",
    "sec-fetch-site": "same-site",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
}


def request_until_response(url, params=None, headers=None):
    for i in range(1, 11):
        try:
            response = requests.get(url, params=params, headers=headers)
            response.raise_for_status()
        except:
            time.sleep(i)
        else:
            return response

    return None


def get_filename(response: requests.Response) -> str | None:
    # Get the filename from the Content-Disposition header if available
    filename = None
    if "Content-Disposition" in response.headers:
        cd: dict[str, str] = dict(
            map(
                lambda x: x.strip().split("=") if "=" in x else (x.strip(), ""),
                response.headers["Content-Disposition"].split(";"),
            )
        )
        filename = cd.get("filename", "").replace('"', "")

    # Sanitize filename for Windows
    filename = re.sub(r'[\\/*?:"<>|]', "_", filename)

    return filename


def download_file(download_url: str, i: int, track):
    logging.info("Downloading track {}: name '{}'".format(i + 1, track["title"]))

    # Make a request to the download URL to get the file content
    response = request_until_response(download_url, headers=HEADERS)
    if response is None:
        logging.warning(
            "Downloading track {} has failed: name '{}'".format(i + 1, track["title"])
        )
        return

    filename = get_filename(response)
    # If filename couldn't be extracted from headers, ignore the download
    if not filename:
        logging.warning("There is no filename: download_url '{}'".format(download_url))
        return

    # Save the file to the desired path
    file_path = os.path.join(DOWNLOAD_PATH, filename)
    with open(file_path, "wb") as f:
        f.write(response.content)


def get_download_url(i: int, track) -> str | None:
    logging.info(
        "Getting the download URL for track {}: name '{}'".format(i + 1, track["title"])
    )
    response = request_until_response(TRACK_URL.format(track["id"]), headers=HEADERS)
    if response is None:
        logging.warning(
            "Getting the download URL for track {} has failed: name '{}'".format(
                i + 1, track["title"]
            )
        )
        return None

    # Process the response data
    response_body: dict = response.json()
    if response_body.get("success", None) == False:
        logging.warning(
            "Getting the download URL for track {} has failed: name '{}', reason '{}'".format(
                i + 1, track["title"], response_body.get("message", None)
            )
        )
        return None

    return response_body["link"]


def download_track(i: int, track):
    download_url = get_download_url(i, track)
    if download_url is None:
        return
    download_file(download_url, i, track)


def get_next_track_batch(playlist_id: str, next_offset: int):
    logging.info(
        "Getting the next batch of track metadata: next_offset {}".format(next_offset)
    )
    response = request_until_response(
        PLAYLIST_URL.format(playlist_id),
        params={"offset": next_offset},
        headers=HEADERS,
    )
    if response is None:
        logging.warning(
            "Getting the next batch of tracks has failed: next_offset '{}'".format(
                next_offset
            )
        )
        return

    # Process the response data
    response_body: dict = response.json()
    if response_body.get("success", None) == False:
        logging.warning(
            "Getting the next batch of tracks has failed: next_offset '{}', reason '{}'".format(
                next_offset, response_body.get("message", None)
            )
        )
        return

    next_offset = (
        int(response_body["nextOffset"])
        if response_body["nextOffset"] is not None
        else None
    )
    return next_offset, response_body["trackList"]


def download_tracklist(playlist_id: str):
    next_offset = 0
    tracklist = []

    while next_offset is not None:
        next_offset, next_track_batch = get_next_track_batch(playlist_id, next_offset)
        tracklist.extend(next_track_batch)

    logging.info("Found {} tracks in the list".format(len(tracklist)))

    with ThreadPoolExecutor(max_workers=4) as executor:
        for i, track in enumerate(tracklist):
            executor.submit(download_track, i, track)

    logging.info("Finished downloading the tracklist!")


if __name__ == "__main__":
    playlist_id = "PLAYLIST_ID"
    download_tracklist(playlist_id)