From f860553b36a67fa96761a84dcd57cc98e33e8a66 Mon Sep 17 00:00:00 2001 From: Rafael Moraes <50295204+glomatico@users.noreply.github.com> Date: Tue, 24 Feb 2026 18:30:40 -0300 Subject: [PATCH] Add video interfaces and episode support --- votify/interface/__init__.py | 2 + votify/interface/episode.py | 10 +- votify/interface/episode_video.py | 61 +++++++ votify/interface/interface.py | 7 +- votify/interface/video.py | 271 ++++++++++++++++++++++++++++++ 5 files changed, 346 insertions(+), 5 deletions(-) create mode 100644 votify/interface/episode_video.py create mode 100644 votify/interface/video.py diff --git a/votify/interface/__init__.py b/votify/interface/__init__.py index 2bb1916..d62ab06 100644 --- a/votify/interface/__init__.py +++ b/votify/interface/__init__.py @@ -4,3 +4,5 @@ from .enums import * from .exceptions import * from .interface import * from .types import * +from .video import * +from .episode_video import * diff --git a/votify/interface/episode.py b/votify/interface/episode.py index c6795ba..dd80068 100644 --- a/votify/interface/episode.py +++ b/votify/interface/episode.py @@ -64,17 +64,21 @@ class SpotifyEpisodeInterface(SpotifyAudioInterface): return media + @staticmethod async def parse_tags( - self, episode_data: dict, show_items: list[dict], ) -> MediaTags: tags = MediaTags( album=episode_data["podcastV2"]["data"]["name"], - date=self.parse_date(episode_data["releaseDate"]["isoString"]), + date=SpotifyAudioInterface.parse_date( + episode_data["releaseDate"]["isoString"] + ), description=episode_data.get("description"), media_type=MediaType.PODCAST, - rating=self.parse_rating(episode_data["contentRating"]["label"]), + rating=SpotifyAudioInterface.parse_rating( + episode_data["contentRating"]["label"] + ), title=episode_data["name"], track=next( ( diff --git a/votify/interface/episode_video.py b/votify/interface/episode_video.py new file mode 100644 index 0000000..2518f87 --- /dev/null +++ b/votify/interface/episode_video.py @@ -0,0 +1,61 @@ +import logging +from .types import DecryptionKey, DecryptionKeyAv, MediaTags, SpotifyMedia +from .episode import SpotifyEpisodeInterface +from .exceptions import VotifyDrmDisabledException + +from .video import SpotifyVideoInterface + +logger = logging.getLogger(__name__) + + +class SpotifyEpisodeVideoInterface(SpotifyVideoInterface): + def __init__( + self, + video: SpotifyVideoInterface, + ): + self.__dict__.update(video.__dict__) + + async def proccess_media( + self, + playback_info: dict, + episode_data: dict | None = None, + show_data: dict | None = None, + show_items: list[dict] | None = None, + ) -> SpotifyMedia: + if not episode_data: + episode_response = await self.api.get_episode( + playback_info["metadata"]["uri"].split(":")[-1] + ) + episode_data = episode_response["data"]["episodeUnionV2"] + + if not show_data: + show_data, show_items = await self.get_show_data_cached( + episode_data["podcastV2"]["data"]["uri"].split(":")[-1] + ) + + media = SpotifyMedia(episode_data["uri"].split(":")[-1]) + + media.media_metadata = episode_data + media.show_metadata = show_data + + media.tags = await SpotifyEpisodeInterface.parse_tags( + episode_data, + show_items, + ) + + media.cover_url = self.parse_cover_url( + episode_data["coverArt"]["sources"][0]["url"] + ) + + media.stream_info = await self.get_stream_info(playback_info) + + if media.stream_info.audio_track.widevine_pssh: + if self.no_drm: + raise VotifyDrmDisabledException(media.media_id, episode_data) + media.decryption_key = await self.get_widevine_decryption_key( + media.stream_info.audio_track.widevine_pssh + ) + + logger.debug(f"Parsed episode video media: {media}") + + return media diff --git a/votify/interface/interface.py b/votify/interface/interface.py index 706d00b..fd15a57 100644 --- a/votify/interface/interface.py +++ b/votify/interface/interface.py @@ -10,6 +10,7 @@ from .exceptions import ( VotifyUnsupportedMediaTypeException, ) from .song import SpotifySongInterface +from .episode_video import SpotifyEpisodeVideoInterface from .types import SpotifyMedia logger = logging.getLogger(__name__) @@ -21,10 +22,12 @@ class SpotifyInterface: base: SpotifyAudioInterface, song: SpotifySongInterface, episode: SpotifyEpisodeInterface, + episode_video: SpotifyEpisodeVideoInterface, ) -> None: self.base = base self.song = song self.episode = episode + self.episode_video = episode_video async def _get_track_media( self, @@ -53,7 +56,7 @@ class SpotifyInterface: assert playback_info, "Playback info should be available for playable track" if self.base.is_video(playback_info): - return await self.base.music_video._proccess_media( + return await self.music_video._proccess_media( playback_info=playback_info, track_data=( track_data @@ -93,7 +96,7 @@ class SpotifyInterface: assert playback_info, "Playback info should be available for playable episode" if self.base.is_video(playback_info): - return await self.base.episode_video._proccess_media( + return await self.episode_video.proccess_media( playback_info=playback_info, episode_data=episode_data, show_data=show_data, diff --git a/votify/interface/video.py b/votify/interface/video.py new file mode 100644 index 0000000..8b70180 --- /dev/null +++ b/votify/interface/video.py @@ -0,0 +1,271 @@ +import logging + +from InquirerPy import inquirer +from InquirerPy.base.control import Choice + +from .base import SpotifyBaseInterface +from .constants import COVER_SIZE_ID_MAP_VIDEO +from .enums import VideoFormat, VideoResolution +from .types import StreamInfo, StreamInfoAv + +logger = logging.getLogger(__name__) + + +class SpotifyVideoInterface(SpotifyBaseInterface): + def __init__( + self, + base: SpotifyBaseInterface, + video_format: VideoFormat = VideoFormat.MP4, + resolution: VideoResolution = VideoResolution.R1080P, + ): + self.__dict__.update(base.__dict__) + self.video_format = video_format + self.resolution = resolution + + def _get_encryption_info( + self, + encryption_infos: list[dict], + key_system: str, + ) -> tuple[int, dict] | None: + for index, encryption_info in enumerate(encryption_infos): + if encryption_info["key_system"] == key_system: + return index, encryption_info + return None + + def _filter_profiles_by_type( + self, + profiles: list[dict], + mime_type_prefix: str, + encryption_index: int, + ) -> list[dict]: + return list( + filter( + lambda x: x["mime_type"].startswith(mime_type_prefix) + and encryption_index + in (x.get("encryption_indices", [encryption_index])), + profiles, + ) + ) + + def _get_best_profile( + self, + profiles: list[dict], + mime_type: str, + ) -> dict: + profiles_filtered = list( + filter(lambda x: x["mime_type"] == mime_type, profiles) + ) + bitrate_key = f"{mime_type.split('/')[0]}_bitrate" + return max(profiles_filtered, key=lambda x: x.get(bitrate_key, 0)) + + def _get_best_video_profile_by_resolution( + self, + profiles: list[dict], + mime_type: str, + ) -> dict: + profiles_filtered = list( + filter(lambda x: x["mime_type"] == mime_type, profiles) + ) + + available_resolutions = sorted( + set(p.get("video_height") for p in profiles_filtered), + reverse=True, + ) + + for res in available_resolutions: + if res <= int(self.resolution): + return max( + (p for p in profiles_filtered if p.get("video_height") == res), + key=lambda x: x.get("video_bitrate", 0), + ) + + return min(profiles_filtered, key=lambda x: x.get("video_height", 0)) + + async def _select_profiles_interactive( + self, + video_profiles: list[dict], + audio_profiles: list[dict], + ) -> tuple[dict, dict]: + profile_video = await self._get_video_profile_from_user(video_profiles) + profile_audio = await self._get_audio_profile_from_user(audio_profiles) + return profile_video, profile_audio + + async def _get_video_profile_from_user( + self, + video_profiles: list[dict], + ) -> dict: + 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 + ] + return await inquirer.select( + message="Select which video codec to download: (Codec | Resolution | Bitrate)", + choices=choices, + ).execute_async() + + async def _get_audio_profile_from_user( + self, + audio_profiles: list[dict], + ) -> dict: + choices = [ + Choice( + name=" | ".join( + [ + profile["audio_codec"], + str(profile["audio_bitrate"]), + ] + ), + value=profile, + ) + for profile in audio_profiles + ] + return await inquirer.select( + message="Select which audio codec to download: (Codec | Bitrate)", + choices=choices, + ).execute_async() + + def _generate_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_url = initialization_template.replace( + "{{profile_id}}", str(profile_id) + ).replace("{{file_type}}", file_type) + segment_urls = [base_url + initialization_url] + + for timestamp in range(0, int(end_time_millis / 1000) + 5, segment_length): + segment_url = ( + segment_template.replace("{{profile_id}}", str(profile_id)) + .replace("{{segment_timestamp}}", str(timestamp)) + .replace("{{file_type}}", file_type) + ) + segment_urls.append(base_url + segment_url) + + return segment_urls + + def _parse_file_id( + self, + playback_info: dict, + ) -> str | None: + manifest_ids_video = playback_info.get("manifest", {}).get("manifest_ids_video") + if manifest_ids_video: + return manifest_ids_video[0]["file_id"] + + return None + + async def get_stream_info( + self, + playback_info: dict, + ) -> StreamInfoAv: + file_id = self._parse_file_id(playback_info) + assert file_id, "File ID should be available in playback info for video content" + + manifest = await self.api.get_video_manifest(file_id) + content = manifest["contents"][0] + + base_url = manifest["base_urls"][0] + initialization_template = manifest["initialization_template"] + segment_template = manifest["segment_template"] + end_time_millis = content["end_time_millis"] + segment_length = content["segment_length"] + + encryption_result = None + widevine_pssh = None + if content.get("encryption_infos"): + encryption_result = self._get_encryption_info( + content["encryption_infos"], "widevine" + ) + if encryption_result: + encryption_index, encryption_info = encryption_result + widevine_pssh = encryption_info.get("encryption_data") + else: + encryption_index = None + else: + encryption_index = None + + video_profiles = self._filter_profiles_by_type( + content["profiles"], + "video", + encryption_index, + ) + audio_profiles = self._filter_profiles_by_type( + content["profiles"], + "audio", + encryption_index, + ) + + if self.video_format == VideoFormat.ASK: + profile_video, profile_audio = await self._select_profiles_interactive( + video_profiles, + audio_profiles, + ) + else: + profile_video = self._get_best_video_profile_by_resolution( + video_profiles, + f"video/{self.video_format.value}", + ) + profile_audio = self._get_best_profile( + audio_profiles, + f"audio/{self.video_format.value}", + ) + + video_urls = self._generate_segment_urls( + base_url, + initialization_template, + segment_template, + end_time_millis, + segment_length, + profile_video["id"], + profile_video["file_type"], + ) + + audio_urls = self._generate_segment_urls( + base_url, + initialization_template, + segment_template, + end_time_millis, + segment_length, + profile_audio["id"], + profile_audio["file_type"], + ) + + stream_info = StreamInfoAv( + audio_track=StreamInfo( + stream_url=audio_urls, + widevine_pssh=widevine_pssh, + ), + video_track=StreamInfo( + stream_url=video_urls, + widevine_pssh=widevine_pssh, + ), + file_format="mp4", + ) + + logger.debug(f"Generated stream info: {stream_info}") + + return stream_info + + def parse_cover_url(self, base_cover_url: str) -> str: + cover_url = self._transform_cover_url(base_cover_url, COVER_SIZE_ID_MAP_VIDEO) + + logger.debug(f"Parsed episode cover URL: {cover_url}") + + return cover_url + + async def get_widevine_decryption_key(self, pssh: str): + return await self._get_widevine_decryption_key(pssh, "video")