diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..88e3d55
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,7 @@
+/*
+__pycache__
+!votify
+!.gitignore
+!pyproject.toml
+!README.md
+!requirements.txt
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..cab8bcf
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,2 @@
+yt-dlp
+pillow
\ No newline at end of file
diff --git a/votify/__init__.py b/votify/__init__.py
new file mode 100644
index 0000000..4802e90
--- /dev/null
+++ b/votify/__init__.py
@@ -0,0 +1 @@
+__version__ = "1.0"
diff --git a/votify/__main__.py b/votify/__main__.py
new file mode 100644
index 0000000..4e28416
--- /dev/null
+++ b/votify/__main__.py
@@ -0,0 +1,3 @@
+from .cli import main
+
+main()
diff --git a/votify/cli.py b/votify/cli.py
new file mode 100644
index 0000000..a12f535
--- /dev/null
+++ b/votify/cli.py
@@ -0,0 +1,516 @@
+from __future__ import annotations
+
+import inspect
+import json
+import logging
+import time
+from enum import Enum
+from pathlib import Path
+
+import click
+
+from . import __version__
+from .constants import EXCLUDED_CONFIG_FILE_PARAMS, X_NOT_FOUND_STRING
+from .downloader import Downloader
+from .downloader_episode import DownloaderEpisode
+from .downloader_song import DownloaderSong
+from .enums import DownloadMode, Quality
+from .spotify_api import SpotifyApi
+
+spotify_api_sig = inspect.signature(SpotifyApi.__init__)
+downloader_sig = inspect.signature(Downloader.__init__)
+downloader_song_sig = inspect.signature(DownloaderSong.__init__)
+
+
+def get_param_string(param: click.Parameter) -> str:
+ if isinstance(param.default, Enum):
+ return param.default.value
+ elif isinstance(param.default, Path):
+ return str(param.default)
+ else:
+ return param.default
+
+
+def write_default_config_file(ctx: click.Context) -> None:
+ ctx.params["config_path"].parent.mkdir(parents=True, exist_ok=True)
+ config_file = {
+ param.name: get_param_string(param)
+ for param in ctx.command.params
+ if param.name not in EXCLUDED_CONFIG_FILE_PARAMS
+ }
+ ctx.params["config_path"].write_text(json.dumps(config_file, indent=4))
+
+
+def load_config_file(
+ ctx: click.Context,
+ param: click.Parameter,
+ no_config_file: bool,
+) -> click.Context:
+ if no_config_file:
+ return ctx
+ if not ctx.params["config_path"].exists():
+ write_default_config_file(ctx)
+ config_file = dict(json.loads(ctx.params["config_path"].read_text()))
+ for param in ctx.command.params:
+ if (
+ config_file.get(param.name) is not None
+ and not ctx.get_parameter_source(param.name)
+ == click.core.ParameterSource.COMMANDLINE
+ ):
+ ctx.params[param.name] = param.type_cast_value(ctx, config_file[param.name])
+ return ctx
+
+
+@click.command()
+@click.help_option("-h", "--help")
+@click.version_option(__version__, "-v", "--version")
+# CLI specific options
+@click.argument(
+ "urls",
+ nargs=-1,
+ type=str,
+ required=True,
+)
+@click.option(
+ "--wait-interval",
+ "-w",
+ type=float,
+ default=10,
+ help="Wait interval between downloads in seconds.",
+)
+@click.option(
+ "--force-premium",
+ "-f",
+ is_flag=True,
+ help="Force to detect the account as premium.",
+)
+@click.option(
+ "--save-cover",
+ "-s",
+ is_flag=True,
+ help="Save cover as a separate file.",
+)
+@click.option(
+ "--overwrite",
+ is_flag=True,
+ help="Overwrite existing files.",
+)
+@click.option(
+ "--read-urls-as-txt",
+ "-r",
+ is_flag=True,
+ help="Interpret URLs as paths to text files containing URLs.",
+)
+@click.option(
+ "--save-playlist",
+ is_flag=True,
+ help="Save a M3U8 playlist file when downloading a playlist.",
+)
+@click.option(
+ "--lrc-only",
+ "-l",
+ is_flag=True,
+ help="Download only the synced lyrics.",
+)
+@click.option(
+ "--no-lrc",
+ is_flag=True,
+ help="Don't download the synced lyrics.",
+)
+@click.option(
+ "--config-path",
+ type=Path,
+ default=Path.home() / ".spotify-web-downloader" / "config.json",
+ help="Path to config file.",
+)
+@click.option(
+ "--log-level",
+ type=str,
+ default="INFO",
+ help="Log level.",
+)
+@click.option(
+ "--print-exceptions",
+ is_flag=True,
+ help="Print exceptions.",
+)
+# SpotifyApi specific options
+@click.option(
+ "--cookies-path",
+ type=Path,
+ default=spotify_api_sig.parameters["cookies_path"].default,
+ help="Path to cookies file.",
+)
+# Downloader specific options
+@click.option(
+ "--quality",
+ "-q",
+ type=Quality,
+ default=downloader_sig.parameters["quality"].default,
+ help="Audio quality.",
+)
+@click.option(
+ "--output-path",
+ "-o",
+ type=Path,
+ default=downloader_sig.parameters["output_path"].default,
+ help="Path to output directory.",
+)
+@click.option(
+ "--temp-path",
+ type=Path,
+ default=downloader_sig.parameters["temp_path"].default,
+ help="Path to temporary directory.",
+)
+@click.option(
+ "--download-mode",
+ "-d",
+ type=DownloadMode,
+ default=downloader_sig.parameters["download_mode"].default,
+ help="Download mode.",
+)
+@click.option(
+ "--aria2c-path",
+ type=str,
+ default=downloader_sig.parameters["aria2c_path"].default,
+ help="Path to aria2c binary.",
+)
+@click.option(
+ "--playplay-path",
+ type=str,
+ default=downloader_sig.parameters["playplay_path"].default,
+ help="Path to playplay binary.",
+)
+@click.option(
+ "--template-folder-album",
+ type=str,
+ default=downloader_sig.parameters["template_folder_album"].default,
+ help="Template folder for tracks that are part of an album.",
+)
+@click.option(
+ "--template-folder-compilation",
+ type=str,
+ default=downloader_sig.parameters["template_folder_compilation"].default,
+ help="Template folder for tracks that are part of a compilation album.",
+)
+@click.option(
+ "--template-file-single-disc",
+ type=str,
+ default=downloader_sig.parameters["template_file_single_disc"].default,
+ help="Template file for the tracks that are part of a single-disc album.",
+)
+@click.option(
+ "--template-file-multi-disc",
+ type=str,
+ default=downloader_sig.parameters["template_file_multi_disc"].default,
+ help="Template file for the tracks that are part of a multi-disc album.",
+)
+@click.option(
+ "--template-folder-episode",
+ type=str,
+ default=downloader_sig.parameters["template_folder_episode"].default,
+ help="Template folder for episodes (podcasts).",
+)
+@click.option(
+ "--template-file-episode",
+ type=str,
+ default=downloader_sig.parameters["template_file_episode"].default,
+ help="Template file for episodes (podcasts).",
+)
+@click.option(
+ "--template-file-playlist",
+ type=str,
+ default=downloader_sig.parameters["template_file_playlist"].default,
+ help="Template file for the M3U8 playlist.",
+)
+@click.option(
+ "--date-tag-template",
+ type=str,
+ default=downloader_sig.parameters["date_tag_template"].default,
+ help="Date tag template.",
+)
+@click.option(
+ "--exclude-tags",
+ type=str,
+ default=downloader_sig.parameters["exclude_tags"].default,
+ help="Comma-separated tags to exclude.",
+)
+@click.option(
+ "--truncate",
+ type=int,
+ default=downloader_sig.parameters["truncate"].default,
+ help="Maximum length of the file/folder names.",
+)
+# This option should always be last
+@click.option(
+ "--no-config-file",
+ "-n",
+ is_flag=True,
+ callback=load_config_file,
+ help="Do not use a config file.",
+)
+def main(
+ urls: list[str],
+ wait_interval: float,
+ force_premium: bool,
+ save_cover: bool,
+ overwrite: bool,
+ read_urls_as_txt: bool,
+ save_playlist: bool,
+ lrc_only: bool,
+ no_lrc: bool,
+ config_path: Path,
+ log_level: str,
+ print_exceptions: bool,
+ cookies_path: Path,
+ quality: Quality,
+ output_path: Path,
+ temp_path: Path,
+ download_mode: DownloadMode,
+ aria2c_path: str,
+ playplay_path: str,
+ template_folder_album: str,
+ template_folder_compilation: str,
+ template_file_single_disc: str,
+ template_file_multi_disc: str,
+ template_folder_episode: str,
+ template_file_episode: str,
+ template_file_playlist: str,
+ date_tag_template: str,
+ exclude_tags: str,
+ truncate: int,
+ no_config_file: bool,
+) -> None:
+ logging.basicConfig(
+ format="[%(levelname)-8s %(asctime)s] %(message)s",
+ datefmt="%H:%M:%S",
+ )
+ logger = logging.getLogger(__name__)
+ logger.setLevel(log_level)
+ logger.debug("Starting downloader")
+ spotify_api = SpotifyApi(cookies_path)
+ downloader = Downloader(
+ spotify_api,
+ quality,
+ output_path,
+ temp_path,
+ download_mode,
+ aria2c_path,
+ playplay_path,
+ template_folder_album,
+ template_folder_compilation,
+ template_file_single_disc,
+ template_file_multi_disc,
+ template_folder_episode,
+ template_file_episode,
+ template_file_playlist,
+ date_tag_template,
+ exclude_tags,
+ truncate,
+ )
+ downloader_song = DownloaderSong(
+ downloader,
+ )
+ downloader_episode = DownloaderEpisode(
+ downloader,
+ )
+ if not lrc_only:
+ if not downloader.playplay_path_full:
+ logger.critical(X_NOT_FOUND_STRING.format("playplay", playplay_path))
+ return
+ if download_mode == DownloadMode.ARIA2C and not downloader.aria2c_path_full:
+ logger.critical(X_NOT_FOUND_STRING.format("aria2c", aria2c_path))
+ return
+ spotify_api.config_info["isPremium"] = (
+ True if force_premium else spotify_api.config_info["isPremium"]
+ )
+ if not spotify_api.config_info["isPremium"] and quality == Quality.HIGH:
+ logger.critical("Cannot download at chosen quality with a free account")
+ return
+ error_count = 0
+ if read_urls_as_txt:
+ _urls = []
+ for url in urls:
+ if Path(url).exists():
+ _urls.extend(Path(url).read_text(encoding="utf-8").splitlines())
+ urls = _urls
+ for url_index, url in enumerate(urls, start=1):
+ url_progress = f"URL {url_index}/{len(urls)}"
+ logger.info(f'({url_progress}) Checking "{url}"')
+ try:
+ url_info = downloader.get_url_info(url)
+ download_queue = downloader.get_download_queue(url_info)
+ except Exception as e:
+ error_count += 1
+ logger.error(
+ f'({url_progress}) Failed to check "{url}"',
+ exc_info=print_exceptions,
+ )
+ continue
+ medias_metadata = download_queue.medias_metadata
+ playlist_metadata = download_queue.playlist_metadata
+ for index, track_metadata in enumerate(medias_metadata, start=1):
+ queue_progress = (
+ f"Track {index}/{len(medias_metadata)} from URL {url_index}/{len(urls)}"
+ )
+ try:
+ logger.info(
+ f'({queue_progress}) Downloading "{track_metadata["name"]}"'
+ )
+ decrypted_path = None
+ media_id = track_metadata["id"]
+ media_type = track_metadata["type"]
+ logger.debug("Getting stream info")
+ stream_info = downloader.get_stream_info(
+ **{f"{media_type}_id": media_id},
+ )
+ if stream_info.quality != quality:
+ logger.warning(
+ f"({queue_progress}) Quality has been changed to {stream_info.quality.value}"
+ )
+ if not stream_info.file_id:
+ logger.warning(
+ f"({queue_progress}) Media is not available on Spotify's "
+ "servers and no alternative found, skipping"
+ )
+ continue
+ logger.debug("Getting decryption key")
+ decryption_key = downloader.get_decryption_key(stream_info.file_id)
+ if media_type == "track":
+ logger.debug("Getting lyrics")
+ lyrics = downloader_song.get_lyrics(media_id)
+ if not download_queue.album_metadata:
+ logger.debug("Getting album metadata")
+ album_metadata = spotify_api.get_album(
+ track_metadata["album"]["id"]
+ )
+ else:
+ album_metadata = download_queue.album_metadata
+ logger.debug("Getting track credits")
+ track_credits = spotify_api.get_track_credits(media_id)
+ tags = downloader_song.get_tags(
+ track_metadata,
+ album_metadata,
+ track_credits,
+ lyrics.unsynced,
+ )
+ if playlist_metadata:
+ tags = {
+ **tags,
+ **downloader.get_playlist_tags(
+ playlist_metadata,
+ index,
+ ),
+ }
+ final_path = downloader.get_final_path(
+ media_type,
+ tags,
+ ".ogg",
+ )
+ lrc_path = downloader.get_lrc_path(final_path)
+ cover_path = downloader_song.get_cover_path(final_path)
+ cover_url = downloader_song.get_cover_url(album_metadata)
+ if lrc_only:
+ pass
+ elif final_path.exists() and not overwrite:
+ logger.warning(
+ f'({queue_progress}) Track already exists at "{final_path}", skipping'
+ )
+ else:
+ encrypted_path = downloader.get_encrypted_path(media_id)
+ decrypted_path = downloader.get_decrypted_path(media_id)
+ logger.debug(f'Downloading to "{encrypted_path}"')
+ downloader.download(encrypted_path, stream_info.stream_url)
+ logger.debug(f'Decrypting to "{decrypted_path}"')
+ downloader.decrypt(
+ decryption_key,
+ encrypted_path,
+ decrypted_path,
+ )
+ if no_lrc or not lyrics.synced:
+ pass
+ elif lrc_path.exists() and not overwrite:
+ logger.debug(
+ f'Synced lyrics already exists at "{lrc_path}", skipping'
+ )
+ else:
+ logger.debug(f'Saving synced lyrics to "{lrc_path}"')
+ downloader.save_lrc(lrc_path, lyrics.synced)
+ elif media_type == "episode" and not lrc_only:
+ if not download_queue.show_metadata:
+ logger.debug("Getting show metadata")
+ show_metadata = spotify_api.get_show(
+ track_metadata["show"]["id"]
+ )
+ else:
+ show_metadata = download_queue.show_metadata
+ tags = downloader_episode.get_tags(
+ track_metadata,
+ show_metadata,
+ )
+ if playlist_metadata:
+ tags = {
+ **tags,
+ **downloader.get_playlist_tags(
+ playlist_metadata,
+ index,
+ ),
+ }
+ final_path = downloader.get_final_path(
+ media_type,
+ tags,
+ ".ogg",
+ )
+ cover_path = downloader_episode.get_cover_path(final_path)
+ cover_url = downloader_song.get_cover_url(track_metadata)
+ if final_path.exists() and not overwrite:
+ logger.warning(
+ f'({queue_progress}) Track already exists at "{final_path}", skipping'
+ )
+ else:
+ encrypted_path = downloader.get_encrypted_path(media_id)
+ decrypted_path = downloader.get_decrypted_path(media_id)
+ logger.debug(f'Downloading to "{encrypted_path}"')
+ downloader.download(encrypted_path, stream_info.stream_url)
+ logger.debug(f'Decrypting to "{decrypted_path}"')
+ downloader.decrypt(
+ decryption_key,
+ encrypted_path,
+ decrypted_path,
+ )
+ if lrc_only or not save_cover:
+ pass
+ elif cover_path.exists() and not overwrite:
+ logger.debug(f'Cover already exists at "{cover_path}", skipping')
+ elif cover_url is not None:
+ logger.debug(f'Saving cover to "{cover_path}"')
+ downloader.save_cover(cover_path, cover_url)
+ if decrypted_path:
+ logger.debug("Applying tags")
+ downloader.apply_tags(decrypted_path, tags, cover_url)
+ logger.debug(f'Moving to "{final_path}"')
+ downloader.move_to_final_path(decrypted_path, final_path)
+ if not lrc_only and save_playlist and playlist_metadata:
+ playlist_file_path = downloader.get_playlist_file_path(tags)
+ logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"')
+ downloader.update_playlist_file(
+ playlist_file_path,
+ final_path,
+ index,
+ )
+ except Exception as e:
+ error_count += 1
+ logger.error(
+ f'({queue_progress}) Failed to download "{track_metadata["name"]}"',
+ exc_info=print_exceptions,
+ )
+ finally:
+ if temp_path.exists():
+ logger.debug(f'Cleaning up "{temp_path}"')
+ downloader.cleanup_temp_path()
+ if wait_interval > 0 and index != len(medias_metadata):
+ logger.debug(
+ f"Waiting for {wait_interval} second(s) before continuing"
+ )
+ time.sleep(wait_interval)
+ logger.info(f"Done ({error_count} error(s))")
diff --git a/votify/constants.py b/votify/constants.py
new file mode 100644
index 0000000..f2bd302
--- /dev/null
+++ b/votify/constants.py
@@ -0,0 +1,39 @@
+from .enums import Quality
+
+EXCLUDED_CONFIG_FILE_PARAMS = (
+ "urls",
+ "config_path",
+ "read_urls_as_txt",
+ "no_config_file",
+ "version",
+ "help",
+)
+
+VORBIS_TAGS_MAPPING = {
+ "album": "ALBUM",
+ "album_artist": "ALBUMARTIST",
+ "artist": "ARTIST",
+ "composer": "COMPOSER",
+ "copyright": "COPYRIGHT",
+ "description": "DESCRIPTION",
+ "disc": "DISC",
+ "disc_total": "DISCTOTAL",
+ "isrc": "ISRC",
+ "label": "LABEL",
+ "lyrics": "LYRICS",
+ "publisher": "PUBLISHER",
+ "producer": "PRODUCER",
+ "release_date": "YEAR",
+ "title": "TITLE",
+ "track": "TRACKNUMBER",
+ "track_total": "TRACKTOTAL",
+ "url": "URL",
+}
+
+QUALITY_X_FORMAT_ID_MAPPING = {
+ Quality.HIGH: "OGG_VORBIS_320",
+ Quality.MEDIUM: "OGG_VORBIS_160",
+ Quality.LOW: "OGG_VORBIS_96",
+}
+
+X_NOT_FOUND_STRING = "{} not found at {}"
diff --git a/votify/downloader.py b/votify/downloader.py
new file mode 100644
index 0000000..d281d28
--- /dev/null
+++ b/votify/downloader.py
@@ -0,0 +1,484 @@
+from __future__ import annotations
+
+import base64
+import datetime
+import functools
+import re
+import shutil
+import subprocess
+from io import BytesIO
+from pathlib import Path
+
+import requests
+from Crypto.Cipher import AES
+from Crypto.Util import Counter
+from mutagen.flac import Picture
+from mutagen.oggvorbis import OggVorbis, OggVorbisHeaderError
+from PIL import Image
+from .playplay_pb2 import (
+ AUDIO_TRACK,
+ Interactivity,
+ PlayPlayLicenseRequest,
+ PlayPlayLicenseResponse,
+)
+from yt_dlp import YoutubeDL
+
+from .constants import QUALITY_X_FORMAT_ID_MAPPING, VORBIS_TAGS_MAPPING
+from .enums import DownloadMode, Quality
+from .models import DownloadQueue, StreamInfo, UrlInfo
+from .spotify_api import SpotifyApi
+from .utils import check_response
+
+
+class Downloader:
+ ILLEGAL_CHARACTERS_REGEX = r'[\\/:*?"<>|;]'
+ URL_RE = r"(album|playlist|track|show|episode)/(\w{22})"
+ ILLEGAL_CHARACTERS_REPLACEMENT = "_"
+ RELEASE_DATE_PRECISION_MAPPING = {
+ "year": "%Y",
+ "month": "%Y-%m",
+ "day": "%Y-%m-%d",
+ }
+
+ def __init__(
+ self,
+ spotify_api: SpotifyApi,
+ quality: Quality = Quality.MEDIUM,
+ output_path: Path = Path("./Spotify"),
+ temp_path: Path = Path("./temp"),
+ download_mode: DownloadMode = DownloadMode.YTDLP,
+ aria2c_path: Path = "aria2c",
+ playplay_path: Path = "playplay",
+ template_folder_album: str = "{album_artist}/{album}",
+ template_folder_compilation: str = "Compilations/{album}",
+ template_file_single_disc: str = "{track:02d} {title}",
+ template_file_multi_disc: str = "{disc}-{track:02d} {title}",
+ template_folder_episode: str = "Podcasts/{album}",
+ template_file_episode: str = "{track:02d} {title}",
+ template_file_playlist: str = "Playlists/{playlist_artist}/{playlist_title}",
+ date_tag_template: str = "%Y-%m-%dT%H:%M:%SZ",
+ exclude_tags: str = None,
+ truncate: int = None,
+ silence: bool = False,
+ ):
+ self.spotify_api = spotify_api
+ self.quality = quality
+ self.output_path = output_path
+ self.temp_path = temp_path
+ self.download_mode = download_mode
+ self.aria2c_path = aria2c_path
+ self.playplay_path = playplay_path
+ self.template_folder_album = template_folder_album
+ self.template_folder_compilation = template_folder_compilation
+ self.template_file_single_disc = template_file_single_disc
+ self.template_file_multi_disc = template_file_multi_disc
+ self.template_folder_episode = template_folder_episode
+ self.template_file_episode = template_file_episode
+ self.template_file_playlist = template_file_playlist
+ self.date_tag_template = date_tag_template
+ self.exclude_tags = exclude_tags
+ self.truncate = truncate
+ self.silence = silence
+ self._set_binaries_full_path()
+ self._set_exclude_tags_list()
+ self._set_truncate()
+ self._set_subprocess_additional_args()
+
+ def _set_binaries_full_path(self):
+ self.aria2c_path_full = shutil.which(self.aria2c_path)
+ self.playplay_path_full = shutil.which(self.playplay_path)
+
+ def _set_exclude_tags_list(self):
+ self.exclude_tags_list = (
+ [i.lower() for i in self.exclude_tags.split(",")]
+ if self.exclude_tags is not None
+ else []
+ )
+
+ def _set_truncate(self):
+ if self.truncate is not None:
+ self.truncate = None if self.truncate < 4 else self.truncate
+
+ def _set_subprocess_additional_args(self):
+ if self.silence:
+ self.subprocess_additional_args = {
+ "stdout": subprocess.DEVNULL,
+ "stderr": subprocess.DEVNULL,
+ }
+ else:
+ self.subprocess_additional_args = {}
+
+ def get_url_info(self, url: str) -> UrlInfo:
+ url_regex_result = re.search(self.URL_RE, url)
+ if url_regex_result is None:
+ raise Exception("Invalid URL")
+ return UrlInfo(type=url_regex_result.group(1), id=url_regex_result.group(2))
+
+ def get_download_queue(
+ self,
+ url_info: UrlInfo,
+ ) -> DownloadQueue:
+ download_queue = DownloadQueue(medias_metadata=[])
+ if url_info.type == "album":
+ album = self.spotify_api.get_album(url_info.id)
+ download_queue.medias_metadata.extend(
+ track for track in album["tracks"]["items"] if track is not None
+ )
+ download_queue.album_metadata = album
+ elif url_info.type == "playlist":
+ playlist = self.spotify_api.get_playlist(url_info.id)
+ download_queue.playlist_metadata = playlist.copy()
+ download_queue.playlist_metadata.pop("tracks")
+ download_queue.medias_metadata.extend(
+ track_metadata["track"]
+ for track_metadata in playlist["tracks"]["items"]
+ if track_metadata["track"] is not None
+ )
+ elif url_info.type == "track":
+ download_queue.medias_metadata.append(
+ self.spotify_api.get_track(url_info.id)
+ )
+ elif url_info.type == "episode":
+ download_queue.medias_metadata.append(
+ self.spotify_api.get_episode(url_info.id)
+ )
+ elif url_info.type == "show":
+ show = self.spotify_api.get_show(url_info.id)
+ download_queue.show_metadata = show.copy()
+ download_queue.medias_metadata.extend(
+ episode for episode in show["episodes"]["items"]
+ )
+ return download_queue
+
+ def get_playlist_tags(self, playlist_metadata: dict, playlist_track: int) -> dict:
+ return {
+ "playlist_artist": playlist_metadata["owner"]["display_name"],
+ "playlist_title": playlist_metadata["name"],
+ "playlist_track": playlist_track,
+ }
+
+ def get_playlist_file_path(
+ self,
+ tags: dict,
+ ):
+ template_file = self.template_file_playlist.split("/")
+ return Path(
+ self.output_path,
+ *[
+ self.get_sanitized_string(i.format(**tags), True)
+ for i in template_file[0:-1]
+ ],
+ *[
+ self.get_sanitized_string(template_file[-1].format(**tags), False)
+ + ".m3u8"
+ ],
+ )
+
+ def get_lrc_path(self, final_path: Path) -> Path:
+ return final_path.with_suffix(".lrc")
+
+ def save_lrc(self, lrc_path: Path, lyrics_synced: str):
+ if lyrics_synced:
+ lrc_path.parent.mkdir(parents=True, exist_ok=True)
+ lrc_path.write_text(lyrics_synced, encoding="utf8")
+
+ def get_final_path(self, media_type: str, tags: dict, file_extension: str) -> Path:
+ if media_type == "track":
+ template_folder = (
+ self.template_folder_compilation.split("/")
+ if tags.get("compilation")
+ else self.template_folder_album.split("/")
+ )
+ template_file = (
+ self.template_file_multi_disc.split("/")
+ if tags["disc_total"] > 1
+ else self.template_file_single_disc.split("/")
+ )
+ elif media_type == "episode":
+ template_folder = self.template_folder_episode.split("/")
+ template_file = self.template_file_episode.split("/")
+ template_final = template_folder + template_file
+ return Path(
+ self.output_path,
+ *[
+ self.get_sanitized_string(i.format(**tags), True)
+ for i in template_final[0:-1]
+ ],
+ (
+ self.get_sanitized_string(template_final[-1].format(**tags), False)
+ + file_extension
+ ),
+ )
+
+ def update_playlist_file(
+ self,
+ playlist_file_path: Path,
+ final_path: Path,
+ playlist_track: int,
+ ):
+ playlist_file_path.parent.mkdir(parents=True, exist_ok=True)
+ playlist_file_path_parent_parts_len = len(playlist_file_path.parent.parts)
+ output_path_parts_len = len(self.output_path.parts)
+ final_path_relative = Path(
+ ("../" * (playlist_file_path_parent_parts_len - output_path_parts_len)),
+ *final_path.parts[output_path_parts_len:],
+ )
+ playlist_file_lines = (
+ playlist_file_path.open("r", encoding="utf8").readlines()
+ if playlist_file_path.exists()
+ else []
+ )
+ if len(playlist_file_lines) < playlist_track:
+ playlist_file_lines.extend(
+ "\n" for _ in range(playlist_track - len(playlist_file_lines))
+ )
+ playlist_file_lines[playlist_track - 1] = final_path_relative.as_posix() + "\n"
+ with playlist_file_path.open("w", encoding="utf8") as playlist_file:
+ playlist_file.writelines(playlist_file_lines)
+
+ def get_audio_file(
+ self,
+ audio_files: list[dict],
+ ) -> tuple[Quality, dict] | tuple[None, None]:
+ qualities = list(Quality)
+ start_index = qualities.index(self.quality)
+ for quality in qualities[start_index:]:
+ for audio_file in audio_files:
+ if audio_file["format"] == QUALITY_X_FORMAT_ID_MAPPING[quality]:
+ return quality, audio_file
+ return None, None
+
+ def get_stream_info(
+ self,
+ track_id: str = None,
+ episode_id: str = None,
+ ) -> StreamInfo:
+ if not track_id and not episode_id:
+ raise RuntimeError()
+ stream_info = StreamInfo()
+ gid = self.spotify_api.media_id_to_gid(track_id or episode_id)
+ if track_id:
+ gid_metadata = self.spotify_api.get_gid_metadata(gid, "track")
+ audio_files = gid_metadata.get("file")
+ elif episode_id:
+ gid_metadata = self.spotify_api.get_gid_metadata(gid, "episode")
+ audio_files = gid_metadata.get("audio")
+ audio_files = audio_files or gid_metadata.get("alternative")
+ if not audio_files:
+ return stream_info
+ quality, audio_file = self.get_audio_file(audio_files)
+ if not audio_file:
+ return stream_info
+ file_id = audio_file["file_id"]
+ stream_url = self.spotify_api.get_stream_urls(file_id)["cdnurl"][0]
+ stream_info.stream_url = stream_url
+ stream_info.file_id = file_id
+ stream_info.quality = quality
+ return stream_info
+
+ def get_decryption_key(self, file_id: str) -> bytes:
+ playplay_license_request = PlayPlayLicenseRequest(
+ version=2,
+ token=bytes.fromhex("01e132cae527bd21620e822f58514932"),
+ interactivity=Interactivity.INTERACTIVE,
+ content_type=AUDIO_TRACK,
+ )
+ playplay_license_response_bytes = self.spotify_api.get_playplay_license(
+ file_id,
+ playplay_license_request.SerializeToString(),
+ )
+ playplay_license_response = PlayPlayLicenseResponse()
+ playplay_license_response.ParseFromString(playplay_license_response_bytes)
+ obfuscated = playplay_license_response.obfuscated_key.hex()
+ output = subprocess.check_output(
+ [
+ self.playplay_path_full,
+ file_id,
+ obfuscated,
+ ],
+ shell=False,
+ )
+ key = bytes.fromhex(output.strip().decode("utf-8"))
+ assert key
+ return key
+
+ def download(self, input_path: Path, stream_url: str):
+ if self.download_mode == DownloadMode.YTDLP:
+ self.download_ytdlp(input_path, stream_url)
+ elif self.download_mode == DownloadMode.ARIA2C:
+ self.download_aria2c(input_path, stream_url)
+
+ def download_ytdlp(self, input_path: Path, stream_url: str) -> None:
+ with YoutubeDL(
+ {
+ "quiet": True,
+ "no_warnings": True,
+ "outtmpl": str(input_path),
+ "allow_unplayable_formats": True,
+ "fixup": "never",
+ "allowed_extractors": ["generic"],
+ "noprogress": self.silence,
+ }
+ ) as ydl:
+ ydl.download(stream_url)
+
+ def download_aria2c(self, input_path: Path, stream_url: str) -> None:
+ input_path.parent.mkdir(parents=True, exist_ok=True)
+ subprocess.run(
+ [
+ self.aria2c_path_full,
+ "--no-conf",
+ "--download-result=hide",
+ "--console-log-level=error",
+ "--summary-interval=0",
+ "--file-allocation=none",
+ stream_url,
+ "--out",
+ input_path,
+ ],
+ check=True,
+ **self.subprocess_additional_args,
+ )
+ print("\r", end="")
+
+ def decrypt(
+ self,
+ decryption_key: bytes,
+ encrypted_path: Path,
+ decrypted_path: Path,
+ iv_diff: int = 0x100,
+ ):
+ with encrypted_path.open("rb") as encrypted_file:
+ encrypted_buffer = encrypted_file.read()
+ iv_int = int.from_bytes(
+ b"r\xe0g\xfb\xdd\xcb\xcfw\xeb\xe8\xbcd?c\r\x93", "big"
+ )
+ chunk_size = 4096
+ decrypted_buffer = BytesIO()
+ chunk_count = len(encrypted_buffer) // chunk_size
+
+ for chunk_index in range(chunk_count + 1):
+ chunk = encrypted_buffer[
+ chunk_index * chunk_size : (chunk_index + 1) * chunk_size
+ ]
+ iv = iv_int + (chunk_index * (chunk_size // 16))
+
+ for i in range(0, len(chunk), chunk_size):
+ cipher = AES.new(
+ key=decryption_key,
+ mode=AES.MODE_CTR,
+ counter=Counter.new(128, initial_value=iv),
+ )
+ block = chunk[i : i + chunk_size]
+ decrypted_chunk = cipher.decrypt(block)
+ decrypted_buffer.write(decrypted_chunk)
+
+ iv += iv_diff
+
+ decrypted_buffer.seek(0)
+ decrypted_content = decrypted_buffer.read()
+
+ assert decrypted_content.startswith(b"OggS"), "Decryption failed"
+
+ with decrypted_path.open("wb") as decrypted_file:
+ decrypted_file.write(decrypted_content[167:])
+
+ def get_sanitized_string(self, dirty_string: str, is_folder: bool) -> str:
+ dirty_string = re.sub(
+ self.ILLEGAL_CHARACTERS_REGEX,
+ self.ILLEGAL_CHARACTERS_REPLACEMENT,
+ dirty_string,
+ )
+ if is_folder:
+ dirty_string = dirty_string[: self.truncate]
+ if dirty_string.endswith("."):
+ dirty_string = dirty_string[:-1] + self.ILLEGAL_CHARACTERS_REPLACEMENT
+ else:
+ if self.truncate is not None:
+ dirty_string = dirty_string[: self.truncate - 4]
+ return dirty_string.strip()
+
+ def get_release_date_datetime_obj(
+ self,
+ release_date: str,
+ release_date_precision: str,
+ ) -> datetime.datetime:
+ return datetime.datetime.strptime(
+ release_date,
+ self.RELEASE_DATE_PRECISION_MAPPING[release_date_precision],
+ )
+
+ def get_release_date_tag(self, datetime_obj: datetime.datetime) -> str:
+ return datetime_obj.strftime(self.date_tag_template)
+
+ def get_artist_string(self, artist_list: list[dict]) -> str:
+ if len(artist_list) == 1:
+ return artist_list[0]["name"]
+ return (
+ ", ".join(i["name"] for i in artist_list[:-1])
+ + f' & {artist_list[-1]["name"]}'
+ )
+
+ def get_cover_url(self, images_dict: list[dict]) -> str:
+ return max(images_dict, key=lambda img: img["height"])["url"]
+
+ def get_encrypted_path(
+ self,
+ track_id: str,
+ ) -> Path:
+ return self.temp_path / (f"{track_id}_encrypted.ogg")
+
+ def get_decrypted_path(
+ self,
+ track_id: str,
+ ) -> Path:
+ return self.temp_path / (f"{track_id}_decrypted.ogg")
+
+ def apply_tags(
+ self,
+ input_path: Path,
+ tags: dict,
+ cover_url: str,
+ ) -> None:
+ file = OggVorbis(input_path)
+ file.clear()
+ ogg_tags = {
+ v: str(tags[k])
+ for k, v in VORBIS_TAGS_MAPPING.items()
+ if k not in self.exclude_tags_list and tags.get(k) is not None
+ }
+ if "cover" not in self.exclude_tags_list and cover_url:
+ cover_bytes = self.get_response_bytes(cover_url)
+ picture = Picture()
+ picture.mime = "image/jpeg"
+ picture.data = cover_bytes
+ picture.type = 3
+ picture.width, picture.height = Image.open(BytesIO(cover_bytes)).size
+ ogg_tags["METADATA_BLOCK_PICTURE"] = base64.b64encode(
+ picture.write()
+ ).decode("ascii")
+ file.update(ogg_tags)
+ try:
+ file.save()
+ except OggVorbisHeaderError:
+ pass
+
+ @staticmethod
+ @functools.lru_cache()
+ def get_response_bytes(url: str) -> bytes:
+ response = requests.get(url)
+ check_response(response)
+ return response.content
+
+ def move_to_final_path(self, input_path: Path, final_path: Path):
+ final_path.parent.mkdir(parents=True, exist_ok=True)
+ shutil.move(input_path, final_path)
+
+ @functools.lru_cache()
+ def save_cover(self, cover_path: Path, cover_url: str):
+ if cover_url is not None:
+ cover_path.parent.mkdir(parents=True, exist_ok=True)
+ cover_path.write_bytes(self.get_response_bytes(cover_url))
+
+ def cleanup_temp_path(self):
+ shutil.rmtree(self.temp_path)
diff --git a/votify/downloader_episode.py b/votify/downloader_episode.py
new file mode 100644
index 0000000..f998db0
--- /dev/null
+++ b/votify/downloader_episode.py
@@ -0,0 +1,43 @@
+from pathlib import Path
+
+from .downloader import Downloader
+
+
+class DownloaderEpisode:
+ def __init__(
+ self,
+ downloader: Downloader,
+ ):
+ self.downloader = downloader
+
+ def get_tags(
+ self,
+ episode_metadata: dict,
+ show_metadata: dict,
+ ) -> dict:
+ release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
+ episode_metadata["release_date"],
+ episode_metadata["release_date_precision"],
+ )
+ tags = {
+ "album": show_metadata["name"],
+ "description": episode_metadata["description"],
+ "publisher": show_metadata.get("publisher"),
+ "rating": "Explicit" if episode_metadata.get("explicit") else "Unknown",
+ "release_date": self.downloader.get_release_date_tag(
+ release_date_datetime_obj
+ ),
+ "release_year": str(release_date_datetime_obj.year),
+ "title": episode_metadata["name"],
+ "track": next(
+ index
+ for index in range(1, len(show_metadata["episodes"]["items"]) + 1)
+ if show_metadata["episodes"]["items"][index - 1]["id"]
+ == episode_metadata["id"]
+ ),
+ "url": f"https://open.spotify.com/episode/{episode_metadata['id']}",
+ }
+ return tags
+
+ def get_cover_path(self, final_path: Path) -> Path:
+ return final_path.with_suffix(".jpg")
diff --git a/votify/downloader_song.py b/votify/downloader_song.py
new file mode 100644
index 0000000..d4096df
--- /dev/null
+++ b/votify/downloader_song.py
@@ -0,0 +1,120 @@
+import datetime
+from pathlib import Path
+
+from .downloader import Downloader
+from .models import Lyrics, StreamInfo
+
+
+class DownloaderSong:
+ def __init__(
+ self,
+ downloader: Downloader,
+ ):
+ self.downloader = downloader
+
+ def get_cover_url(self, album_metadata: dict) -> str | None:
+ if not album_metadata.get("images"):
+ return None
+ return self.downloader.get_cover_url(album_metadata["images"])
+
+ def get_tags(
+ self,
+ track_metadata: dict,
+ album_metadata: dict,
+ track_credits: dict,
+ lyrics_unsynced: str,
+ ) -> dict:
+ external_ids = track_metadata.get("external_ids")
+ release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
+ album_metadata["release_date"],
+ album_metadata["release_date_precision"],
+ )
+ producers = next(
+ role
+ for role in track_credits["roleCredits"]
+ if role["roleTitle"] == "Producers"
+ )["artists"]
+ composers = next(
+ role
+ for role in track_credits["roleCredits"]
+ if role["roleTitle"] == "Writers"
+ )["artists"]
+ disc = next(
+ (
+ i["disc_number"]
+ for i in album_metadata["tracks"]["items"]
+ if i["id"] == track_metadata["id"]
+ ),
+ )
+ tags = {
+ "album": album_metadata["name"],
+ "album_artist": self.downloader.get_artist_string(
+ album_metadata["artists"]
+ ),
+ "artist": self.downloader.get_artist_string(track_metadata["artists"]),
+ "compilation": (
+ True if album_metadata["album_type"] == "compilation" else False
+ ),
+ "composer": (
+ self.downloader.get_artist_string(composers) if composers else None
+ ),
+ "copyright": next(
+ (i["text"] for i in album_metadata["copyrights"] if i["type"] == "P"),
+ None,
+ ),
+ "disc": int(disc),
+ "disc_total": int(album_metadata["tracks"]["items"][-1]["disc_number"]),
+ "isrc": external_ids.get("isrc") if external_ids is not None else None,
+ "label": album_metadata.get("label"),
+ "lyrics": lyrics_unsynced,
+ "producer": (
+ self.downloader.get_artist_string(producers) if producers else None
+ ),
+ "rating": "Explicit" if track_metadata.get("explicit") else "Unknown",
+ "release_date": self.downloader.get_release_date_tag(
+ release_date_datetime_obj
+ ),
+ "release_year": str(release_date_datetime_obj.year),
+ "title": track_metadata["name"],
+ "track": int(
+ next(
+ (
+ i["track_number"]
+ for i in album_metadata["tracks"]["items"]
+ if i["id"] == track_metadata["id"]
+ ),
+ )
+ ),
+ "track_total": int(
+ max(
+ i["track_number"]
+ for i in album_metadata["tracks"]["items"]
+ if i["disc_number"] == disc
+ )
+ ),
+ "url": f"https://open.spotify.com/track/{track_metadata['id']}",
+ }
+ return tags
+
+ def get_lyrics_synced_timestamp_lrc(self, time: int) -> str:
+ lrc_timestamp = datetime.datetime.fromtimestamp(
+ time / 1000.0, tz=datetime.timezone.utc
+ )
+ return lrc_timestamp.strftime("%M:%S.%f")[:-4]
+
+ def get_lyrics(self, track_id: str) -> Lyrics:
+ lyrics = Lyrics()
+ raw_lyrics = self.downloader.spotify_api.get_lyrics(track_id)
+ if raw_lyrics is None:
+ return lyrics
+ lyrics.synced = ""
+ lyrics.unsynced = ""
+ for line in raw_lyrics["lyrics"]["lines"]:
+ if raw_lyrics["lyrics"]["syncType"] == "LINE_SYNCED":
+ lyrics.synced += f'[{self.get_lyrics_synced_timestamp_lrc(int(line["startTimeMs"]))}]{line["words"]}\n'
+ lyrics.unsynced += f'{line["words"]}\n'
+ lyrics.unsynced = lyrics.unsynced[:-1]
+ return lyrics
+
+ def get_cover_path(self, final_path: Path) -> Path:
+ return final_path.parent / "Cover.jpg"
diff --git a/votify/enums.py b/votify/enums.py
new file mode 100644
index 0000000..058a5b9
--- /dev/null
+++ b/votify/enums.py
@@ -0,0 +1,12 @@
+from enum import Enum
+
+
+class Quality(Enum):
+ HIGH = "high"
+ MEDIUM = "medium"
+ LOW = "low"
+
+
+class DownloadMode(Enum):
+ YTDLP = "ytdlp"
+ ARIA2C = "aria2c"
diff --git a/votify/models.py b/votify/models.py
new file mode 100644
index 0000000..a5c87e3
--- /dev/null
+++ b/votify/models.py
@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .enums import Quality
+
+
+@dataclass
+class Lyrics:
+ synced: str = None
+ unsynced: str = None
+
+
+@dataclass
+class UrlInfo:
+ type: str = None
+ id: str = None
+
+
+@dataclass
+class DownloadQueue:
+ playlist_metadata: dict = None
+ album_metadata: dict = None
+ show_metadata: dict = None
+ medias_metadata: list[dict] = None
+
+
+@dataclass
+class StreamInfo:
+ stream_url: str = None
+ file_id: str = None
+ quality: Quality = None
diff --git a/votify/playplay_pb2.py b/votify/playplay_pb2.py
new file mode 100644
index 0000000..d36e9de
--- /dev/null
+++ b/votify/playplay_pb2.py
@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler. DO NOT EDIT!
+# NO CHECKED-IN PROTOBUF GENCODE
+# source: playplay.proto
+# Protobuf Python Version: 5.27.2
+"""Generated protocol buffer code."""
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import descriptor_pool as _descriptor_pool
+from google.protobuf import runtime_version as _runtime_version
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf.internal import builder as _builder
+_runtime_version.ValidateProtobufRuntimeVersion(
+ _runtime_version.Domain.PUBLIC,
+ 5,
+ 27,
+ 2,
+ '',
+ 'playplay.proto'
+)
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0eplayplay.proto\x12\x16spotify.playplay.proto\"\xc3\x01\n\x16PlayPlayLicenseRequest\x12\x0f\n\x07version\x18\x01 \x01(\x05\x12\r\n\x05token\x18\x02 \x01(\x0c\x12\x10\n\x08\x63\x61\x63he_id\x18\x03 \x01(\x0c\x12<\n\rinteractivity\x18\x04 \x01(\x0e\x32%.spotify.playplay.proto.Interactivity\x12\x39\n\x0c\x63ontent_type\x18\x05 \x01(\x0e\x32#.spotify.playplay.proto.ContentType\"1\n\x17PlayPlayLicenseResponse\x12\x16\n\x0eobfuscated_key\x18\x01 \x01(\x0c*I\n\rInteractivity\x12\x19\n\x15UNKNOWN_INTERACTIVITY\x10\x00\x12\x0f\n\x0bINTERACTIVE\x10\x01\x12\x0c\n\x08\x44OWNLOAD\x10\x02*Z\n\x0b\x43ontentType\x12\x18\n\x14UNKNOWN_CONTENT_TYPE\x10\x00\x12\x0f\n\x0b\x41UDIO_TRACK\x10\x01\x12\x11\n\rAUDIO_EPISODE\x10\x02\x12\r\n\tAUDIO_ADD\x10\x03')
+
+_globals = globals()
+_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'playplay_pb2', _globals)
+if not _descriptor._USE_C_DESCRIPTORS:
+ DESCRIPTOR._loaded_options = None
+ _globals['_INTERACTIVITY']._serialized_start=291
+ _globals['_INTERACTIVITY']._serialized_end=364
+ _globals['_CONTENTTYPE']._serialized_start=366
+ _globals['_CONTENTTYPE']._serialized_end=456
+ _globals['_PLAYPLAYLICENSEREQUEST']._serialized_start=43
+ _globals['_PLAYPLAYLICENSEREQUEST']._serialized_end=238
+ _globals['_PLAYPLAYLICENSERESPONSE']._serialized_start=240
+ _globals['_PLAYPLAYLICENSERESPONSE']._serialized_end=289
+# @@protoc_insertion_point(module_scope)
diff --git a/votify/playplay_pb2.pyi b/votify/playplay_pb2.pyi
new file mode 100644
index 0000000..beffacf
--- /dev/null
+++ b/votify/playplay_pb2.pyi
@@ -0,0 +1,46 @@
+from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class Interactivity(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+ __slots__ = ()
+ UNKNOWN_INTERACTIVITY: _ClassVar[Interactivity]
+ INTERACTIVE: _ClassVar[Interactivity]
+ DOWNLOAD: _ClassVar[Interactivity]
+
+class ContentType(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+ __slots__ = ()
+ UNKNOWN_CONTENT_TYPE: _ClassVar[ContentType]
+ AUDIO_TRACK: _ClassVar[ContentType]
+ AUDIO_EPISODE: _ClassVar[ContentType]
+ AUDIO_ADD: _ClassVar[ContentType]
+UNKNOWN_INTERACTIVITY: Interactivity
+INTERACTIVE: Interactivity
+DOWNLOAD: Interactivity
+UNKNOWN_CONTENT_TYPE: ContentType
+AUDIO_TRACK: ContentType
+AUDIO_EPISODE: ContentType
+AUDIO_ADD: ContentType
+
+class PlayPlayLicenseRequest(_message.Message):
+ __slots__ = ("version", "token", "cache_id", "interactivity", "content_type")
+ VERSION_FIELD_NUMBER: _ClassVar[int]
+ TOKEN_FIELD_NUMBER: _ClassVar[int]
+ CACHE_ID_FIELD_NUMBER: _ClassVar[int]
+ INTERACTIVITY_FIELD_NUMBER: _ClassVar[int]
+ CONTENT_TYPE_FIELD_NUMBER: _ClassVar[int]
+ version: int
+ token: bytes
+ cache_id: bytes
+ interactivity: Interactivity
+ content_type: ContentType
+ def __init__(self, version: _Optional[int] = ..., token: _Optional[bytes] = ..., cache_id: _Optional[bytes] = ..., interactivity: _Optional[_Union[Interactivity, str]] = ..., content_type: _Optional[_Union[ContentType, str]] = ...) -> None: ...
+
+class PlayPlayLicenseResponse(_message.Message):
+ __slots__ = ("obfuscated_key",)
+ OBFUSCATED_KEY_FIELD_NUMBER: _ClassVar[int]
+ obfuscated_key: bytes
+ def __init__(self, obfuscated_key: _Optional[bytes] = ...) -> None: ...
diff --git a/votify/spotify_api.py b/votify/spotify_api.py
new file mode 100644
index 0000000..83a07e5
--- /dev/null
+++ b/votify/spotify_api.py
@@ -0,0 +1,251 @@
+from __future__ import annotations
+
+import functools
+import json
+import re
+import time
+import typing
+from http.cookiejar import MozillaCookieJar
+from pathlib import Path
+
+import base62
+import requests
+
+from .utils import check_response
+
+
+class SpotifyApi:
+ SPOTIFY_HOME_PAGE_URL = "https://open.spotify.com/"
+ CLIENT_VERSION = "1.2.46.25.g7f189073"
+ LYRICS_API_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/{track_id}"
+ METADATA_API_URL = "https://api.spotify.com/v1/{type}/{track_id}"
+ GID_METADATA_API_URL = "https://spclient.wg.spotify.com/metadata/4/{media_type}/{gid}?market=from_token"
+ PLAYPLAY_LICENSE_API_URL = (
+ "https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}"
+ )
+ TRACK_CREDITS_API_URL = "https://spclient.wg.spotify.com/track-credits-view/v0/experimental/{track_id}/credits"
+ STREAM_URLS_API_URL = (
+ "https://gue1-spclient.spotify.com/storage-resolve/v2/files/audio/interactive/11/"
+ "{file_id}?version=10000000&product=9&platform=39&alt=json"
+ )
+ EXTEND_TRACK_COLLECTION_WAIT_TIME = 0.5
+
+ def __init__(
+ self,
+ cookies_path: Path | None = Path("./cookies.txt"),
+ ):
+ self.cookies_path = cookies_path
+ self._set_session()
+
+ def _set_session(self):
+ self.session = requests.Session()
+ if self.cookies_path:
+ cookies = MozillaCookieJar(self.cookies_path)
+ cookies.load(ignore_discard=True, ignore_expires=True)
+ self.session.cookies.update(cookies)
+ self.session.headers.update(
+ {
+ "accept": "application/json",
+ "accept-language": "en-US",
+ "content-type": "application/json",
+ "origin": self.SPOTIFY_HOME_PAGE_URL,
+ "priority": "u=1, i",
+ "referer": self.SPOTIFY_HOME_PAGE_URL,
+ "sec-ch-ua": '"Not)A;Brand";v="99", "Google Chrome";v="127", "Chromium";v="127"',
+ "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/127.0.0.0 Safari/537.36",
+ "spotify-app-version": self.CLIENT_VERSION,
+ "app-platform": "WebPlayer",
+ }
+ )
+ self._set_session_auth()
+
+ def _set_session_auth(self):
+ home_page = self.get_home_page()
+ self.session_info = json.loads(
+ re.search(
+ r'',
+ home_page,
+ ).group(1)
+ )
+ self.config_info = json.loads(
+ re.search(
+ r'',
+ home_page,
+ ).group(1)
+ )
+ self.session.headers.update(
+ {
+ "Authorization": f"Bearer {self.session_info['accessToken']}",
+ }
+ )
+
+ def _refresh_session_auth(self):
+ timestamp_session_expire = int(
+ self.session_info["accessTokenExpirationTimestampMs"]
+ )
+ timestamp_now = time.time() * 1000
+ if timestamp_now < timestamp_session_expire:
+ return
+ self._set_session_auth()
+
+ def get_home_page(self) -> str:
+ response = self.session.get(
+ SpotifyApi.SPOTIFY_HOME_PAGE_URL,
+ )
+ check_response(response)
+ return response.text
+
+ @staticmethod
+ def media_id_to_gid(media_id: str) -> str:
+ return hex(base62.decode(media_id, base62.CHARSET_INVERTED))[2:].zfill(32)
+
+ @staticmethod
+ def gid_to_media_id(gid: str) -> str:
+ return base62.encode(int(gid, 16), charset=base62.CHARSET_INVERTED).zfill(22)
+
+ def get_gid_metadata(
+ self,
+ gid: str,
+ media_type: str,
+ ) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.GID_METADATA_API_URL.format(gid=gid, media_type=media_type)
+ )
+ check_response(response)
+ return response.json()
+
+ def get_lyrics(self, track_id: str) -> dict | None:
+ self._refresh_session_auth()
+ response = self.session.get(self.LYRICS_API_URL.format(track_id=track_id))
+ if response.status_code == 404:
+ return None
+ check_response(response)
+ return response.json()
+
+ def get_track(self, track_id: str) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.METADATA_API_URL.format(type="tracks", track_id=track_id)
+ )
+ check_response(response)
+ return response.json()
+
+ def extend_track_collection(
+ self,
+ track_collection: dict,
+ media_type: str,
+ ) -> typing.Generator[dict, None, None]:
+ next_url = track_collection[media_type]["next"]
+ while next_url is not None:
+ response = self.session.get(next_url)
+ check_response(response)
+ extended_collection = response.json()
+ yield extended_collection
+ next_url = extended_collection["next"]
+ time.sleep(self.EXTEND_TRACK_COLLECTION_WAIT_TIME)
+
+ @functools.lru_cache()
+ def get_album(
+ self,
+ album_id: str,
+ extend: bool = True,
+ ) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.METADATA_API_URL.format(type="albums", track_id=album_id)
+ )
+ check_response(response)
+ album = response.json()
+ if extend:
+ album["tracks"]["items"].extend(
+ [
+ item
+ for extended_collection in self.extend_track_collection(
+ album,
+ "tracks",
+ )
+ for item in extended_collection["items"]
+ ]
+ )
+ return album
+
+ def get_playlist(
+ self,
+ playlist_id: str,
+ extend: bool = True,
+ ) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.METADATA_API_URL.format(type="playlists", track_id=playlist_id)
+ )
+ check_response(response)
+ playlist = response.json()
+ if extend:
+ playlist["tracks"]["items"].extend(
+ [
+ item
+ for extended_collection in self.extend_track_collection(
+ playlist,
+ "tracks",
+ )
+ for item in extended_collection["items"]
+ ]
+ )
+ return playlist
+
+ def get_track_credits(self, track_id: str) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.TRACK_CREDITS_API_URL.format(track_id=track_id)
+ )
+ check_response(response)
+ return response.json()
+
+ def get_episode(self, episode_id: str) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.METADATA_API_URL.format(type="episodes", track_id=episode_id)
+ )
+ check_response(response)
+ return response.json()
+
+ def get_show(self, show_id: str, extend: bool = True) -> dict:
+ self._refresh_session_auth()
+ response = self.session.get(
+ self.METADATA_API_URL.format(type="shows", track_id=show_id)
+ )
+ check_response(response)
+ show = response.json()
+ if extend:
+ show["episodes"]["items"].extend(
+ [
+ item
+ for extended_collection in self.extend_track_collection(
+ show,
+ "episodes",
+ )
+ for item in extended_collection["items"]
+ ]
+ )
+ return show
+
+ def get_playplay_license(self, file_id: str, challenge: bytes) -> bytes:
+ self._refresh_session_auth()
+ response = self.session.post(
+ self.PLAYPLAY_LICENSE_API_URL.format(file_id=file_id),
+ challenge,
+ )
+ check_response(response)
+ return response.content
+
+ def get_stream_urls(self, file_id: str) -> str:
+ self._refresh_session_auth()
+ response = self.session.get(self.STREAM_URLS_API_URL.format(file_id=file_id))
+ check_response(response)
+ return response.json()
diff --git a/votify/utils.py b/votify/utils.py
new file mode 100644
index 0000000..1dca001
--- /dev/null
+++ b/votify/utils.py
@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import requests
+
+
+def check_response(response: requests.Response):
+ try:
+ response.raise_for_status()
+ except requests.HTTPError:
+ _raise_response_exception(response)
+
+
+def _raise_response_exception(response: requests.Response):
+ raise Exception(
+ f"Request failed with status code {response.status_code}: {response.text}"
+ )