glomatico/votify
1
0
Fork 0
mirror of https://github.com/glomatico/votify.git synced 2026-09-26 19:12:27 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

initial files

Browse Source
This commit is contained in:
Rafael Moraes
2024-09-20 15:10:43 -03:00
parent d3680d6116
commit 8df9cff1d6
15 changed files with 1614 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+7
View File
@@ -0,0 +1,7 @@
/*
__pycache__
!votify
!.gitignore
!pyproject.toml
!README.md
!requirements.txt
requirements.txt
+2
View File
@@ -0,0 +1,2 @@
yt-dlp
pillow
votify/__init__.py
+1
View File
@@ -0,0 +1 @@
__version__ = "1.0"
votify/__main__.py
+3
View File
@@ -0,0 +1,3 @@
from .cli import main
main()
votify/cli.py
+516
View File
@@ -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))")
votify/constants.py
+39
View File
@@ -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 {}"
votify/downloader.py
+484
View File
@@ -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)
votify/downloader_episode.py
+43
View File
@@ -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")
votify/downloader_song.py
+120
View File
@@ -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"
votify/enums.py
+12
View File
@@ -0,0 +1,12 @@
from enum import Enum
class Quality(Enum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class DownloadMode(Enum):
YTDLP = "ytdlp"
ARIA2C = "aria2c"
votify/models.py
+32
View File
@@ -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
votify/playplay_pb2.py
+42
View File
@@ -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)
votify/playplay_pb2.pyi
+46
View File
@@ -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: ...
votify/spotify_api.py
+251
View File
@@ -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'<script id="session" data-testid="session" type="application/json">(.+?)</script>',
home_page,
).group(1)
)
self.config_info = json.loads(
re.search(
r'<script id="config" data-testid="config" type="application/json">(.+?)</script>',
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()
votify/utils.py
+16
View File
@@ -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}"
)