Remove legacy files

This commit is contained in:
Rafael Moraes
2026-02-26 13:00:01 -03:00
parent 58d155ce93
commit e8baeb107f
15 changed files with 0 additions and 3881 deletions
-3
View File
@@ -1,3 +0,0 @@
from .cli import main
main()
-799
View File
@@ -1,799 +0,0 @@
from __future__ import annotations
import traceback
import inspect
import logging
import time
import typing
from pathlib import Path
import click
import colorama
from . import __version__
from .config_file import ConfigFile
from .constants import (
AAC_AUDIO_QUALITIES,
EXCLUDED_CONFIG_FILE_PARAMS,
PREMIUM_AUDIO_QUALITIES,
VORBIS_AUDIO_QUALITIES,
X_NOT_FOUND_STRING,
)
from .custom_logger_formatter import CustomLoggerFormatter
from .downloader import Downloader
from .downloader_audio import DownloaderAudio
from .downloader_episode import DownloaderEpisode
from .downloader_episode_video import DownloaderEpisodeVideo
from .downloader_music_video import DownloaderMusicVideo
from .downloader_song import DownloaderSong
from .downloader_video import DownloaderVideo
from .enums import (
AudioQuality,
CoverSize,
DownloadMode,
RemuxModeAudio,
RemuxModeVideo,
VideoFormat,
)
from .spotify_api import SpotifyApi
from .utils import color_text, prompt_path
logger = logging.getLogger("votify")
spotify_api_sig = inspect.signature(SpotifyApi.__init__)
downloader_sig = inspect.signature(Downloader.__init__)
downloader_audio_sig = inspect.signature(DownloaderAudio.__init__)
downloader_song_sig = inspect.signature(DownloaderSong.__init__)
downloader_video_sig = inspect.signature(DownloaderVideo.__init__)
class Csv(click.ParamType):
name = "csv"
def __init__(
self,
subtype: typing.Any,
) -> None:
self.subtype = subtype
def convert(
self,
value: str | typing.Any,
param: click.Parameter,
ctx: click.Context,
) -> list[typing.Any]:
if not isinstance(value, str):
return value
items = [v.strip() for v in value.split(",") if v.strip()]
result = []
for item in items:
try:
result.append(self.subtype(item))
except ValueError as e:
self.fail(
f"'{item}' is not a valid value for {self.subtype.__name__}",
param,
ctx,
)
return result
def load_config_file(
ctx: click.Context,
param: click.Parameter,
no_config_file: bool,
) -> click.Context:
if no_config_file:
return ctx
filtered_params = [
param
for param in ctx.command.params
if param.name not in EXCLUDED_CONFIG_FILE_PARAMS
]
config_file = ConfigFile(ctx.params["config_path"])
config_file.add_params_default_to_config(
filtered_params,
)
parsed_params = config_file.parse_params_from_config(
[
param
for param in filtered_params
if ctx.get_parameter_source(param.name)
!= click.core.ParameterSource.COMMANDLINE
]
)
ctx.params.update(parsed_params)
return ctx
def check_and_fix_config():
config_path = Path.home() / ".votify" / "config.ini"
if config_path.exists():
try:
content = config_path.read_text(encoding='utf-8')
if "template_folder_music_video" in content and "Unknown Album" in content:
lines = content.splitlines()
new_lines = []
updated = False
for line in lines:
if "template_folder_music_video" in line and "Unknown Album" in line:
new_lines.append("template_folder_music_video = {artist}/{album}")
updated = True
else:
new_lines.append(line)
if updated:
new_content = "\n".join(new_lines)
config_path.write_text(new_content, encoding='utf-8')
except Exception as e:
pass
try:
check_and_fix_config()
except:
pass
@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=5,
help="Wait interval between downloads in seconds.",
)
@click.option(
"--disable-wvd",
is_flag=True,
help="Disable Widevine decryption.",
)
@click.option(
"--download-music-videos",
is_flag=True,
help="List and select a related music video to download from songs.",
)
@click.option(
"--download-podcast-videos",
is_flag=True,
help="Attempt to download the video version of podcasts.",
)
@click.option(
"--force-premium",
"-f",
is_flag=True,
help="Force to detect the account as premium.",
)
@click.option(
"--read-urls-as-txt",
"-r",
is_flag=True,
help="Interpret URLs as paths to text files containing URLs.",
)
@click.option(
"--config-path",
type=Path,
default=Path.home() / ".votify" / "config.ini",
help="Path to config file.",
)
@click.option(
"--log-level",
type=str,
default="INFO",
help="Log level.",
)
@click.option(
"--no-exceptions",
is_flag=True,
help="Don't print exceptions.",
)
@click.option(
"--cookies-path",
type=Path,
default=Path("./cookies.txt"),
help="Path to cookies file.",
)
# Downloader specific options
@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(
"--wvd-path",
type=Path,
default=downloader_sig.parameters["wvd_path"].default,
help="Path to .wvd file.",
)
@click.option(
"--aria2c-path",
type=str,
default=downloader_sig.parameters["aria2c_path"].default,
help="Path to aria2c binary.",
)
@click.option(
"--ffmpeg-path",
type=str,
default=downloader_sig.parameters["ffmpeg_path"].default,
help="Path to ffmpeg binary.",
)
@click.option(
"--mp4box-path",
type=str,
default=downloader_sig.parameters["mp4box_path"].default,
help="Path to MP4Box binary.",
)
@click.option(
"--mp4decrypt-path",
type=str,
default=downloader_sig.parameters["mp4decrypt_path"].default,
help="Path to mp4decrypt binary.",
)
@click.option(
"--packager-path",
type=str,
default=downloader_sig.parameters["packager_path"].default,
help="Path to Shaka Packager binary.",
)
@click.option(
"--spotify-secrets-url",
type=str,
default="https://code.thetadev.de/ThetaDev/spotify-secrets/raw/branch/main/secrets/secretDict.json",
help="Spotify secrets for TOTP generation"
)
@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 music videos.",
)
@click.option(
"--template-folder-music-video",
type=str,
default=downloader_sig.parameters["template_folder_music_video"].default,
help="Template folder for music videos",
)
@click.option(
"--template-file-music-video",
type=str,
default=downloader_sig.parameters["template_file_music_video"].default,
help="Template file for the tracks that are not part of an album.",
)
@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(
"--cover-size",
type=CoverSize,
default=downloader_sig.parameters["cover_size"].default,
help="Cover size.",
)
@click.option(
"--save-cover",
is_flag=True,
help="Save cover as a separate file.",
)
@click.option(
"--save-playlist",
is_flag=True,
help="Save a M3U8 playlist file when downloading a playlist.",
)
@click.option(
"--overwrite",
is_flag=True,
help="Overwrite existing files.",
)
@click.option(
"--exclude-tags",
type=Csv(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.",
)
# DownloaderAudio specific options
@click.option(
"--audio-quality",
"-a",
type=AudioQuality,
default=downloader_audio_sig.parameters["audio_quality"].default,
help="Audio quality for songs and podcasts.",
)
@click.option(
"--download-mode",
"-d",
type=DownloadMode,
default=downloader_audio_sig.parameters["download_mode"].default,
help="Download mode for songs and podcasts.",
)
@click.option(
"--remux-mode-audio",
type=RemuxModeAudio,
default=downloader_audio_sig.parameters["remux_mode"].default,
help="Remux mode for songs and podcasts.",
)
# DownloaderSong specific options
@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.",
)
# DownloaderVideo specific options
@click.option(
"--video-format",
type=VideoFormat,
default=downloader_video_sig.parameters["video_format"].default,
help="Video format.",
)
@click.option(
"--remux-mode-video",
type=RemuxModeVideo,
default=downloader_video_sig.parameters["remux_mode"].default,
help="Remux mode for videos.",
)
# 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,
disable_wvd: bool,
download_music_videos: bool,
download_podcast_videos: bool,
force_premium: bool,
read_urls_as_txt: bool,
config_path: Path,
log_level: str,
no_exceptions: bool,
cookies_path: Path,
output_path: Path,
temp_path: Path,
wvd_path: Path,
download_mode: DownloadMode,
aria2c_path: str,
ffmpeg_path: str,
mp4box_path: str,
mp4decrypt_path: str,
packager_path: str,
spotify_secrets_url: 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_folder_music_video: str,
template_file_music_video: str,
template_file_playlist: str,
date_tag_template: str,
cover_size: CoverSize,
save_cover: bool,
save_playlist: bool,
overwrite: bool,
exclude_tags: list[str],
truncate: int,
audio_quality: AudioQuality,
remux_mode_audio: RemuxModeAudio,
lrc_only: bool,
no_lrc: bool,
video_format: VideoFormat,
remux_mode_video: RemuxModeVideo,
no_config_file: bool,
) -> None:
colorama.just_fix_windows_console()
logger.setLevel(log_level)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(CustomLoggerFormatter())
logger.addHandler(stream_handler)
cookies_path = prompt_path(True, cookies_path, "Cookies file")
logger.info("Starting Votify")
spotify_api = SpotifyApi.from_cookies_file(cookies_path, secrets_url=spotify_secrets_url)
downloader = Downloader(
spotify_api,
output_path,
temp_path,
wvd_path,
aria2c_path,
ffmpeg_path,
mp4box_path,
mp4decrypt_path,
packager_path,
template_folder_album,
template_folder_compilation,
template_file_single_disc,
template_file_multi_disc,
template_folder_episode,
template_file_episode,
template_folder_music_video,
template_file_music_video,
template_file_playlist,
date_tag_template,
cover_size,
save_cover,
save_playlist,
overwrite,
exclude_tags,
truncate,
)
downloader_audio = DownloaderAudio(
downloader,
audio_quality,
download_mode,
remux_mode_audio,
)
downloader_song = DownloaderSong(
downloader_audio,
lrc_only,
no_lrc,
)
downloader_episode = DownloaderEpisode(
downloader_audio,
)
downloader_video = DownloaderVideo(
downloader,
video_format,
remux_mode_video,
)
downloader_episode_video = DownloaderEpisodeVideo(
downloader_video,
downloader_episode,
)
downloader_music_video = DownloaderMusicVideo(
downloader_video,
)
is_premium = (
True if force_premium else spotify_api.user_profile['data']['me']['account']['product'] == "premium"
)
if not lrc_only:
if download_mode == DownloadMode.ARIA2C and not downloader.aria2c_path_full:
logger.critical(X_NOT_FOUND_STRING.format("aria2c", aria2c_path))
return
if not is_premium and audio_quality in PREMIUM_AUDIO_QUALITIES:
logger.critical("Cannot download at chosen quality with a free account")
return
if audio_quality in AAC_AUDIO_QUALITIES:
if (
remux_mode_audio == RemuxModeAudio.FFMPEG
and not downloader.ffmpeg_path_full
):
logger.critical(X_NOT_FOUND_STRING.format("FFmpeg", ffmpeg_path))
if (
remux_mode_audio == RemuxModeAudio.MP4BOX
and not downloader.mp4box_path_full
):
logger.critical(X_NOT_FOUND_STRING.format("MP4Box", mp4box_path))
return
if (
remux_mode_audio
in (
RemuxModeAudio.MP4DECRYPT,
RemuxModeAudio.MP4BOX,
)
and not downloader.mp4decrypt_path_full
):
logger.critical(
X_NOT_FOUND_STRING.format("mp4decrypt", mp4decrypt_path)
)
return
if download_podcast_videos or download_music_videos:
if (
downloader_music_video.remux_mode == RemuxModeVideo.FFMPEG
and not downloader.ffmpeg_path_full
):
logger.critical(X_NOT_FOUND_STRING.format("FFmpeg", ffmpeg_path))
return
if (
downloader_music_video.remux_mode == RemuxModeVideo.MP4BOX
and not downloader.mp4box_path_full
):
logger.critical(X_NOT_FOUND_STRING.format("MP4Box", mp4box_path))
return
if download_music_videos:
if not downloader.mp4decrypt_path_full and video_format == VideoFormat.MP4:
logger.critical(
X_NOT_FOUND_STRING.format("mp4decrypt", mp4decrypt_path)
)
return
if not downloader.packager_path_full and video_format == VideoFormat.WEBM:
logger.critical(
X_NOT_FOUND_STRING.format("Shaka Packager", packager_path)
)
if not disable_wvd:
wvd_path = prompt_path(True, wvd_path, ".wvd file", optional=True)
downloader.set_cdm()
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
error_count = 0
global_playlist_metadata = None
for url_index, url in enumerate(urls, start=1):
url_progress = color_text(f"URL {url_index}/{len(urls)}", colorama.Style.DIM)
logger.info(f'({url_progress}) Checking "{url}"')
try:
url_info = downloader.get_url_info(url)
if url_info.type == 'playlist':
if hasattr(url_info, 'name') and url_info.name:
global_playlist_metadata = {'name': url_info.name, 'owner_name': getattr(url_info, 'owner', 'Unknown')}
if url_info.type == "artist":
download_queue = downloader.get_download_queue_from_artist(url_info.id)
else:
download_queue = downloader.get_download_queue(
url_info.type,
url_info.id,
)
except Exception as e:
error_count += 1
logger.error(
f'({url_progress}) Failed to check "{url}"',
exc_info=not no_exceptions,
)
continue
for index, download_queue_item in enumerate(download_queue, start=1):
queue_progress = color_text(
f"Track {index}/{len(download_queue)} from URL {url_index}/{len(urls)}",
colorama.Style.DIM,
)
media_metadata = None
if isinstance(download_queue_item, dict):
media_metadata = download_queue_item.get("media_metadata", download_queue_item)
else:
media_metadata = getattr(download_queue_item, "media_metadata", download_queue_item)
if not isinstance(media_metadata, dict):
media_metadata = getattr(media_metadata, "__dict__", {})
if isinstance(media_metadata, dict) and 'track' in media_metadata and isinstance(media_metadata['track'],dict):
clean_track_metadata = media_metadata['track']
else:
clean_track_metadata = media_metadata
track_name = "Unknown Track"
try:
track_name = media_metadata["data"]["trackUnion"]["name"]
except (KeyError, TypeError):
try:
track_name = media_metadata["data"]["episodeUnionV2"]["name"]
except (KeyError, TypeError):
track_name = clean_track_metadata.get("name", "Unknown Track")
is_playable = True
if isinstance(clean_track_metadata, dict):
if clean_track_metadata.get('isPlayable') is False:
is_playable = False
elif clean_track_metadata.get('is_playable') is False:
is_playable = False
elif clean_track_metadata.get('playable') is False:
is_playable = False
elif 'data' in clean_track_metadata:
try:
union = clean_track_metadata['data']['trackUnion']
if union.get('isPlayable') is False or union.get('playable') is False:
is_playable = False
except:
pass
if not is_playable:
logger.warning(f'({queue_progress}) Skipping "{track_name}": Track unavailable/unplayable.')
continue
try:
logger.info(f'({queue_progress}) Downloading "{track_name}"')
try:
media_id = downloader.get_media_id(clean_track_metadata)
except ValueError:
if isinstance(download_queue_item, dict) and 'id' in download_queue_item:
media_id = download_queue_item['id']
else:
raise
media_metadata_for_download = media_metadata
media_type = "track"
target_uri = ""
try:
target_uri = media_metadata['data']['trackUnion']['uri']
except (KeyError, TypeError):
target_uri = clean_track_metadata.get('uri', '')
if "episode" in target_uri:
media_type = "episode"
elif "track" in target_uri:
media_type = "track"
else:
try:
type_from_meta = media_metadata['data']['trackUnion']['__typename']
media_type = type_from_meta.lower()
except (KeyError, TypeError, AttributeError):
if isinstance(media_metadata, dict):
if media_metadata.get('type') == 'episode' or 'episode' in media_metadata:
media_type = "episode"
if 'episode' in media_metadata:
media_metadata_for_download = None
gid_metadata = downloader.get_gid_metadata(media_id, media_type, spotify_api.user_profile, track_name, download_music_videos, download_podcast_videos)
def get_meta(item, key):
val = getattr(item, key, None)
if val is None and isinstance(item, dict):
val = item.get(key)
return val
safe_album_metadata = get_meta(download_queue_item, "album_metadata")
safe_playlist_metadata = get_meta(download_queue_item, "playlist_metadata")
safe_show_metadata = get_meta(download_queue_item, "show_metadata")
if not safe_playlist_metadata and global_playlist_metadata:
safe_playlist_metadata = global_playlist_metadata
if media_type == "track":
if disable_wvd:
logger.warning("Widevine decryption is disabled, skipping.")
continue
if audio_quality in VORBIS_AUDIO_QUALITIES:
logger.warning("Vorbis audio quality is only supported for podcasts.")
continue
if download_music_videos:
downloader_music_video.download(
music_video_id=media_id,
music_video_metadata=media_metadata_for_download,
album_metadata=safe_album_metadata,
gid_metadata=gid_metadata,
playlist_metadata=safe_playlist_metadata,
playlist_track=index,
)
else:
downloader_song.download(
track_id=media_id,
track_metadata=media_metadata_for_download,
album_metadata=safe_album_metadata,
gid_metadata=gid_metadata,
playlist_metadata=safe_playlist_metadata,
product_name=spotify_api.user_profile,
playlist_track=index,
)
elif media_type == "episode":
if download_podcast_videos:
downloader_episode_video.download(
episode_id=media_id,
episode_metadata=media_metadata_for_download,
show_metadata=safe_show_metadata,
gid_metadata=gid_metadata,
playlist_metadata=safe_playlist_metadata,
playlist_track=index,
)
else:
downloader_episode.download(
episode_id=media_id,
episode_metadata=media_metadata_for_download,
show_metadata=safe_show_metadata,
gid_metadata=gid_metadata,
playlist_metadata=safe_playlist_metadata,
product_name=spotify_api.user_profile,
playlist_track=index,
)
except Exception as e:
error_count += 1
logger.error(f'({queue_progress}) Failed to download "{track_name}"', exc_info=False)
#logger.error(f'({queue_progress}) Failed to download "{track_name}"', exc_info=not no_exceptions)
finally:
if wait_interval > 0 and index != len(download_queue):
logger.debug(
f"Waiting for {wait_interval} second(s) before continuing"
)
time.sleep(wait_interval)
logger.info(f"Done ({error_count} error(s))")
-101
View File
@@ -1,101 +0,0 @@
from __future__ import annotations
import configparser
from enum import Enum
from pathlib import Path
import click
import typing
class ConfigFile:
def __init__(
self,
config_path: Path,
section_name: str = "votify",
) -> None:
self.config_path = config_path
self.section_name = section_name
self._read_config_file()
def _read_config_file(self) -> None:
self.config = configparser.ConfigParser(interpolation=None)
if self.config_path.exists():
self.config.read(self.config_path, encoding="utf-8")
else:
self.config_path.parent.mkdir(parents=True, exist_ok=True)
if not self.config.has_section(self.section_name):
self.config.add_section(self.section_name)
def _write_config_file(self) -> None:
with self.config_path.open("w", encoding="utf-8") as config_file:
self.config.write(config_file)
def _serialize_param_default(self, param: click.Parameter) -> str:
if not isinstance(param.default, (list, tuple)):
param_default = [param.default]
else:
param_default = param.default
if not param_default:
return ""
first = param_default[0]
if isinstance(first, Enum):
return ",".join(str(item.value) for item in param_default)
if isinstance(first, bool):
return ",".join(str(item).lower() for item in param_default)
if first is None:
return "null"
return ",".join(str(item) for item in param_default)
def _add_param_default_to_config(
self,
param: click.Parameter,
) -> bool:
if self.config[self.section_name].get(param.name):
return False
value = self._serialize_param_default(param)
self.config[self.section_name][param.name] = value
return True
def _parse_param_from_config(
self,
param: click.Parameter,
) -> typing.Any:
value = self.config[self.section_name].get(param.name)
if value == "null":
return None
return param.type_cast_value(None, value)
def add_params_default_to_config(
self,
params: list[click.Parameter],
) -> None:
has_changes = False
for param in params:
has_changes = self._add_param_default_to_config(param) or has_changes
if has_changes:
self._write_config_file()
def parse_params_from_config(
self,
params: list[click.Parameter],
) -> dict[str, typing.Any]:
parsed_params = {}
for param in params:
parsed_params[param.name] = self._parse_param_from_config(param)
return parsed_params
-94
View File
@@ -1,94 +0,0 @@
from __future__ import annotations
from .enums import AudioQuality, CoverSize
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",
}
MP4_TAGS_MAP = {
"album": "\xa9alb",
"album_artist": "aART",
"artist": "\xa9ART",
"composer": "\xa9wrt",
"copyright": "cprt",
"lyrics": "\xa9lyr",
"publisher": "\xa9pub",
"producer": "\xa9prd",
"rating": "rtng",
"release_date": "\xa9day",
"title": "\xa9nam",
"url": "\xa9url",
}
AUDIO_QUALITY_X_FORMAT_ID_MAPPING = {
AudioQuality.VORBIS_HIGH: "OGG_VORBIS_320",
AudioQuality.VORBIS_MEDIUM: "OGG_VORBIS_160",
AudioQuality.VORBIS_LOW: "OGG_VORBIS_96",
AudioQuality.AAC_HIGH: "MP4_256",
AudioQuality.AAC_MEDIUM: "MP4_128",
}
VORBIS_AUDIO_QUALITIES = (
AudioQuality.VORBIS_HIGH,
AudioQuality.VORBIS_MEDIUM,
AudioQuality.VORBIS_LOW,
)
AAC_AUDIO_QUALITIES = (AudioQuality.AAC_HIGH, AudioQuality.AAC_MEDIUM)
X_NOT_FOUND_STRING = "{} not found at {}"
PREMIUM_AUDIO_QUALITIES = (AudioQuality.VORBIS_HIGH, AudioQuality.AAC_HIGH)
MEDIA_TYPE_MP4_MAPPING = {
"Song": 1,
"Podcast": 21,
"Music video": 6,
}
COVER_SIZE_X_KEY_MAPPING_SONG = {
CoverSize.SMALL: "ab67616d00004851",
CoverSize.MEDIUM: "ab67616d00001e02",
CoverSize.LARGE: "ab67616d0000b273",
CoverSize.EXTRA_LARGE: "ab67616d000082c1",
}
COVER_SIZE_X_KEY_MAPPING_EPISODE= {
CoverSize.SMALL: "ab6765630000f68d",
CoverSize.MEDIUM: "ab67656300005f1f",
CoverSize.LARGE: "ab6765630000ba8a",
CoverSize.EXTRA_LARGE: "ab6765630000ba8a",
}
COVER_SIZE_X_KEY_MAPPING_VIDEO = {
CoverSize.SMALL: "ab6742d3000052b7",
CoverSize.MEDIUM: "ab6742d3000052b7",
CoverSize.LARGE: "ab6742d3000053b7",
CoverSize.EXTRA_LARGE: "ab6742d3000053b7",
}
-24
View File
@@ -1,24 +0,0 @@
import logging
import colorama
from .utils import color_text
class CustomLoggerFormatter(logging.Formatter):
base_format = "[%(levelname)-8s %(asctime)s]"
format_colors = {
logging.DEBUG: colorama.Style.DIM,
logging.INFO: colorama.Fore.GREEN,
logging.WARNING: colorama.Fore.YELLOW,
logging.ERROR: colorama.Fore.RED,
logging.CRITICAL: colorama.Fore.RED,
}
date_format = "%H:%M:%S"
def format(self, record: logging.LogRecord) -> str:
return logging.Formatter(
color_text(self.base_format, self.format_colors.get(record.levelno))
+ " %(message)s",
datefmt=self.date_format,
).format(record)
-119
View File
@@ -1,119 +0,0 @@
from __future__ import annotations
import json
import re
import time
from urllib.parse import parse_qs, urlparse
import requests
from .utils import check_response
class SpotifyDeviceFlow:
DEVICE_AUTH_URL = "https://accounts.spotify.com/oauth2/device/authorize"
DEVICE_TOKEN_URL = "https://accounts.spotify.com/api/token"
DEVICE_RESOLVE_URL = "https://accounts.spotify.com/pair/api/resolve"
DEVICE_CLIENT_ID = "65b708073fc0480ea92a077233ca87bd" # Spotify for Desktop
DEVICE_SCOPE = "app-remote-control,playlist-modify,playlist-modify-private,playlist-modify-public,playlist-read,playlist-read-collaborative,playlist-read-private,streaming,transfer-auth-session,ugc-image-upload,user-follow-modify,user-follow-read,user-library-modify,user-library-read,user-modify,user-modify-playback-state,user-modify-private,user-personalized,user-read-birthdate,user-read-currently-playing,user-read-email,user-read-play-history,user-read-playback-position,user-read-playback-state,user-read-private,user-read-recently-played,user-top-read"
DEVICE_FLOW_USER_AGENT = "Spotify/126600447 Win32_x86_64/0 (PC laptop)"
def __init__(self, sp_dc: str) -> None:
self.session = requests.Session()
self.session.cookies.update({"sp_dc": sp_dc})
def get_token(self) -> dict:
"""Get access token via device flow."""
auth_data = self._initiate_device_authorization()
device_code = auth_data["device_code"]
user_code = auth_data["user_code"]
verification_url = auth_data["verification_uri_complete"]
flow_ctx, csrf_token = self._parse_verification_page(verification_url)
self._submit_user_code(user_code, flow_ctx, csrf_token, verification_url)
return self._exchange_device_code(device_code)
def _initiate_device_authorization(self) -> dict:
response = requests.post(
self.DEVICE_AUTH_URL,
data={
"client_id": self.DEVICE_CLIENT_ID,
"scope": self.DEVICE_SCOPE,
},
headers={
"User-Agent": self.DEVICE_FLOW_USER_AGENT,
"Content-Type": "application/x-www-form-urlencoded",
},
)
check_response(response)
return response.json()
def _parse_verification_page(self, verification_url: str) -> tuple[str, str]:
response = self.session.get(
verification_url,
allow_redirects=True,
)
check_response(response)
parsed_url = urlparse(response.url)
try:
flow_ctx_full = parse_qs(parsed_url.query)["flow_ctx"][0]
flow_ctx = flow_ctx_full.split(":")[0]
except (KeyError, IndexError):
raise ValueError("Failed to extract flow_ctx")
csrf_token = self._extract_csrf_token(response.text)
return flow_ctx, csrf_token
def _extract_csrf_token(self, html_content: str) -> str | None:
pattern = (
r'<script id="__NEXT_DATA__" type="application/json"[^>]*>(.*?)</script>'
)
match = re.search(pattern, html_content, re.DOTALL)
try:
json_data = json.loads(match.group(1))
return json_data["props"]["initialToken"]
except (AttributeError, json.JSONDecodeError, KeyError):
raise ValueError("Failed to extract CSRF token")
def _submit_user_code(
self,
user_code: str,
flow_ctx: str,
csrf_token: str,
referer_url: str,
) -> None:
current_ts = int(time.time())
response = self.session.post(
self.DEVICE_RESOLVE_URL,
params={"flow_ctx": f"{flow_ctx}:{current_ts}"},
json={"code": user_code},
headers={
"x-csrf-token": csrf_token,
"referer": referer_url,
"origin": "https://accounts.spotify.com",
"content-type": "application/json",
},
)
check_response(response)
try:
submit_result = response.json()
assert submit_result.get("result") == "ok"
except (json.JSONDecodeError, AssertionError):
raise ValueError("Failed to submit user code")
def _exchange_device_code(self, device_code: str) -> dict:
response = requests.post(
self.DEVICE_TOKEN_URL,
data={
"client_id": self.DEVICE_CLIENT_ID,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
},
headers={
"User-Agent": self.DEVICE_FLOW_USER_AGENT,
"Content-Type": "application/x-www-form-urlencoded",
},
)
check_response(response)
return response.json()
-951
View File
@@ -1,951 +0,0 @@
from __future__ import annotations
import base64
import datetime
import functools
import logging
import re
import shutil
import subprocess
from io import BytesIO
from pathlib import Path
from .utils import prompt_path
import requests
from InquirerPy import inquirer
from InquirerPy.base.control import Choice
from mutagen.flac import Picture
from mutagen.mp4 import MP4, MP4Cover, MP4FreeForm
from mutagen.oggvorbis import OggVorbis, OggVorbisHeaderError
from PIL import Image
from pywidevine import PSSH, Cdm, Device
from .constants import (
MEDIA_TYPE_MP4_MAPPING,
MP4_TAGS_MAP,
VORBIS_TAGS_MAPPING,
)
from .enums import CoverSize
from .models import DownloadQueueItem, UrlInfo
from .spotify_api import SpotifyApi
from .utils import check_response
logger = logging.getLogger("votify")
class Downloader:
ILLEGAL_CHARACTERS_REGEX = r'[\\/:*?"<>|;]'
URL_RE = r"(album|playlist|track|show|episode|artist)/(\w{22})"
ILLEGAL_CHARACTERS_REPLACEMENT = "_"
RELEASE_DATE_PRECISION_MAPPING = {
"year": "%Y",
"month": "%Y-%m",
"day": "%Y-%m-%d",
}
def __init__(
self,
spotify_api: SpotifyApi,
output_path: Path = Path("./Spotify"),
temp_path: Path = Path("./temp"),
wvd_path: Path = Path("./device.wvd"),
aria2c_path: str = "aria2c",
ffmpeg_path: str = "ffmpeg",
mp4box_path: str = "mp4box",
mp4decrypt_path: str = "mp4decrypt",
packager_path: str = "packager",
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_folder_music_video: str = "{artist}/{album}",
template_file_music_video: str = "{title}",
template_file_playlist: str = "Playlists/{playlist_artist}/{playlist_title}",
date_tag_template: str = "%Y-%m-%dT%H:%M:%SZ",
cover_size: CoverSize = CoverSize.EXTRA_LARGE,
save_cover: bool = False,
save_playlist: bool = False,
overwrite: bool = False,
exclude_tags: str = None,
truncate: int = None,
silence: bool = False,
skip_cleanup: bool = False,
):
self.spotify_api = spotify_api
self.output_path = output_path
self.temp_path = temp_path
self.wvd_path = prompt_path(
is_file=True,
initial_path=wvd_path,
description="device.wvd",
optional=True
)
self.aria2c_path = aria2c_path
self.ffmpeg_path = ffmpeg_path
self.mp4box_path = mp4box_path
self.mp4decrypt_path = mp4decrypt_path
self.packager_path = packager_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_folder_music_video = template_folder_music_video
self.template_file_music_video = template_file_music_video
self.template_file_playlist = template_file_playlist
self.date_tag_template = date_tag_template
self.cover_size = cover_size
self.save_cover = save_cover
self.save_playlist = save_playlist
self.overwrite = overwrite
self.exclude_tags = exclude_tags
self.truncate = truncate
self.silence = silence
self.skip_cleanup = skip_cleanup
self.cdm = None
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.ffmpeg_path_full = shutil.which(self.ffmpeg_path)
self.mp4box_path_full = shutil.which(self.mp4box_path)
self.mp4decrypt_path_full = shutil.which(self.mp4decrypt_path)
self.packager_path_full = shutil.which(self.packager_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 set_cdm(self) -> None:
if self.wvd_path.exists():
self.cdm = Cdm.from_device(Device.load(self.wvd_path))
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,
media_type: str,
media_id: str,
) -> list[DownloadQueueItem]:
download_queue = []
if media_type == "album":
album = self.spotify_api.get_album(media_id)
for item in album:
t = item['track']
track_id = t['uri'].split(':')[-1]
download_queue.append({
'data': {
'trackUnion': {
'id': track_id,
'uri': t['uri'],
'name': t['name'],
'trackNumber': t['trackNumber'],
'duration': {
'totalMilliseconds': t['duration']['totalMilliseconds']
},
'is_playable': t.get('is_playable', True),
'artists': {
'items': [
{
'profile': {
'name': t['artists']['items'][0]['profile']['name']
}
}
]
}
}
}
})
elif media_type == "playlist":
playlist = self.spotify_api.get_playlist(media_id)
playlist_metadata = playlist
if 'content' in playlist and 'items' in playlist['content']:
items_list = playlist['content']['items']
else:
items_list = []
for item in items_list:
if 'itemV2' not in item or 'data' not in item['itemV2']:
continue
t = item['itemV2']['data']
if 'uri' not in t:
continue
track_id = t['uri'].split(':')[-1]
duration_ms = 0
if 'duration' in t and isinstance(t['duration'], dict):
duration_ms = t['duration'].get('totalMilliseconds', 0)
elif 'trackDuration' in t and isinstance(t['trackDuration'], dict):
duration_ms = t['trackDuration'].get('totalMilliseconds', 0)
album_data = t.get('albumOfTrack', {})
if 'uri' in album_data and 'id' not in album_data:
album_data['id'] = album_data['uri'].split(':')[-1]
if 'date' in album_data and isinstance(album_data['date'], dict):
date_obj = album_data['date']
if 'isoString' in date_obj and date_obj['isoString']:
album_data['release_date'] = date_obj['isoString']
album_data['release_date_precision'] = 'day'
elif 'year' in date_obj and date_obj['year']:
album_data['release_date'] = str(date_obj['year'])
album_data['release_date_precision'] = 'year'
if 'release_date' not in album_data:
album_data['release_date'] = '1970-01-01'
album_data['release_date_precision'] = 'day'
artist_name = "Unknown Artist"
try:
if 'artists' in t and 'items' in t['artists'] and len(t['artists']['items']) > 0:
profile = t['artists']['items'][0].get('profile')
if profile:
artist_name = profile.get('name', "Unknown Artist")
except Exception:
pass
is_playable = t.get('isPlayable', True)
track_object = {
'data': {
'trackUnion': {
'id': track_id,
'uri': t['uri'],
'__typename': 'Track',
'name': t.get('name', 'Unknown Track'),
'trackNumber': t.get('trackNumber', 1),
'duration': {
'totalMilliseconds': duration_ms
},
'isPlayable': is_playable,
'albumOfTrack': album_data,
'artists': {
'items': [
{
'profile': {
'name': artist_name
}
}
]
}
}
}
}
download_queue.append({
'media_metadata': track_object,
'playlist_metadata': playlist_metadata
})
elif media_type == "track":
download_queue.append(
DownloadQueueItem(
media_metadata=self.spotify_api.get_track(media_id),
)
)
elif media_type == "episode":
episode = self.spotify_api.get_episode(media_id)
podcast_name = "Unknown Podcast"
podcast_uri = ""
podcast_cover = ""
if 'podcastV2' in episode and 'data' in episode['podcastV2']:
p_data = episode['podcastV2']['data']
podcast_name = p_data.get('name', 'Unknown Podcast')
podcast_uri = p_data.get('uri', '')
if 'coverArt' in p_data and 'sources' in p_data['coverArt']:
podcast_cover = p_data['coverArt']['sources'][0]['url']
release_date = ""
precision = ""
if 'releaseDate' in episode and episode['releaseDate']:
release_date = episode['releaseDate'].get('isoString')
precision = episode['releaseDate'].get('isoString')
cover_url = podcast_cover
try:
if 'coverArt' in episode and 'sources' in episode['coverArt']:
sources = episode['coverArt']['sources']
cover_url = max(sources, key=lambda x: x.get('width', 0))['url']
except Exception:
pass
download_queue.append({
'data': {
'trackUnion': {
'id': episode['id'],
'uri': episode['uri'],
'name': episode['name'],
'description': episode.get('description', ''),
'trackNumber': 1,
'duration': {
'totalMilliseconds': episode['duration']['totalMilliseconds']
},
'is_playable': episode.get('playability', {}).get('playable', True),
'release_date': release_date,
'release_date_precision': precision,
'cover_url': cover_url,
'artists': {
'items': [
{
'profile': {
'name': podcast_name
}
}
]
},
'albumOfTrack': {
'uri': podcast_uri,
'name': podcast_name,
'coverArt': {
'sources': [{'url': cover_url}]
},
'id': podcast_uri.split(':')[-1] if ':' in podcast_uri else None,
'release_date': release_date
}
}
},
'show_metadata': {
'name': podcast_name,
'publisher': 'Unknown',
'uri': podcast_uri
}
})
elif media_type == "show":
show = self.spotify_api.get_show(media_id)
items_list = []
if 'episodesV2' in show and 'items' in show['episodesV2']:
items_list = show['episodesV2']['items']
elif 'episodes' in show and 'items' in show['episodes']:
items_list = show['episodes']['items']
total_items = len(items_list)
for index, item in enumerate(items_list):
episode_raw = None
if 'entity' in item and 'data' in item['entity']:
episode_raw = item['entity']['data']
elif 'data' in item:
episode_raw = item['data']
else:
episode_raw = item
if not episode_raw or not isinstance(episode_raw, dict):
continue
podcast_name = "Unknown Podcast"
podcast_uri = ""
if 'podcastV2' in episode_raw and 'data' in episode_raw['podcastV2']:
p_data = episode_raw['podcastV2']['data']
podcast_name = p_data.get('name', 'Unknown Podcast')
podcast_uri = p_data.get('uri', '')
elif show.get('name'):
podcast_name = show.get('name')
podcast_uri = show.get('uri')
cover_url = None
if 'coverArt' in episode_raw and episode_raw['coverArt'].get('sources'):
try:
sources = episode_raw['coverArt']['sources']
cover_url = max(sources, key=lambda x: x.get('width', 0))['url']
except Exception:
pass
release_date = ""
precision = "day"
if 'releaseDate' in episode_raw and isinstance(episode_raw['releaseDate'], dict):
release_date = episode_raw['releaseDate'].get('isoString')
if 'T' in release_date:
release_date = release_date.split('T')[0]
precision = episode_raw['releaseDate'].get('precision', 'day')
elif 'release_date' in episode_raw:
release_date = episode_raw['release_date']
current_track_number = total_items - index
formatted_episode = {
'data': {
'trackUnion': {
'id': episode_raw.get('id'),
'uri': episode_raw.get('uri'),
'name': episode_raw.get('name'),
'description': episode_raw.get('description', ''),
'trackNumber': current_track_number,
'duration': {
'totalMilliseconds': episode_raw.get('duration', {}).get('totalMilliseconds', 0)
},
'is_playable': episode_raw.get('playability', {}).get('playable', True),
'release_date': release_date,
'release_date_precision': precision,
'cover_url': cover_url,
'artists': {'items': [{'profile': {'name': podcast_name}}]},
'albumOfTrack': {
'uri': podcast_uri,
'name': podcast_name,
'coverArt': {'sources': [{'url': cover_url}]},
'id': podcast_uri.split(':')[-1] if ':' in podcast_uri else None,
'release_date': release_date
}
}
}
}
current_show_metadata = {
'name': podcast_name,
'publisher': 'Unknown',
'uri': podcast_uri
}
download_queue.append(
DownloadQueueItem(
show_metadata=current_show_metadata,
media_metadata=formatted_episode,
)
)
return download_queue
def _filter_artist_albums(
self,
artist_albums: list[dict],
album_type: str,
) -> list[dict]:
return [album for album in artist_albums if album == album_type]
def get_download_queue_from_artist(self, artist_id: str) -> list[DownloadQueueItem]:
download_queue = []
selected_option = inquirer.select(
message="Select which album type to download for artist:",
choices=[
Choice(name="Albuns", value="album"),
Choice(name="Singles", value="single"),
Choice(name="Compilations", value="compilation"),
Choice(name="Collaborations", value="appears_on"),
],
).execute()
selected_ids = []
if selected_option == "album":
selected_ids = self.spotify_api.get_artist_albums_selection(artist_id)
elif selected_option == "single":
selected_ids = self.spotify_api.get_artist_singles_selection(artist_id)
elif selected_option == "compilation":
selected_ids = self.spotify_api.get_artist_compilations_selection(artist_id)
elif selected_option == "appears_on":
selected_ids = self.spotify_api.get_artist_collaborations_selection(artist_id)
if selected_ids:
print(f"Processing {len(selected_ids)} items...")
for album_id in selected_ids:
download_queue.extend(
self.get_download_queue("album", album_id)
)
return download_queue
def get_cover_url(self, metadata: dict, cover_size_mapping: dict) -> str | None:
if not isinstance(metadata, dict):
return None
images = None
try:
images = (
metadata.get("data", {})
.get("trackUnion", {})
.get("albumOfTrack", {})
.get("coverArt", {})
.get("sources")
)
except AttributeError:
pass
if not images:
images = metadata.get("images")
if not images:
images = metadata.get("coverArt", {}).get("sources")
if not images:
return None
return self._get_cover_url(images, cover_size_mapping)
def _get_cover_url(self, images_dict: list[dict], cover_size_mapping: dict) -> str:
original_cover_url = images_dict[0]["url"]
original_cover_id = original_cover_url.split("/")[-1]
cover_key = cover_size_mapping[self.cover_size]
cover_id = cover_key + original_cover_id[len(cover_key) :]
cover_url = f"{original_cover_url.rpartition('/')[0]}/{cover_id}"
return cover_url
def get_media_id(self, media_metadata: dict) -> str:
linked_from = media_metadata.get("linked_from")
if linked_from:
media_metadata = linked_from
if "data" in media_metadata and isinstance(media_metadata["data"], dict):
track_union = media_metadata["data"].get("trackUnion")
if track_union and "id" in track_union:
return track_union["id"]
if "id" in media_metadata:
return media_metadata["id"]
if "itemV2" in media_metadata:
return media_metadata["itemV2"]["data"]["id"]
if "media_metadata" in media_metadata:
nested = media_metadata["media_metadata"]
if "data" in nested and "trackUnion" in nested["data"]:
return nested["data"]["trackUnion"]["id"]
if "id" in nested:
return nested["id"]
if "uri" in media_metadata:
return media_metadata["uri"].split(":")[-1]
raise ValueError("Could not find media ID in metadata")
def get_playlist_tags(self, playlist_metadata: dict, playlist_track: int) -> dict:
playlist_title = playlist_metadata.get("name", "Unknown Playlist")
playlist_artist = "Unknown Artist"
try:
if "ownerV2" in playlist_metadata:
playlist_artist = playlist_metadata['ownerV2']['data']['name']
except (KeyError, TypeError):
pass
if playlist_artist == "Unknown Artist" and "owner" in playlist_metadata:
owner_obj = playlist_metadata["owner"]
if isinstance(owner_obj, dict):
playlist_artist = owner_obj.get("display_name") or owner_obj.get("id", "Unknown Artist")
elif isinstance(owner_obj, str):
playlist_artist = owner_obj
if playlist_artist == "Unknown Artist":
playlist_artist = playlist_metadata.get("owner_name", "Unknown Artist")
return {
"playlist_artist": playlist_artist,
"playlist_title": playlist_title,
"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":
if tags.get("playlist_title"):
template_folder = self.template_file_playlist.split("/")
else:
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("/")
elif media_type == "music-video":
template_folder = self.template_folder_music_video.split("/")
template_file = self.template_file_music_video.split("/")
else:
raise RuntimeError()
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_gid_metadata(
self,
media_id: str,
media_type: str,
audio_quality: dict,
track_name: str,
download_music_video: str,
download_podcast_videos: str,
) -> str:
playback_info = self.spotify_api.get_track_playback_info(media_id, media_type)
user_product = audio_quality.get('data', {}).get('me', {}).get('account', {}).get('product', 'free')
if media_type == 'episode':
target_bitrate = 128000
else:
target_bitrate = 256000 if user_product == 'PREMIUM' else 128000
media_items = playback_info.get("media", {})
if download_music_video or download_podcast_videos:
video_file_id = next(
(
data['item']['manifest']['manifest_ids_video'][0]['file_id']
for data in playback_info['media'].values()
if 'item' in data
and 'manifest' in data['item']
and 'manifest_ids_video' in data['item']['manifest']
),
None
)
if video_file_id is None:
logger.error("There is no video for this media.")
return
return video_file_id
else:
if not media_items:
raise ValueError("No available Track (Empty response)")
expected_key = f"spotify:{media_type}:{media_id}"
track_manifest = media_items.get(expected_key)
if not track_manifest:
track_manifest = list(media_items.values())[0]
metadata = track_manifest.get("item", {}).get("metadata", {})
server_name = metadata.get("name")
if not server_name:
raise ValueError(f"Track Unavailable: No name returned")
req_clean = track_name.lower().strip()
srv_clean = server_name.lower().strip()
if (req_clean in srv_clean) or (srv_clean in req_clean):
track_manifest = track_manifest
else:
raise ValueError(f"Track Unavailable: No name returned")
try:
audio_file_id = next(
f["file_id"]
for files in track_manifest["item"]["manifest"].values()
if isinstance(files, list)
for f in files
if isinstance(f, dict) and f.get("bitrate") == int(target_bitrate)
)
except StopIteration:
raise ValueError(f"No audio file found with bitrate {target_bitrate}")
return audio_file_id
def get_playplay_decryption_key(self, file_id: str) -> bytes:
raise NotImplementedError()
def get_widevine_decryption_key(
self,
pssh: bytes,
media_type: str,
) -> tuple[str, str]:
try:
if self.cdm is None:
cmd = self.spotify_api.extract_keys_with_cdrm(pssh, media_type)
key_id, decryption_key = cmd.split(':')
else:
pssh = PSSH(pssh)
cdm_session = self.cdm.open()
challenge = self.cdm.get_license_challenge(cdm_session, pssh)
license = self.spotify_api.get_widevine_license(challenge, media_type)
self.cdm.parse_license(cdm_session, license)
keys = next(
i for i in self.cdm.get_keys(cdm_session) if i.type == "CONTENT"
)
decryption_key = keys.key.hex()
key_id = keys.kid.hex
finally:
try:
self.cdm.close(cdm_session)
except:
pass
return key_id, decryption_key
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:
if not release_date:
return datetime.datetime(1970, 1, 1)
if not release_date_precision:
if len(release_date) == 4:
release_date_precision = 'year'
elif len(release_date) == 7:
release_date_precision = 'month'
else:
release_date_precision = 'day'
try:
precision_key = str(release_date_precision).lower()
fmt = self.RELEASE_DATE_PRECISION_MAPPING.get(precision_key, "%Y-%m-%d")
return datetime.datetime.strptime(release_date, fmt)
except (ValueError, TypeError, KeyError, AttributeError):
for fmt in ["%Y", "%Y-%m", "%Y-%m-%d"]:
try:
return datetime.datetime.strptime(release_date, fmt)
except ValueError:
continue
return datetime.datetime(1970, 1, 1)
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 not artist_list:
return ""
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_file_temp_path(
self,
track_id: str,
extra_string: str,
extension: str,
) -> Path:
return self.temp_path / (track_id + extra_string + extension)
def apply_tags_ogg(
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
def apply_tags_mp4(self, fixed_location: Path, tags: dict, cover_url: str):
to_apply_tags = [
tag_name
for tag_name in tags.keys()
if tag_name not in self.exclude_tags_list
]
mp4_tags = {}
for tag_name in to_apply_tags:
if tags.get(tag_name) is None:
continue
if tag_name in ("disc", "disc_total"):
if mp4_tags.get("disk") is None:
mp4_tags["disk"] = [[0, 0]]
if tag_name == "disc":
mp4_tags["disk"][0][0] = tags[tag_name]
elif tag_name == "disc_total":
mp4_tags["disk"][0][1] = tags[tag_name]
elif tag_name in ("track", "track_total"):
if mp4_tags.get("trkn") is None:
mp4_tags["trkn"] = [[0, 0]]
if tag_name == "track":
mp4_tags["trkn"][0][0] = tags[tag_name]
elif tag_name == "track_total":
mp4_tags["trkn"][0][1] = tags[tag_name]
elif tag_name == "compilation":
mp4_tags["cpil"] = tags["compilation"]
elif tag_name == "isrc":
mp4_tags["----:com.apple.iTunes:ISRC"] = [
MP4FreeForm(tags["isrc"].encode("utf-8"))
]
elif tag_name == "label":
mp4_tags["----:com.apple.iTunes:LABEL"] = [
MP4FreeForm(tags["label"].encode("utf-8"))
]
elif tag_name == "rating":
mp4_tags["rtng"] = [1] if tags[tag_name] == "Explicit" else [0]
elif tag_name == "media_type":
mp4_tags["stik"] = [MEDIA_TYPE_MP4_MAPPING[tags[tag_name]]]
elif MP4_TAGS_MAP.get(tag_name) is not None:
mp4_tags[MP4_TAGS_MAP[tag_name]] = [tags[tag_name]]
if "cover" not in self.exclude_tags_list and cover_url is not None:
mp4_tags["covr"] = [
MP4Cover(
self.get_response_bytes(cover_url), imageformat=MP4Cover.FORMAT_JPEG
)
]
mp4 = MP4(fixed_location)
mp4.clear()
mp4.update(mp4_tags)
mp4.save()
def _final_processing(
self,
cover_path: Path,
cover_url: str,
media_temp_path: Path,
final_path: Path,
tags: dict,
playlist_metadata: dict,
playlist_track: int,
):
if self.save_cover and cover_path.exists() and not self.overwrite:
logger.debug(f'Cover already exists at "{cover_path}", skipping')
elif self.save_cover and cover_url is not None:
logger.debug(f'Saving cover to "{cover_path}"')
self.save_cover_file(cover_path, cover_url)
if media_temp_path:
logger.debug("Applying tags")
if media_temp_path.suffix in (".mp4", ".m4a"):
self.apply_tags_mp4(media_temp_path, tags, cover_url)
elif media_temp_path.suffix == ".ogg":
self.apply_tags_ogg(media_temp_path, tags, cover_url)
logger.debug(f'Moving to "{final_path}"')
self.move_to_final_path(media_temp_path, final_path)
if self.save_playlist and playlist_metadata:
playlist_file_path = self.get_playlist_file_path(tags)
logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"')
self.update_playlist_file(
playlist_file_path,
final_path,
playlist_track,
)
def cleanup_temp_path(self):
if self.temp_path.exists() and not self.skip_cleanup:
logger.debug(f'Cleaning up "{self.temp_path}"')
shutil.rmtree(self.temp_path)
@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_file(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))
-275
View File
@@ -1,275 +0,0 @@
from __future__ import annotations
import subprocess
import time
from pathlib import Path
from Crypto.Cipher import AES
from yt_dlp.downloader.http import HttpFD
from yt_dlp.YoutubeDL import YoutubeDL
from .constants import (
AAC_AUDIO_QUALITIES,
AUDIO_QUALITY_X_FORMAT_ID_MAPPING,
VORBIS_AUDIO_QUALITIES,
)
from .downloader import Downloader
from .enums import AudioQuality, DownloadMode, RemuxModeAudio
from .models import StreamInfoAudio
class DownloaderAudio:
DEFAULT_EPISODE_DECRYPTION_KEY = (
b"\xde\xad\xbe\xef\xde\xad\xbe\xef\xde\xad\xbe\xef\xde\xad\xbe\xef" # lmao wtf
)
def __init__(
self,
downloader: Downloader,
audio_quality: AudioQuality = AudioQuality.AAC_MEDIUM,
download_mode: DownloadMode = DownloadMode.YTDLP,
remux_mode: RemuxModeAudio = RemuxModeAudio.FFMPEG,
):
self.downloader = downloader
self.audio_quality = audio_quality
self.download_mode = download_mode
self.remux_mode = remux_mode
def get_file_extension(
self,
) -> str:
if self.audio_quality in AAC_AUDIO_QUALITIES:
return ".m4a"
else:
return ".ogg"
def get_audio_file(
self,
audio_files: list[dict],
) -> tuple[AudioQuality, dict] | tuple[None, None]:
if self.audio_quality in AAC_AUDIO_QUALITIES:
qualities = AAC_AUDIO_QUALITIES
else:
qualities = VORBIS_AUDIO_QUALITIES
start_index = qualities.index(self.audio_quality)
for quality in qualities[start_index:]:
for audio_file in audio_files:
if audio_file["format"] == AUDIO_QUALITY_X_FORMAT_ID_MAPPING[quality]:
return quality, audio_file
return None, None
def get_stream_info(
self,
gid_metadata: dict,
product_name: dict,
media_type: str,
) -> StreamInfoAudio:
stream_info = StreamInfoAudio()
if media_type == "track":
audio_files = gid_metadata
elif media_type == "episode":
audio_files = gid_metadata
else:
raise RuntimeError()
audio_files = audio_files or gid_metadata.get("alternative")
if not audio_files:
return stream_info
if audio_files:
audio_files = audio_files
file_id = audio_files
stream_url = self.downloader.spotify_api.get_stream_urls(file_id)["cdnurl"][0]
stream_info.stream_url = stream_url
stream_info.file_id = file_id
product = product_name['data']['me']['account']['product']
if 'FREE' == product:
quality = 128000
else:
quality = 256000
stream_info.quality = quality
if self.audio_quality in AAC_AUDIO_QUALITIES:
seek_table = self.downloader.spotify_api.get_seek_table(file_id)
pssh = seek_table["pssh"]
stream_info.widevine_pssh = pssh
return stream_info
def get_decryption_key(self, stream_info: StreamInfoAudio) -> str:
decryption_key = None
for attempt in range(10):
try:
if self.audio_quality in AAC_AUDIO_QUALITIES:
_, decryption_key = self.downloader.get_widevine_decryption_key(
stream_info.widevine_pssh,
"audio",
)
else:
decryption_key = self.downloader.get_playplay_decryption_key(
stream_info.file_id
)
if decryption_key:
return decryption_key
except Exception as e:
time.sleep(attempt + 10)
return decryption_key
def download_stream_url(self, input_path: Path, stream_url: str):
if self.download_mode == DownloadMode.YTDLP:
self.download_stream_url_ytdlp(input_path, stream_url)
elif self.download_mode == DownloadMode.ARIA2C:
self.download_stream_url_aria2c(input_path, stream_url)
def download_stream_url_ytdlp(self, input_path: Path, stream_url: str) -> None:
input_path.parent.mkdir(parents=True, exist_ok=True)
with YoutubeDL(
{
"quiet": True,
"no_warnings": True,
"noprogress": self.downloader.silence,
}
) as ydl:
http_downloader = HttpFD(ydl, ydl.params)
http_downloader.download(
str(input_path),
{
"url": stream_url,
},
)
def download_stream_url_aria2c(self, input_path: Path, stream_url: str) -> None:
input_path.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(
[
self.downloader.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.downloader.subprocess_additional_args,
)
print("\r", end="")
def decrypt(
self,
decryption_key: bytes | str,
encrypted_path: Path,
decrypted_path: Path,
remuxed_path: Path,
):
if self.audio_quality in AAC_AUDIO_QUALITIES:
if self.remux_mode == RemuxModeAudio.FFMPEG:
self.decrypt_widevine_ffmpeg(
decryption_key,
encrypted_path,
decrypted_path,
)
else:
self.decrypt_widevine_mp4decrypt(
decryption_key,
encrypted_path,
decrypted_path,
)
if self.remux_mode == RemuxModeAudio.MP4BOX:
self.remux_mp4box(decrypted_path, remuxed_path)
else:
self.decrypt_playplay(
decryption_key,
encrypted_path,
decrypted_path,
)
def decrypt_playplay(
self,
decryption_key: bytes,
encrypted_path: Path,
decrypted_path: Path,
):
cipher = AES.new(
decryption_key,
AES.MODE_CTR,
nonce=bytes.fromhex("72e067fbddcbcf77"),
initial_value=bytes.fromhex("ebe8bc643f630d93"),
)
with encrypted_path.open("rb") as encrypted_file:
encrypted_data = encrypted_file.read()
with decrypted_path.open("wb") as decrypted_file:
decrypted_data = cipher.decrypt(encrypted_data)
offset = decrypted_data.find(b"OggS")
if offset == -1:
msg = "Unable to find ogg header"
raise ValueError(msg)
decrypted_file.write(decrypted_data[offset:])
def decrypt_widevine_ffmpeg(
self,
decryption_key: str,
encrypted_path: Path,
decrypted_path: Path,
):
subprocess.run(
[
self.downloader.ffmpeg_path_full,
"-loglevel",
"error",
"-y",
"-decryption_key",
decryption_key,
"-i",
encrypted_path,
"-c",
"copy",
decrypted_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
def decrypt_widevine_mp4decrypt(
self,
decryption_key: str,
encrypted_path: Path,
decrypted_path: Path,
):
subprocess.run(
[
self.downloader.mp4decrypt_path_full,
"--key",
f"1:{decryption_key}",
encrypted_path,
decrypted_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
def remux_mp4box(
self,
decrypted_path: Path,
remuxed_path: Path,
):
subprocess.run(
[
self.downloader.mp4box_path_full,
"-quiet",
"-itags",
"artist=placeholder",
"-keep-utc",
"-add",
decrypted_path,
"-new",
remuxed_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
-231
View File
@@ -1,231 +0,0 @@
from __future__ import annotations
import logging
from pathlib import Path
from .constants import AAC_AUDIO_QUALITIES, COVER_SIZE_X_KEY_MAPPING_EPISODE
from .downloader_audio import DownloaderAudio
from .models import StreamInfoAudio
logger = logging.getLogger("votify")
class DownloaderEpisode(DownloaderAudio):
def __init__(
self,
downloader_audio: DownloaderAudio,
):
self.__dict__.update(downloader_audio.__dict__)
def get_tags(
self,
episode_metadata: dict,
show_metadata: dict,
) -> dict:
try:
release_date = episode_metadata.get("release_date")
if release_date and "T" in release_date:
release_date = release_date.split("T")[0]
release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
release_date,
episode_metadata.get("release_date_precision", "day"),
)
tag_release_date = self.downloader.get_release_date_tag(release_date_datetime_obj)
tag_release_year = str(release_date_datetime_obj.year)
except Exception:
tag_release_date = str(episode_metadata.get("release_date"))
tag_release_year = tag_release_date.split("-")[0] if "-" in tag_release_date else "2024"
track_number = 1
try:
if 'trackNumber' in episode_metadata and isinstance(episode_metadata['trackNumber'], int):
track_number = episode_metadata['trackNumber']
elif show_metadata and "episodes" in show_metadata and "items" in show_metadata["episodes"]:
items = show_metadata["episodes"]["items"]
total_items = len(items)
for index, item in enumerate(items):
item_id = item.get('id')
if not item_id and 'entity' in item:
item_id = item['entity'].get('data', {}).get('id')
if item_id == episode_metadata['id']:
track_number = total_items - index
break
except (KeyError, TypeError, IndexError):
track_number = 1
tags = {
"album": show_metadata.get("name"),
"description": episode_metadata.get("description"),
"media_type": "Podcast",
"publisher": show_metadata.get("publisher"),
"rating": "Explicit" if episode_metadata.get("explicit") else "Unknown",
"release_date": tag_release_date,
"release_year": tag_release_year,
"title": episode_metadata.get("name"),
"track": track_number,
"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")
def download(
self,
*args,
**kwargs,
):
try:
self._download(*args, **kwargs)
finally:
self.downloader.cleanup_temp_path()
def _download(
self,
episode_id: str,
episode_metadata: dict = None,
show_metadata: dict = None,
gid_metadata: dict = None,
stream_info: StreamInfoAudio = None,
playlist_metadata: dict = None,
playlist_track: int = None,
product_name: dict = None,
decryption_key: bytes = None,
):
if not episode_metadata:
episode_metadata = self.downloader.spotify_api.get_episode(episode_id)
if 'data' in episode_metadata and 'trackUnion' in episode_metadata['data']:
episode_metadata = episode_metadata['data']['trackUnion']
if 'release_date' in episode_metadata:
raw_date = str(episode_metadata['release_date'])
if 'T' in raw_date:
episode_metadata['release_date'] = raw_date.split('T')[0]
precision = str(episode_metadata.get('release_date_precision', ''))
if 'T' in precision or len(precision) > 10:
episode_metadata['release_date_precision'] = 'day'
else:
if 'releaseDate' in episode_metadata and isinstance(episode_metadata['releaseDate'], dict):
r_node = episode_metadata['releaseDate']
episode_metadata['release_date'] = r_node.get('isoString', '1970-01-01').split('T')[0]
episode_metadata['release_date_precision'] = r_node.get('precision', 'day')
else:
episode_metadata['release_date'] = '1970-01-01'
episode_metadata['release_date_precision'] = 'year'
if not show_metadata or show_metadata.get('name') == 'Unknown Show':
logger.debug("Attempting to extract Show Metadata from alternative fields...")
found_show_id = None
try:
if 'podcastV2' in episode_metadata:
uri = episode_metadata['podcastV2']['data']['uri']
found_show_id = uri.split(':')[-1]
except (KeyError, IndexError, AttributeError):
pass
if not found_show_id and 'show' in episode_metadata:
found_show_id = episode_metadata['show'].get('id')
if found_show_id:
logger.debug(f"Found Show ID: {found_show_id}, fetching details...")
try:
show_metadata = self.downloader.spotify_api.get_show(found_show_id)
except Exception as e:
logger.warning(f"Failed to fetch show details: {e}")
if not show_metadata:
show_name = "Unknown Show"
try:
show_name = episode_metadata['data']['trackUnion']['artists']['items'][0]['profile']['name']
except:
pass
show_metadata = {'name': show_name, 'publisher': 'Unknown', 'episodes': {'items': []}}
if not gid_metadata:
logger.debug("Getting GID metadata")
try:
gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode", None)
except TypeError:
gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode")
if not stream_info:
logger.debug("Getting stream info")
stream_info = self.get_stream_info(gid_metadata, product_name, "episode")
if not stream_info.file_id:
logger.warning("Episode not available, skipping")
return
tags = self.get_tags(episode_metadata, show_metadata)
if playlist_metadata:
tags = {**tags, **self.downloader.get_playlist_tags(playlist_metadata, playlist_track)}
file_extension = self.get_file_extension()
final_path = self.downloader.get_final_path("episode", tags, file_extension)
cover_url = None
if 'cover_url' in episode_metadata:
cover_url = episode_metadata['cover_url']
if not cover_url and 'coverArt' in episode_metadata and 'sources' in episode_metadata['coverArt']:
try:
cover_url = episode_metadata['coverArt']['sources'][0]['url']
except (IndexError, KeyError, TypeError):
pass
if not cover_url and 'albumOfTrack' in episode_metadata:
alb = episode_metadata['albumOfTrack']
if 'coverArt' in alb and 'sources' in alb['coverArt']:
try:
cover_url = alb['coverArt']['sources'][0]['url']
except (IndexError, KeyError, TypeError):
pass
elif 'images' in alb and isinstance(alb['images'], list) and len(alb['images']) > 0:
cover_url = alb['images'][0]['url']
if not cover_url:
key_mapping = COVER_SIZE_X_KEY_MAPPING_EPISODE if 'COVER_SIZE_X_KEY_MAPPING_EPISODE' in globals() else COVER_SIZE_X_KEY_MAPPING_SONG
cover_url = self.downloader.get_cover_url(episode_metadata, key_mapping)
cover_path = self.get_cover_path(final_path)
decrypted_path = None
remuxed_path = None
if final_path.exists() and not self.downloader.overwrite:
logger.warning(f'Episode already exists at "{final_path}", skipping')
else:
if not decryption_key:
decryption_key = self.get_decryption_key(stream_info)
encrypted_path = self.downloader.get_file_temp_path(episode_id, "_encrypted", file_extension)
decrypted_path = self.downloader.get_file_temp_path(episode_id, "_decrypted", file_extension)
remuxed_path = self.downloader.get_file_temp_path(episode_id, "_remuxed", file_extension)
self.download_stream_url(encrypted_path, stream_info.stream_url)
self.decrypt(decryption_key, encrypted_path, decrypted_path, remuxed_path)
media_temp_path = remuxed_path if remuxed_path and remuxed_path.exists() else decrypted_path
self.downloader._final_processing(
cover_path,
cover_url,
media_temp_path,
final_path,
tags,
playlist_metadata,
playlist_track,
)
-153
View File
@@ -1,153 +0,0 @@
from __future__ import annotations
import logging
from .downloader_episode import DownloaderEpisode
from .downloader_video import DownloaderVideo
from .constants import COVER_SIZE_X_KEY_MAPPING_EPISODE
logger = logging.getLogger("votify")
class DownloaderEpisodeVideo(DownloaderVideo):
def __init__(
self,
downloader_video: DownloaderVideo,
downloader_episode: DownloaderEpisode,
):
self.__dict__.update(downloader_video.__dict__)
self.downloader_episode = downloader_episode
def get_video_gid(self, gid_metadata: dict) -> str | None:
if not gid_metadata:
return None
return gid_metadata
def download(
self,
*args,
**kwargs,
):
try:
self._download(*args, **kwargs)
finally:
self.downloader.cleanup_temp_path()
def _download(
self,
episode_id: str,
episode_metadata: dict = None,
show_metadata: dict = None,
gid_metadata: dict = None,
playlist_metadata: dict = None,
playlist_track: int = None,
):
if not episode_metadata:
logger.debug("Getting episode metadata")
episode_metadata = self.downloader.spotify_api.get_episode(episode_id)
if not show_metadata:
logger.debug("Getting show metadata")
show_metadata = self.downloader.spotify_api.get_show(
episode_metadata["show"]["id"]
)
if not gid_metadata:
logger.debug("Getting GID metadata")
gid_metadata = self.downloader.get_gid_metadata(episode_id, "episode")
video_gid = self.get_video_gid(gid_metadata)
if not video_gid:
logger.warning("Episode has no video, skipping")
return
stream_info = self.get_stream_info(video_gid)
if stream_info.widevine_pssh and self.downloader.cdm is None:
logger.warning(
"Podcast video has Widevine encryption, "
"but Widevine decryption is disabled, skipping"
)
return
tags = self.downloader_episode.get_tags(
episode_metadata['data']['trackUnion'],
show_metadata,
)
file_extension = self.get_file_extension(
stream_info.file_type_video,
stream_info.file_type_audio,
)
final_path = self.downloader.get_final_path(
"episode",
tags,
file_extension,
)
cover_path = self.get_cover_path(final_path)
cover_url = self.downloader.get_cover_url(
show_metadata,
COVER_SIZE_X_KEY_MAPPING_EPISODE,
)
remuxed_path = None
if final_path.exists() and not self.downloader.overwrite:
logger.warning(f'Episode already exists at "{final_path}", skipping')
return
else:
key_id, decryption_key = (
self.downloader.get_widevine_decryption_key(
stream_info.widevine_pssh,
"video",
)
if stream_info.widevine_pssh
else (None, None)
)
encrypted_path_video = self.downloader.get_file_temp_path(
episode_id,
"_video_encrypted",
file_extension,
)
encrypted_path_audio = self.downloader.get_file_temp_path(
episode_id,
"_audio_encrypted",
file_extension,
)
decrypted_path_video = self.downloader.get_file_temp_path(
episode_id,
"_video_decrypted",
file_extension,
)
decrypted_path_audio = self.downloader.get_file_temp_path(
episode_id,
"_audio_decrypted",
file_extension,
)
remuxed_path = self.downloader.get_file_temp_path(
episode_id,
"_remuxed",
file_extension,
)
temp_path_video = encrypted_path_video if key_id else decrypted_path_video
temp_path_audio = encrypted_path_audio if key_id else decrypted_path_audio
logger.debug(f'Downloading video to "{temp_path_video}"')
self.download_segments(stream_info.segment_urls_video, temp_path_video)
logger.debug(f'Downloading audio to "{temp_path_audio}"')
self.download_segments(stream_info.segment_urls_audio, temp_path_audio)
if stream_info.widevine_pssh:
logger.debug(
f'Decryping video/audio to "{decrypted_path_video}"/"{decrypted_path_audio}" '
f'and remuxing to "{remuxed_path}"'
)
else:
logger.debug(f'Remuxing to "{remuxed_path}"')
self.remux(
decrypted_path_video,
decrypted_path_audio,
remuxed_path,
key_id,
decryption_key,
encrypted_path_video,
encrypted_path_audio,
)
self.downloader._final_processing(
cover_path,
cover_url,
remuxed_path,
final_path,
tags,
playlist_metadata,
playlist_track,
)
-291
View File
@@ -1,291 +0,0 @@
from __future__ import annotations
import logging
from InquirerPy import inquirer
from InquirerPy.base.control import Choice
from .constants import COVER_SIZE_X_KEY_MAPPING_VIDEO
from .downloader_video import DownloaderVideo
logger = logging.getLogger("votify")
class DownloaderMusicVideo(DownloaderVideo):
def __init__(
self,
downloader_video: DownloaderVideo,
):
self.__dict__.update(downloader_video.__dict__)
def get_video_gid(self, gid_metadata: dict) -> str | None:
if not gid_metadata:
return None
return gid_metadata
def get_tags(
self,
track_metadata: dict,
album_metadata: dict,
track_credits: dict,
) -> dict:
track_data = track_metadata.get('data', {}).get('trackUnion', {})
if not track_data:
track_data = track_metadata
album_info = track_data.get('albumOfTrack', {})
spotify_url = track_data.get('sharingInfo', {}).get('shareUrl')
if not spotify_url and 'id' in track_data:
spotify_url = f"https://open.spotify.com/track/{track_data['id']}"
release_date_iso = album_info.get("date", {}).get("isoString")
precision_raw = album_info.get("date", {}).get("precision")
precision = precision_raw.lower() if precision_raw else "day"
release_date_str = release_date_iso.split("T")[0] if release_date_iso else None
release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
release_date_str,
precision,
)
try:
main_artists = next(
role
for role in track_credits.get("roleCredits", [])
if role["roleTitle"] == "Performers"
)["artists"]
artist_string = self.downloader.get_artist_string(main_artists)
except (StopIteration, KeyError):
artist_string = "Unknown Artist"
def get_credits_by_role(role_name):
try:
return next(
role
for role in track_credits.get("roleCredits", [])
if role["roleTitle"] == role_name
)["artists"]
except (StopIteration, KeyError):
return None
producers = get_credits_by_role("Producers")
composers = get_credits_by_role("Writers")
copyright_text = None
copyrights_list = album_info.get('copyright', {}).get('items', [])
if copyrights_list:
copyright_obj = next((i for i in copyrights_list if i.get("type") == "P"), copyrights_list[0])
copyright_text = copyright_obj.get('text')
album_name = album_info.get("name")
if not album_name and isinstance(album_metadata, dict):
album_name = album_metadata.get("name")
tags = {
"album": album_name or "Unknown Album",
"artist": artist_string,
"composer": (
self.downloader.get_artist_string(composers) if composers else None
),
"copyright": copyright_text,
"isrc": None,
"label": None,
"media_type": "Music video",
"producer": (
self.downloader.get_artist_string(producers) if producers else None
),
"rating": "Explicit" if track_data.get("contentRating", {}).get("label") == "EXPLICIT" else "Clean",
"title": track_data.get("name"),
"release_date": self.downloader.get_release_date_tag(
release_date_datetime_obj
),
"release_year": str(release_date_datetime_obj.year),
"url": spotify_url,
}
return tags
def get_music_video_id_from_song_id(
self,
track_id: str,
artist_id: str,
) -> dict | None:
now_playing_view = self.downloader.spotify_api.get_now_playing_view(
track_id, artist_id
)
related_music_videos = now_playing_view["data"]["trackUnion"]["relatedVideos"][
"items"
]
if not related_music_videos:
return
choices = [
Choice(
name="None",
value=None,
)
]
choices.extend(
[
Choice(
name=" | ".join(
[
self.downloader.get_artist_string(
[
{"name": artist["profile"]["name"]}
for artist in related_music_video["trackOfVideo"][
"data"
]["artists"]["items"]
]
),
related_music_video["trackOfVideo"]["data"]["name"],
]
),
value=related_music_video["trackOfVideo"]["data"]["uri"].split(":")[
-1
],
)
for related_music_video in related_music_videos
]
)
selected_music_video_id = inquirer.select(
message="Select which music video to download: (Artist | Title)",
choices=choices,
).execute()
return selected_music_video_id
def download(
self,
*args,
**kwargs,
):
try:
self._download(*args, **kwargs)
finally:
self.downloader.cleanup_temp_path()
def _download(
self,
music_video_id: str,
music_video_metadata: dict = None,
album_metadata: dict = None,
gid_metadata: dict = None,
playlist_metadata: dict = None,
playlist_track: int = None,
):
if not music_video_metadata:
logger.debug("Getting music video metadata")
music_video_metadata = self.downloader.spotify_api.get_track(music_video_id)
if not album_metadata:
logger.debug("Getting album metadata")
album_metadata = self.downloader.spotify_api.get_album(
music_video_metadata['data']['trackUnion']['albumOfTrack']["id"]
)
if not gid_metadata:
logger.debug("Getting GID metadata")
gid_metadata = self.downloader.get_gid_metadata(music_video_id, "track")
video_gid = self.get_video_gid(gid_metadata)
if not video_gid:
logger.debug("Getting equivalent music video ID from song ID")
music_video_id = self.get_music_video_id_from_song_id(
music_video_id,
album_metadata["artists"][0]["id"],
)
if not music_video_id:
logger.warning(
"No related music videos found or no music video selected, skipping"
)
return
music_video_metadata = self.downloader.spotify_api.get_track(music_video_id)
logger.warning(
f'Switching to downloading music video "{music_video_metadata["name"]}"'
)
album_metadata = self.downloader.spotify_api.get_album(
music_video_metadata["album"]["id"]
)
gid_metadata = self.downloader.get_gid_metadata(music_video_id, "track")
video_gid = self.get_video_gid(gid_metadata)
stream_info = self.get_stream_info(video_gid)
logger.debug("Getting credits")
track_credits = self.downloader.spotify_api.get_track_credits(music_video_id)
tags = self.get_tags(
music_video_metadata,
album_metadata,
track_credits,
)
file_extension = self.get_file_extension(
stream_info.file_type_video,
stream_info.file_type_audio,
)
final_path = self.downloader.get_final_path(
"music-video",
tags,
file_extension,
)
cover_path = self.get_cover_path(final_path)
cover_url = self.downloader.get_cover_url(
album_metadata,
COVER_SIZE_X_KEY_MAPPING_VIDEO,
)
remuxed_path = None
if final_path.exists() and not self.downloader.overwrite:
logger.warning(f'Music video already exists at "{final_path}", skipping')
return
else:
key_id, decryption_key = self.downloader.get_widevine_decryption_key(
stream_info.widevine_pssh,
"video",
)
encrypted_path_video = self.downloader.get_file_temp_path(
music_video_id,
"_video_encrypted",
file_extension,
)
encrypted_path_audio = self.downloader.get_file_temp_path(
music_video_id,
"_audio_encrypted",
file_extension,
)
decrypted_path_video = self.downloader.get_file_temp_path(
music_video_id,
"_video_decrypted",
file_extension,
)
decrypted_path_audio = self.downloader.get_file_temp_path(
music_video_id,
"_audio_decrypted",
file_extension,
)
remuxed_path = self.downloader.get_file_temp_path(
music_video_id,
"_remuxed",
file_extension,
)
logger.debug(f'Downloading video to "{encrypted_path_video}"')
self.download_segments(stream_info.segment_urls_video, encrypted_path_video)
logger.debug(f'Downloading audio to "{encrypted_path_audio}"')
self.download_segments(stream_info.segment_urls_audio, encrypted_path_audio)
logger.debug(
f'Decryping video/audio to "{decrypted_path_video}"/"{decrypted_path_audio}" '
f'and remuxing to "{remuxed_path}"'
)
self.remux(
decrypted_path_video,
decrypted_path_audio,
remuxed_path,
key_id,
decryption_key,
encrypted_path_video,
encrypted_path_audio,
)
self.downloader._final_processing(
cover_path,
cover_url,
remuxed_path,
final_path,
tags,
playlist_metadata,
playlist_track,
)
-367
View File
@@ -1,367 +0,0 @@
from __future__ import annotations
import datetime
import logging
from pathlib import Path
from .constants import COVER_SIZE_X_KEY_MAPPING_SONG
from .downloader_audio import DownloaderAudio
from .models import Lyrics, StreamInfoAudio
logger = logging.getLogger("votify")
class DownloaderSong(DownloaderAudio):
def __init__(
self,
downloader_audio: DownloaderAudio,
lrc_only: bool = False,
no_lrc: bool = False,
):
self.__dict__.update(downloader_audio.__dict__)
self.lrc_only = lrc_only
self.no_lrc = no_lrc
def get_tags(
self,
track_metadata: dict,
album_metadata: list,
track_credits: dict,
lyrics_unsynced: str,
) -> dict:
track_info = track_metadata.get("data", {}).get("trackUnion", {})
album_info = track_info.get("albumOfTrack", {})
external_url = track_info.get("sharingInfo", {}).get("shareUrl")
isrc = None
release_date_iso = album_info.get("date", {}).get("isoString")
precision_raw = album_info.get("date", {}).get("precision")
precision = precision_raw.lower() if precision_raw else "day"
release_date_str = release_date_iso.split("T")[0] if release_date_iso else None
release_date_datetime_obj = self.downloader.get_release_date_datetime_obj(
release_date_str,
precision,
)
track_artists = []
raw_artist_items = track_info.get("firstArtist", {}).get("items", [])
other_artist_items = track_info.get("otherArtists", {}).get("items", [])
all_artist_items = raw_artist_items + other_artist_items
for item in all_artist_items:
name = item.get("profile", {}).get("name")
if name:
track_artists.append({"name": name})
if not track_artists:
alt_artists = track_info.get("artists", {}).get("items", [])
for item in alt_artists:
name = item.get("profile", {}).get("name") or item.get("name")
if name:
track_artists.append({"name": name})
if not track_artists:
track_artists = [{"name": "Unknown Artist"}]
album_artists = track_artists
def get_credits_artists(role_title):
try:
role = next(
r for r in track_credits.get("roleCredits", [])
if r.get("roleTitle") == role_title
)
return role.get("artists", [])
except StopIteration:
return []
producers = get_credits_artists("Producers")
composers = get_credits_artists("Writers")
current_uri = track_info.get("uri")
current_track_entry = next(
(item.get("track", {}) for item in album_metadata if item.get("track", {}).get("uri") == current_uri),
{}
)
disc_number = current_track_entry.get("discNumber", 1)
try:
disc_total = max(
(item.get("track", {}).get("discNumber", 1) for item in album_metadata),
default=1
)
except Exception:
disc_total = 1
track_total = album_info.get("tracks", {}).get("totalCount")
if not track_total:
track_total = len(album_metadata)
copyright_text = next(
(c["text"] for c in album_info.get("copyright", {}).get("items", []) if c.get("type") == "P"),
None
)
label = album_info.get("courtesyLine") or album_info.get("label")
tags = {
"album": album_info.get("name"),
"album_artist": self.downloader.get_artist_string(album_artists),
"artist": self.downloader.get_artist_string(track_artists),
"compilation": False,
"composer": (
self.downloader.get_artist_string(composers) if composers else None
),
"copyright": copyright_text,
"disc": int(disc_number),
"disc_total": int(disc_total),
"isrc": isrc,
"label": label,
"lyrics": lyrics_unsynced,
"media_type": "Song",
"producer": (
self.downloader.get_artist_string(producers) if producers else None
),
"rating": "Explicit" if track_info.get("contentRating", {}).get("label") == "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_info.get("name"),
"track": int(track_info.get("trackNumber", 1)),
"track_total": int(track_total),
"url": external_url,
}
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"
def download(
self,
*args,
**kwargs,
):
try:
self._download(*args, **kwargs)
finally:
self.downloader.cleanup_temp_path()
def _download(
self,
track_id: str,
track_metadata: dict = None,
album_metadata: dict = None,
gid_metadata: dict = None,
stream_info: StreamInfoAudio = None,
playlist_metadata: dict = None,
product_name: dict = None,
playlist_track: int = None,
decryption_key: bytes = None,
):
if not album_metadata:
logger.debug("Getting track metadata")
track_metadata = self.downloader.spotify_api.get_track(track_id)
try:
current_date = track_metadata['data']['trackUnion']['albumOfTrack'].get('release_date')
if current_date == '1970-01-01':
fresh_data = self.downloader.spotify_api.get_track(track_id)
real_date = None
real_precision = 'day'
real_album_name = None
if 'album' in fresh_data:
real_date = fresh_data['album']['release_date']
real_precision = fresh_data['album'].get('release_date_precision', 'day')
real_album_name = fresh_data['album']['name']
elif 'data' in fresh_data:
src_album = fresh_data['data']['trackUnion']['albumOfTrack']
if 'date' in src_album and 'isoString' in src_album['date']:
real_date = src_album['date']['isoString']
real_precision = 'day'
if real_date and not str(real_date).startswith('1970'):
track_metadata['data']['trackUnion']['albumOfTrack']['release_date'] = real_date
track_metadata['data']['trackUnion']['albumOfTrack']['release_date_precision'] = real_precision
if not album_metadata:
album_metadata = {}
album_metadata['release_date'] = real_date
album_metadata['release_date_precision'] = real_precision
if real_album_name:
album_metadata['name'] = real_album_name
except Exception as e:
pass
if not album_metadata:
logger.debug("Getting album metadata")
try:
alb_node = track_metadata['data']['trackUnion']['albumOfTrack']
album_id = alb_node['id']
album_metadata = self.downloader.spotify_api.get_album(album_id)
except (KeyError, TypeError):
pass
if not gid_metadata:
logger.debug("Getting GID metadata")
try:
gid_metadata = self.downloader.get_gid_metadata(track_id, "track", product_name)
except TypeError:
gid_metadata = self.downloader.get_gid_metadata(track_id, "track")
if gid_metadata.get("original_video"):
logger.warning("Track is a music video, skipping.")
return
if not stream_info:
logger.debug("Getting stream info")
stream_info = self.get_stream_info(gid_metadata, product_name, "track")
if not stream_info.file_id:
logger.warning(
"Track is not available on Spotify's "
"servers and no alternative found, skipping"
)
return
if stream_info.quality != self.audio_quality:
if stream_info.quality == 128000:
info = '128 kb/s'
else:
info = '256 kb/s'
logger.info(f"The current bit rate is {info}")
if gid_metadata:
logger.debug("Getting lyrics")
lyrics = self.get_lyrics(track_id)
else:
lyrics = Lyrics()
logger.debug("Getting track credits")
track_credits = self.downloader.spotify_api.get_track_credits(track_id)
tags = self.get_tags(
track_metadata,
album_metadata,
track_credits,
lyrics.unsynced,
)
if playlist_metadata:
tags = {
**tags,
**self.downloader.get_playlist_tags(
playlist_metadata,
playlist_track,
),
}
file_extension = self.get_file_extension()
final_path = self.downloader.get_final_path(
"track",
tags,
file_extension,
)
lrc_path = self.downloader.get_lrc_path(final_path)
cover_path = self.get_cover_path(final_path)
cover_url = self.downloader.get_cover_url(
track_metadata,
COVER_SIZE_X_KEY_MAPPING_SONG,
)
decrypted_path = None
remuxed_path = None
if self.lrc_only:
pass
elif final_path.exists() and not self.downloader.overwrite:
logger.warning(f'Track already exists at "{final_path}", skipping')
else:
if not decryption_key:
logger.debug("Getting decryption key")
decryption_key = self.get_decryption_key(stream_info)
encrypted_path = self.downloader.get_file_temp_path(
track_id,
"_encrypted",
file_extension,
)
decrypted_path = self.downloader.get_file_temp_path(
track_id,
"_decrypted",
file_extension,
)
remuxed_path = self.downloader.get_file_temp_path(
track_id,
"_remuxed",
file_extension,
)
logger.debug(f'Downloading to "{encrypted_path}"')
self.download_stream_url(encrypted_path, stream_info.stream_url)
logger.debug(
f'Decrypting to "{decrypted_path}" and remuxing to "{remuxed_path}"'
)
self.decrypt(
decryption_key,
encrypted_path,
decrypted_path,
remuxed_path,
)
if self.no_lrc:
pass
elif lrc_path.exists() and not self.downloader.overwrite:
logger.debug(f'Synced lyrics already exists at "{lrc_path}", skipping')
else:
logger.debug(f'Saving synced lyrics to "{lrc_path}"')
lyrics = None
media_temp_path = (
remuxed_path
if remuxed_path is not None and remuxed_path.exists()
else decrypted_path
)
self.downloader._final_processing(
cover_path,
cover_url,
media_temp_path,
final_path,
tags,
playlist_metadata,
playlist_track,
)
-393
View File
@@ -1,393 +0,0 @@
from __future__ import annotations
import subprocess
from pathlib import Path
from InquirerPy import inquirer
from InquirerPy.base.control import Choice
from yt_dlp.downloader.fragment import FragmentFD
from yt_dlp.YoutubeDL import YoutubeDL
from .downloader import Downloader
from .enums import RemuxModeAudio, RemuxModeVideo, VideoFormat
from .models import StreamInfoVideo
class DownloaderVideo:
def __init__(
self,
downloader: Downloader,
video_format: VideoFormat = VideoFormat.MP4,
remux_mode: RemuxModeVideo = RemuxModeVideo.FFMPEG,
):
self.downloader = downloader
self.video_format = video_format
self.remux_mode = remux_mode
self._adjust_remux_mode()
def _adjust_remux_mode(self):
if self.video_format == VideoFormat.WEBM:
self.remux_mode = RemuxModeVideo.FFMPEG
def get_cover_path(self, final_path: Path) -> Path:
return final_path.with_suffix(".jpg")
def get_file_extension(
self,
file_type_video: str,
file_type_audio: str,
) -> str:
if file_type_video == file_type_audio:
return "." + file_type_video
else:
return ".mp4"
def get_encryption_info(
self,
encryption_infos: list[dict],
key_system: str,
) -> tuple[int, str]:
for index, encryption_info in enumerate(encryption_infos):
if encryption_info["key_system"] == key_system:
return index, encryption_info
def get_stream_info(
self,
gid: str,
) -> StreamInfoVideo:
stream_info = StreamInfoVideo()
manifest = self.downloader.spotify_api.get_video_manifest(gid)
base_url = manifest["base_urls"][0]
initialization_template = manifest["initialization_template"]
segment_template = manifest["segment_template"]
end_time_millis = manifest["end_time_millis"]
segment_length = manifest["contents"][0]["segment_length"]
encryption_index = None
if manifest["contents"][0].get("encryption_infos"):
encryption_index, encryption_info = self.get_encryption_info(
manifest["contents"][0]["encryption_infos"],
"widevine",
)
stream_info.widevine_pssh = encryption_info["encryption_data"]
video_profiles = list(
filter(
lambda x: x["mime_type"].startswith("video")
and encryption_index
in (x.get("encryption_indices", [encryption_index])),
manifest["contents"][0]["profiles"],
)
)
audio_profiles = list(
filter(
lambda x: x["mime_type"].startswith("audio")
and encryption_index
in (x.get("encryption_indices", [encryption_index])),
manifest["contents"][0]["profiles"],
)
)
if not video_profiles or not audio_profiles:
return stream_info
if self.video_format == VideoFormat.ASK:
profile_video, profile_audio = (
self.get_video_profile_from_user(
video_profiles,
),
self.get_audio_profile_from_user(
audio_profiles,
),
)
else:
profile_video, profile_audio = (
self.get_best_profile_by_bitrate(
video_profiles,
f"video/{self.video_format.value}",
),
self.get_best_profile_by_bitrate(
audio_profiles,
f"audio/{self.video_format.value}",
),
)
profile_id_video = profile_video["id"]
file_type_video = profile_video["file_type"]
profile_id_audio = profile_audio["id"]
file_type_audio = profile_audio["file_type"]
stream_info.segment_urls_video = self.get_segment_urls(
base_url,
initialization_template,
segment_template,
end_time_millis,
segment_length,
profile_id_video,
file_type_video,
)
stream_info.segment_urls_audio = self.get_segment_urls(
base_url,
initialization_template,
segment_template,
end_time_millis,
segment_length,
profile_id_audio,
file_type_audio,
)
stream_info.file_type_video = file_type_video
stream_info.file_type_audio = file_type_audio
return stream_info
def get_best_profile_by_bitrate(
self,
profiles: list[dict],
mime_type: str,
) -> str:
profiles_filtered = list(
filter(
lambda x: x["mime_type"] == mime_type,
profiles,
)
)
bitrate_key = f"{mime_type.split('/')[0]}_bitrate"
best_profile = max(profiles_filtered, key=lambda x: x[bitrate_key])
return best_profile
def get_video_profile_from_user(
self,
video_profiles: list[dict],
) -> str:
choices = [
Choice(
name=" | ".join(
[
profile["video_codec"],
f"{profile['video_width']}x{profile['video_height']}",
str(profile["video_bitrate"]),
]
),
value=profile,
)
for profile in video_profiles
]
selected = inquirer.select(
message="Select which video codec to download: (Codec | Resolution | Bitrate)",
choices=choices,
).execute()
return selected
def get_audio_profile_from_user(
self,
audio_profiles: list[dict],
) -> str:
choices = [
Choice(
name=" | ".join(
[
profile["audio_codec"],
str(profile["audio_bitrate"]),
]
),
value=profile,
)
for profile in audio_profiles
]
selected = inquirer.select(
message="Select which audio codec to download: (Codec | Bitrate)",
choices=choices,
).execute()
return selected
def get_segment_urls(
self,
base_url: str,
initialization_template: str,
segment_template: str,
end_time_millis: int,
segment_length: int,
profile_id: int,
file_type: str,
) -> list[str]:
initialization_template_formatted = initialization_template.replace(
"{{profile_id}}", str(profile_id)
).replace("{{file_type}}", file_type)
segment_urls = []
first_segment = base_url + initialization_template_formatted
segment_urls.append(first_segment)
for i in range(0, int(end_time_millis / 1000) + 5, segment_length):
segment_template_url_formatted = (
segment_template.replace("{{profile_id}}", str(profile_id))
.replace("{{segment_timestamp}}", str(i))
.replace("{{file_type}}", file_type)
)
segment_urls.append(base_url + segment_template_url_formatted)
return segment_urls
def download_segments(
self,
segment_urls: list[str],
input_path: Path,
):
input_path.parent.mkdir(parents=True, exist_ok=True)
segments_dict = [
{"url": url, "frag_index": idx + 1} for idx, url in enumerate(segment_urls)
]
ctx = {
"filename": str(input_path),
"total_frags": len(segments_dict),
}
info_dict = {}
with YoutubeDL(
{
"quiet": True,
"no_warnings": True,
"noprogress": self.downloader.silence,
},
) as ydl:
try:
fragment_downloader = FragmentFD(ydl, ydl.params)
fragment_downloader._prepare_and_start_frag_download(
ctx,
info_dict,
)
fragment_downloader.download_and_append_fragments(
ctx,
segments_dict,
info_dict,
)
finally:
fragment_downloader._finish_multiline_status()
def remux(
self,
decrypted_path_video: Path,
decrypted_path_audio: Path,
remuxed_path: Path,
key_id: str | None = None,
decryption_key: str | None = None,
encrypted_path_video: Path | None = None,
encrypted_path_audio: Path | None = None,
):
if decryption_key:
if remuxed_path.suffix == ".webm":
self.decrypt_packager(
key_id,
decryption_key,
encrypted_path_video,
decrypted_path_video,
)
self.decrypt_packager(
key_id,
decryption_key,
encrypted_path_audio,
decrypted_path_audio,
)
else:
self.decrypt_mp4decrypt(
decryption_key,
encrypted_path_video,
decrypted_path_video,
)
self.decrypt_mp4decrypt(
decryption_key,
encrypted_path_audio,
decrypted_path_audio,
)
if self.remux_mode == RemuxModeAudio.MP4BOX:
self.remux_mp4box(
decrypted_path_video,
decrypted_path_audio,
remuxed_path,
)
else:
self.remux_ffmpeg(
decrypted_path_video,
decrypted_path_audio,
remuxed_path,
)
def decrypt_packager(
self,
key_id: str,
decryption_key: str,
encrypted_path: Path,
decrypted_path: Path,
):
subprocess.run(
[
self.downloader.packager_path_full,
"--quiet",
f"stream=0,in={encrypted_path},output={decrypted_path}",
"-enable_raw_key_decryption",
"-keys",
f"key_id={key_id}:key={decryption_key}",
],
check=True,
**self.downloader.subprocess_additional_args,
)
def decrypt_mp4decrypt(
self,
decryption_key: str,
encrypted_path: Path,
decrypted_path: Path,
):
subprocess.run(
[
self.downloader.mp4decrypt_path_full,
"--key",
f"1:{decryption_key}",
encrypted_path,
decrypted_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
def remux_ffmpeg(
self,
input_path_video: Path,
input_path_audio: Path,
output_path: Path,
):
output_path.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(
[
self.downloader.ffmpeg_path_full,
"-loglevel",
"error",
"-y",
"-i",
input_path_video,
"-i",
input_path_audio,
"-c",
"copy",
"-map",
"0:v:0",
"-map",
"1:a:0",
output_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
def remux_mp4box(
self,
decrypted_path_video: Path,
decrypted_path_audio: Path,
remuxed_path: Path,
):
subprocess.run(
[
self.downloader.mp4box_path_full,
"-quiet",
"-itags",
"artist=placeholder",
"-keep-utc",
"-add",
decrypted_path_video,
"-add",
decrypted_path_audio,
"-new",
remuxed_path,
],
check=True,
**self.downloader.subprocess_additional_args,
)
-38
View File
@@ -1,38 +0,0 @@
from enum import Enum
class AudioQuality(Enum):
VORBIS_HIGH = "vorbis-high"
VORBIS_MEDIUM = "vorbis-medium"
VORBIS_LOW = "vorbis-low"
AAC_MEDIUM = "aac-medium"
AAC_HIGH = "aac-high"
class VideoFormat(Enum):
MP4 = "mp4"
WEBM = "webm"
ASK = "ask"
class DownloadMode(Enum):
YTDLP = "ytdlp"
ARIA2C = "aria2c"
class RemuxModeAudio(Enum):
FFMPEG = "ffmpeg"
MP4BOX = "mp4box"
MP4DECRYPT = "mp4decrypt"
class RemuxModeVideo(Enum):
FFMPEG = "ffmpeg"
MP4BOX = "mp4box"
class CoverSize(Enum):
SMALL = "small"
MEDIUM = "medium"
LARGE = "large"
EXTRA_LARGE = "extra-large"
-42
View File
@@ -1,42 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from .enums import AudioQuality
@dataclass
class Lyrics:
synced: str = None
unsynced: str = None
@dataclass
class UrlInfo:
type: str = None
id: str = None
@dataclass
class DownloadQueueItem:
playlist_metadata: dict = None
album_metadata: dict = None
show_metadata: dict = None
media_metadata: dict = None
@dataclass
class StreamInfoAudio:
stream_url: str = None
file_id: str = None
widevine_pssh: str = None
quality: AudioQuality = None
@dataclass
class StreamInfoVideo:
segment_urls_video: list[str] = None
segment_urls_audio: list[str] = None
file_type_video: str = None
file_type_audio: str = None
widevine_pssh: str = None