From e8baeb107f185d2f7f550fc09d470d7946b96a5f Mon Sep 17 00:00:00 2001 From: Rafael Moraes <50295204+glomatico@users.noreply.github.com> Date: Thu, 26 Feb 2026 13:00:01 -0300 Subject: [PATCH] Remove legacy files --- votify/__main__.py | 3 - votify/cli.py | 799 ------------------------ votify/config_file.py | 101 --- votify/constants.py | 94 --- votify/custom_logger_formatter.py | 24 - votify/device_flow.py | 119 ---- votify/downloader.py | 951 ----------------------------- votify/downloader_audio.py | 275 --------- votify/downloader_episode.py | 231 ------- votify/downloader_episode_video.py | 153 ----- votify/downloader_music_video.py | 291 --------- votify/downloader_song.py | 367 ----------- votify/downloader_video.py | 393 ------------ votify/enums.py | 38 -- votify/models.py | 42 -- 15 files changed, 3881 deletions(-) delete mode 100644 votify/__main__.py delete mode 100644 votify/cli.py delete mode 100644 votify/config_file.py delete mode 100644 votify/constants.py delete mode 100644 votify/custom_logger_formatter.py delete mode 100644 votify/device_flow.py delete mode 100644 votify/downloader.py delete mode 100644 votify/downloader_audio.py delete mode 100644 votify/downloader_episode.py delete mode 100644 votify/downloader_episode_video.py delete mode 100644 votify/downloader_music_video.py delete mode 100644 votify/downloader_song.py delete mode 100644 votify/downloader_video.py delete mode 100644 votify/enums.py delete mode 100644 votify/models.py diff --git a/votify/__main__.py b/votify/__main__.py deleted file mode 100644 index 4e28416..0000000 --- a/votify/__main__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .cli import main - -main() diff --git a/votify/cli.py b/votify/cli.py deleted file mode 100644 index 0d4fd50..0000000 --- a/votify/cli.py +++ /dev/null @@ -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))") diff --git a/votify/config_file.py b/votify/config_file.py deleted file mode 100644 index f98fc7b..0000000 --- a/votify/config_file.py +++ /dev/null @@ -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 diff --git a/votify/constants.py b/votify/constants.py deleted file mode 100644 index 43d1d9f..0000000 --- a/votify/constants.py +++ /dev/null @@ -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", -} diff --git a/votify/custom_logger_formatter.py b/votify/custom_logger_formatter.py deleted file mode 100644 index bfcd95c..0000000 --- a/votify/custom_logger_formatter.py +++ /dev/null @@ -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) diff --git a/votify/device_flow.py b/votify/device_flow.py deleted file mode 100644 index e90556f..0000000 --- a/votify/device_flow.py +++ /dev/null @@ -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'' - ) - 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() diff --git a/votify/downloader.py b/votify/downloader.py deleted file mode 100644 index 7349977..0000000 --- a/votify/downloader.py +++ /dev/null @@ -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)) diff --git a/votify/downloader_audio.py b/votify/downloader_audio.py deleted file mode 100644 index 420c517..0000000 --- a/votify/downloader_audio.py +++ /dev/null @@ -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, - ) diff --git a/votify/downloader_episode.py b/votify/downloader_episode.py deleted file mode 100644 index eef6c37..0000000 --- a/votify/downloader_episode.py +++ /dev/null @@ -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, - ) diff --git a/votify/downloader_episode_video.py b/votify/downloader_episode_video.py deleted file mode 100644 index 1708abb..0000000 --- a/votify/downloader_episode_video.py +++ /dev/null @@ -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, - ) diff --git a/votify/downloader_music_video.py b/votify/downloader_music_video.py deleted file mode 100644 index cebadbb..0000000 --- a/votify/downloader_music_video.py +++ /dev/null @@ -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, - ) diff --git a/votify/downloader_song.py b/votify/downloader_song.py deleted file mode 100644 index 918a6c8..0000000 --- a/votify/downloader_song.py +++ /dev/null @@ -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, - ) diff --git a/votify/downloader_video.py b/votify/downloader_video.py deleted file mode 100644 index 2332bf7..0000000 --- a/votify/downloader_video.py +++ /dev/null @@ -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, - ) diff --git a/votify/enums.py b/votify/enums.py deleted file mode 100644 index 089099b..0000000 --- a/votify/enums.py +++ /dev/null @@ -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" diff --git a/votify/models.py b/votify/models.py deleted file mode 100644 index 5acbdb5..0000000 --- a/votify/models.py +++ /dev/null @@ -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