From f8fc7f954dc0a56a7ebb8472cdb82637328447d5 Mon Sep 17 00:00:00 2001 From: Rafael Moraes <50295204+glomatico@users.noreply.github.com> Date: Sun, 22 Sep 2024 05:31:39 -0300 Subject: [PATCH] add podacast video support and general refactoring --- votify/cli.py | 59 ++++++- votify/constants.py | 19 ++- votify/downloader.py | 119 ++++++++++++-- votify/downloader_episode.py | 43 ++--- votify/downloader_episode_video.py | 108 ++++++++++++ votify/downloader_song.py | 44 ++--- votify/downloader_video.py | 253 +++++++++++++++++++++++++++++ votify/enums.py | 6 + votify/models.py | 9 + votify/spotify_api.py | 10 ++ 10 files changed, 592 insertions(+), 78 deletions(-) create mode 100644 votify/downloader_episode_video.py create mode 100644 votify/downloader_video.py diff --git a/votify/cli.py b/votify/cli.py index 8ab9794..f44ff24 100644 --- a/votify/cli.py +++ b/votify/cli.py @@ -3,7 +3,6 @@ from __future__ import annotations import inspect import json import logging -import time from enum import Enum from pathlib import Path @@ -13,8 +12,10 @@ from . import __version__ from .constants import EXCLUDED_CONFIG_FILE_PARAMS, X_NOT_FOUND_STRING from .downloader import Downloader from .downloader_episode import DownloaderEpisode +from .downloader_episode_video import DownloaderEpisodeVideo from .downloader_song import DownloaderSong -from .enums import DownloadMode, AudioQuality +from .downloader_video import DownloaderVideo +from .enums import AudioQuality, DownloadMode, VideoFormat from .spotify_api import SpotifyApi logger = logging.getLogger("votify") @@ -80,6 +81,11 @@ def load_config_file( default=10, help="Wait interval between downloads in seconds.", ) +@click.option( + "--download-podcast-video", + is_flag=True, + help="Attempt to download the video version of podcasts.", +) @click.option( "--force-premium", "-f", @@ -124,6 +130,12 @@ def load_config_file( default=downloader_sig.parameters["audio_quality"].default, help="Audio quality for songs and podcasts.", ) +@click.option( + "--video-format", + type=VideoFormat, + default=downloader_sig.parameters["video_format"].default, + help="Video format.", +) @click.option( "--output-path", "-o", @@ -156,6 +168,12 @@ def load_config_file( default=downloader_sig.parameters["unplayplay_path"].default, help="Path to unplayplay binary.", ) +@click.option( + "--ffmpeg-path", + type=str, + default=downloader_sig.parameters["ffmpeg_path"].default, + help="Path to ffmpeg binary.", +) @click.option( "--template-folder-album", type=str, @@ -254,6 +272,7 @@ def load_config_file( def main( urls: list[str], wait_interval: float, + download_podcast_video: bool, force_premium: bool, read_urls_as_txt: bool, config_path: Path, @@ -261,11 +280,13 @@ def main( print_exceptions: bool, cookies_path: Path, audio_quality: AudioQuality, + video_format: VideoFormat, output_path: Path, temp_path: Path, download_mode: DownloadMode, aria2c_path: str, unplayplay_path: str, + ffmpeg_path: str, template_folder_album: str, template_folder_compilation: str, template_file_single_disc: str, @@ -293,11 +314,13 @@ def main( downloader = Downloader( spotify_api, audio_quality, + video_format, output_path, temp_path, download_mode, aria2c_path, unplayplay_path, + ffmpeg_path, template_folder_album, template_folder_compilation, template_file_single_disc, @@ -320,6 +343,12 @@ def main( downloader_episode = DownloaderEpisode( downloader, ) + downloader_video = DownloaderVideo( + downloader, + ) + downloader_episode_video = DownloaderEpisodeVideo( + downloader_episode, + ) if not lrc_only: if not downloader.unplayplay_path_full: logger.critical(X_NOT_FOUND_STRING.format("Unplayplay", unplayplay_path)) @@ -327,6 +356,9 @@ def main( if download_mode == DownloadMode.ARIA2C and not downloader.aria2c_path_full: logger.critical(X_NOT_FOUND_STRING.format("aria2c", aria2c_path)) return + if download_podcast_video and not downloader.ffmpeg_path_full: + logger.critical(X_NOT_FOUND_STRING.format("ffmpeg", ffmpeg_path)) + return spotify_api.config_info["isPremium"] = ( True if force_premium else spotify_api.config_info["isPremium"] ) @@ -382,13 +414,22 @@ def main( playlist_track=index, ) elif media_type == "episode" and not lrc_only: - downloader_episode.download( - episode_id=media_id, - episode_metadata=media_metadata, - show_metadata=download_queue_item.show_metadata, - playlist_metadata=download_queue_item.playlist_metadata, - playlist_track=index, - ) + if download_podcast_video: + downloader_episode_video.download( + episode_id=media_id, + episode_metadata=media_metadata, + show_metadata=download_queue_item.show_metadata, + playlist_metadata=download_queue_item.playlist_metadata, + playlist_track=index, + ) + else: + downloader_episode.download( + episode_id=media_id, + episode_metadata=media_metadata, + show_metadata=download_queue_item.show_metadata, + playlist_metadata=download_queue_item.playlist_metadata, + playlist_track=index, + ) except Exception as e: error_count += 1 logger.error( diff --git a/votify/constants.py b/votify/constants.py index bc4aae4..60e7c00 100644 --- a/votify/constants.py +++ b/votify/constants.py @@ -1,6 +1,6 @@ from __future__ import annotations -from .enums import AudioQuality +from .enums import AudioQuality, VideoFormat EXCLUDED_CONFIG_FILE_PARAMS = ( "urls", @@ -32,7 +32,22 @@ VORBIS_TAGS_MAPPING = { "url": "URL", } -QUALITY_X_FORMAT_ID_MAPPING = { +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", diff --git a/votify/downloader.py b/votify/downloader.py index 2dae4d2..b0e5f3d 100644 --- a/votify/downloader.py +++ b/votify/downloader.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 import datetime import functools +import logging import re import shutil import subprocess @@ -14,12 +15,17 @@ from Crypto.Cipher import AES 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 yt_dlp import YoutubeDL -from .constants import QUALITY_X_FORMAT_ID_MAPPING, VORBIS_TAGS_MAPPING -from .enums import AudioQuality, DownloadMode +from .constants import ( + AUDIO_QUALITY_X_FORMAT_ID_MAPPING, + MP4_TAGS_MAP, + VORBIS_TAGS_MAPPING, +) +from .enums import AudioQuality, DownloadMode, VideoFormat from .models import DownloadQueueItem, StreamInfoAudio, UrlInfo from .playplay_pb2 import ( AUDIO_TRACK, @@ -30,6 +36,8 @@ from .playplay_pb2 import ( from .spotify_api import SpotifyApi from .utils import check_response +logger = logging.getLogger("votify") + class Downloader: ILLEGAL_CHARACTERS_REGEX = r'[\\/:*?"<>|;]' @@ -45,11 +53,13 @@ class Downloader: self, spotify_api: SpotifyApi, audio_quality: AudioQuality = AudioQuality.VORBIS_MEDIUM, + video_format: VideoFormat = VideoFormat.MP4, output_path: Path = Path("./Spotify"), temp_path: Path = Path("./temp"), download_mode: DownloadMode = DownloadMode.YTDLP, aria2c_path: Path = "aria2c", unplayplay_path: Path = "unplayplay", + ffmpeg_path: Path = "ffmpeg", template_folder_album: str = "{album_artist}/{album}", template_folder_compilation: str = "Compilations/{album}", template_file_single_disc: str = "{track:02d} {title}", @@ -67,11 +77,13 @@ class Downloader: ): self.spotify_api = spotify_api self.audio_quality = audio_quality + self.video_format = video_format self.output_path = output_path self.temp_path = temp_path self.download_mode = download_mode self.aria2c_path = aria2c_path self.unplayplay_path = unplayplay_path + self.ffmpeg_path = ffmpeg_path self.template_folder_album = template_folder_album self.template_folder_compilation = template_folder_compilation self.template_file_single_disc = template_file_single_disc @@ -94,6 +106,7 @@ class Downloader: def _set_binaries_full_path(self): self.aria2c_path_full = shutil.which(self.aria2c_path) self.unplayplay_path_full = shutil.which(self.unplayplay_path) + self.ffmpeg_path_full = shutil.which(self.ffmpeg_path) def _set_exclude_tags_list(self): self.exclude_tags_list = ( @@ -338,7 +351,7 @@ class Downloader: start_index = qualities.index(self.audio_quality) for quality in qualities[start_index:]: for audio_file in audio_files: - if audio_file["format"] == QUALITY_X_FORMAT_ID_MAPPING[quality]: + if audio_file["format"] == AUDIO_QUALITY_X_FORMAT_ID_MAPPING[quality]: return quality, audio_file return None, None @@ -503,19 +516,15 @@ class Downloader: def _get_cover_url(self, images_dict: list[dict]) -> str: return max(images_dict, key=lambda img: img["height"])["url"] - def get_encrypted_path( + def get_file_temp_path( self, track_id: str, + extra_string: str, + extension: str, ) -> Path: - return self.temp_path / (f"{track_id}_encrypted.ogg") + return self.temp_path / (track_id + extra_string + extension) - def get_decrypted_path( - self, - track_id: str, - ) -> Path: - return self.temp_path / (f"{track_id}_decrypted.ogg") - - def apply_tags( + def apply_tags_ogg( self, input_path: Path, tags: dict, @@ -544,6 +553,92 @@ class Downloader: 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 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 == ".mp4": + 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(): + logger.debug(f'Cleaning up "{self.temp_path}"') + self.cleanup_temp_path() + @staticmethod @functools.lru_cache() def get_response_bytes(url: str) -> bytes: diff --git a/votify/downloader_episode.py b/votify/downloader_episode.py index 0fd2d56..9b379c4 100644 --- a/votify/downloader_episode.py +++ b/votify/downloader_episode.py @@ -58,9 +58,7 @@ class DownloaderEpisode: try: self._download(*args, **kwargs) finally: - if self.downloader.temp_path.exists(): - logger.debug(f'Cleaning up "{self.downloader.temp_path}"') - self.downloader.cleanup_temp_path() + self.downloader._cleanup_temp_path() def _download( self, @@ -121,8 +119,12 @@ class DownloaderEpisode: if not decryption_key: logger.debug("Getting decryption key") decryption_key = self.downloader.get_decryption_key(stream_info.file_id) - encrypted_path = self.downloader.get_encrypted_path(episode_id) - decrypted_path = self.downloader.get_decrypted_path(episode_id) + encrypted_path = self.downloader.get_file_temp_path( + episode_id, "_encrypted", ".ogg" + ) + decrypted_path = self.downloader.get_file_temp_path( + episode_id, "_decrypted", ".ogg" + ) logger.debug(f'Downloading to "{encrypted_path}"') self.downloader.download_stream_url(encrypted_path, stream_info.stream_url) logger.debug(f'Decrypting to "{decrypted_path}"') @@ -131,25 +133,12 @@ class DownloaderEpisode: encrypted_path, decrypted_path, ) - if ( - self.downloader.save_cover - and cover_path.exists() - and not self.downloader.overwrite - ): - logger.debug(f'Cover already exists at "{cover_path}", skipping') - elif self.downloader.save_cover and cover_url is not None: - logger.debug(f'Saving cover to "{cover_path}"') - self.downloader.save_cover_file(cover_path, cover_url) - if decrypted_path: - logger.debug("Applying tags") - self.downloader.apply_tags(decrypted_path, tags, cover_url) - logger.debug(f'Moving to "{final_path}"') - self.downloader.move_to_final_path(decrypted_path, final_path) - if self.downloader.save_playlist and playlist_metadata: - playlist_file_path = self.downloader.get_playlist_file_path(tags) - logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"') - self.downloader.update_playlist_file( - playlist_file_path, - final_path, - playlist_track, - ) + self.downloader._final_processing( + cover_path, + cover_url, + decrypted_path, + final_path, + tags, + playlist_metadata, + playlist_track, + ) diff --git a/votify/downloader_episode_video.py b/votify/downloader_episode_video.py new file mode 100644 index 0000000..c9587a6 --- /dev/null +++ b/votify/downloader_episode_video.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import logging + +from .downloader_episode import DownloaderEpisode +from .downloader_video import DownloaderVideo + +logger = logging.getLogger("votify") + + +class DownloaderEpisodeVideo(DownloaderVideo): + def __init__( + self, + downloader_episode: DownloaderEpisode, + ): + super().__init__(downloader_episode.downloader) + self.downloader_episode = downloader_episode + + def get_video_gid(self, gid_metadata: dict) -> str | None: + if not gid_metadata.get("video"): + return None + return gid_metadata["video"][0]["file_id"] + + 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) + tags = self.downloader_episode.get_tags( + episode_metadata, + 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.downloader_episode.get_cover_path(final_path) + cover_url = self.downloader_episode.downloader.get_cover_url(show_metadata) + remuxed_path = None + if final_path.exists() and not self.downloader.overwrite: + logger.warning(f'Episode already exists at "{final_path}", skipping') + return + else: + temp_path_video = self.downloader.get_file_temp_path( + episode_id, + "_video", + file_extension, + ) + temp_path_audio = self.downloader.get_file_temp_path( + episode_id, + "_audio", + file_extension, + ) + remuxed_path = self.downloader.get_file_temp_path( + episode_id, + "_remuxed", + file_extension, + ) + 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) + logger.debug(f'Remuxing to "{remuxed_path}"') + self.remux(temp_path_video, temp_path_audio, remuxed_path) + 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 index a0dcd10..1108566 100644 --- a/votify/downloader_song.py +++ b/votify/downloader_song.py @@ -135,9 +135,7 @@ class DownloaderSong: try: self._download(*args, **kwargs) finally: - if self.downloader.temp_path.exists(): - logger.debug(f'Cleaning up "{self.downloader.temp_path}"') - self.downloader.cleanup_temp_path() + self.downloader._cleanup_temp_path() def _download( self, @@ -210,8 +208,12 @@ class DownloaderSong: if not decryption_key: logger.debug("Getting decryption key") decryption_key = self.downloader.get_decryption_key(stream_info.file_id) - encrypted_path = self.downloader.get_encrypted_path(track_id) - decrypted_path = self.downloader.get_decrypted_path(track_id) + encrypted_path = self.downloader.get_file_temp_path( + track_id, "_encrypted", ".ogg" + ) + decrypted_path = self.downloader.get_file_temp_path( + track_id, "_decrypted", ".ogg" + ) logger.debug(f'Downloading to "{encrypted_path}"') self.downloader.download_stream_url(encrypted_path, stream_info.stream_url) logger.debug(f'Decrypting to "{decrypted_path}"') @@ -227,26 +229,12 @@ class DownloaderSong: else: logger.debug(f'Saving synced lyrics to "{lrc_path}"') self.downloader.save_lrc(lrc_path, lyrics.synced) - if ( - self.downloader.save_cover - and cover_path.exists() - and not self.downloader.overwrite - ): - logger.debug(f'Cover already exists at "{cover_path}", skipping') - return - elif self.downloader.save_cover and cover_url is not None: - logger.debug(f'Saving cover to "{cover_path}"') - self.downloader.save_cover_file(cover_path, cover_url) - if decrypted_path: - logger.debug("Applying tags") - self.downloader.apply_tags(decrypted_path, tags, cover_url) - logger.debug(f'Moving to "{final_path}"') - self.downloader.move_to_final_path(decrypted_path, final_path) - if not self.lrc_only and self.downloader.save_playlist and playlist_metadata: - playlist_file_path = self.downloader.get_playlist_file_path(tags) - logger.debug(f'Updating M3U8 playlist from "{playlist_file_path}"') - self.downloader.update_playlist_file( - playlist_file_path, - final_path, - playlist_track, - ) + self.downloader._final_processing( + cover_path, + cover_url, + decrypted_path, + final_path, + tags, + playlist_metadata, + playlist_track, + ) diff --git a/votify/downloader_video.py b/votify/downloader_video.py new file mode 100644 index 0000000..15e8663 --- /dev/null +++ b/votify/downloader_video.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + +import requests +import tqdm +from InquirerPy import inquirer +from InquirerPy.base.control import Choice + +from .downloader import Downloader +from .enums import VideoFormat +from .models import StreamInfoVideo +from .utils import check_response + + +class DownloaderVideo: + def __init__( + self, + downloader: Downloader, + ): + self.downloader = downloader + + 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.encryption_data_widevine = encryption_info["encryption_data"] + video_profiles = list( + profile + for profile in manifest["contents"][0]["profiles"] + if profile["mime_type"].startswith("video") + and encryption_index + in (profile.get("encryption_indices", [encryption_index])) + ) + audio_profiles = list( + profile + for profile in manifest["contents"][0]["profiles"] + if profile["mime_type"].startswith("audio") + and encryption_index + in (profile.get("encryption_indices", [encryption_index])) + ) + if not video_profiles or not audio_profiles: + return stream_info + if self.downloader.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( + manifest["contents"][0]["profiles"], + f"video/{self.downloader.video_format.value}", + encryption_info, + ), + self.get_best_profile_by_bitrate( + manifest["contents"][0]["profiles"], + f"audio/{self.downloader.video_format.value}", + encryption_index, + ), + ) + 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: + bitrate_key = f"{mime_type.split('/')[0]}_bitrate" + best_profile = max(profiles, 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), 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_segment( + self, + segment_url: str, + input_path: Path, + ): + response = requests.get(segment_url, stream=True) + check_response(response) + chunk_size = 1024 + input_path.parent.mkdir(parents=True, exist_ok=True) + with input_path.open("ab") as file: + for chunk in response.iter_content(chunk_size): + if chunk: + file.write(chunk) + + def download_segments( + self, + segment_urls: list[str], + input_path: Path, + ): + for segment_url in tqdm.tqdm(segment_urls, leave=False): + self.download_segment(segment_url, input_path) + + 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 remux( + 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, + ) diff --git a/votify/enums.py b/votify/enums.py index 5ff7d91..36902c2 100644 --- a/votify/enums.py +++ b/votify/enums.py @@ -7,6 +7,12 @@ class AudioQuality(Enum): VORBIS_LOW = "low" +class VideoFormat(Enum): + MP4 = "mp4" + WEBM = "webm" + ASK = "ask" + + class DownloadMode(Enum): YTDLP = "ytdlp" ARIA2C = "aria2c" diff --git a/votify/models.py b/votify/models.py index c58f773..8f31ab9 100644 --- a/votify/models.py +++ b/votify/models.py @@ -30,3 +30,12 @@ class StreamInfoAudio: stream_url: str = None file_id: 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 + encryption_data_widevine: str = None diff --git a/votify/spotify_api.py b/votify/spotify_api.py index 5a075c6..b376f35 100644 --- a/votify/spotify_api.py +++ b/votify/spotify_api.py @@ -20,6 +20,7 @@ class SpotifyApi: LYRICS_API_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/{track_id}" METADATA_API_URL = "https://api.spotify.com/v1/{type}/{item_id}" GID_METADATA_API_URL = "https://spclient.wg.spotify.com/metadata/4/{media_type}/{gid}?market=from_token" + VIDEO_MANIFEST_API_URL = "https://gue1-spclient.spotify.com/manifests/v7/json/sources/{gid}/options/supports_drm" PLAYPLAY_LICENSE_API_URL = ( "https://gew4-spclient.spotify.com/playplay/v1/key/{file_id}" ) @@ -253,6 +254,15 @@ class SpotifyApi: ) return artist_albums + def get_video_manifest( + self, + gid: str, + ) -> dict: + self._refresh_session_auth() + response = self.session.get(self.VIDEO_MANIFEST_API_URL.format(gid=gid)) + check_response(response) + return response.json() + def get_playplay_license(self, file_id: str, challenge: bytes) -> bytes: self._refresh_session_auth() response = self.session.post(