add podacast video support and general refactoring

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(

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",

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:

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,
+        )

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,
+        )

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,
+        )

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,
+        )

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"

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

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(