mirror of
https://github.com/glomatico/votify.git
synced 2026-09-29 20:42:32 +02:00
Add video interfaces and episode support
This commit is contained in:
Rafael Moraes
@@ -4,3 +4,5 @@ from .enums import *
|
||||
from .exceptions import *
|
||||
from .interface import *
|
||||
from .types import *
|
||||
from .video import *
|
||||
from .episode_video import *
|
||||
|
||||
@@ -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(
|
||||
(
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user