diff --git a/votify/interface/audio.py b/votify/interface/audio.py index 9953d0f..f897e16 100644 --- a/votify/interface/audio.py +++ b/votify/interface/audio.py @@ -2,13 +2,24 @@ import asyncio import logging from pywidevine.license_protocol_pb2 import WidevinePsshData +from unplayplay.consts import EMULATOR_SIZES, PLAYPLAY_TOKEN from ..api.enums import SessionType +from ..api.proto.audio_files_extension_pb2 import AudioFilesExtensionResponse +from ..api.proto.extendedmetadata_pb2 import ( + BatchedEntityRequest, + BatchedEntityRequestHeader, + EntityRequest, + ExtensionKind, + ExtensionQuery, +) +from ..api.proto.playplay_pb2 import ContentType, Interactivity, PlayPlayLicenseRequest from .base import SpotifyBaseInterface from .enums import AudioQuality from .exceptions import ( VotifyMediaFormatNotAvailableException, VotifyMediaFormatNotAvailableForSessionTypeException, + VotifyNoKeyEmuException, ) from .types import DecryptionKey, StreamInfo, StreamInfoAv @@ -50,37 +61,65 @@ class SpotifyAudioInterface(SpotifyBaseInterface): media_type: str, skip_pssh: bool, ) -> StreamInfoAv: + session_type_skipped = False for audio_quality in self.audio_quality_priority: + stream_info = None + if audio_quality.mp4: - if self.api.session_type in {SessionType.LIBRESPOT, SessionType.WEB}: - stream_info = await self._get_stream_info_web( - media_id=media_id, - media_type=media_type, - skip_pssh=skip_pssh, - audio_quality=audio_quality, - ) - else: - raise VotifyMediaFormatNotAvailableForSessionTypeException( - media_id=media_id, - session_type=self.api.session_type, - ) + if self.api.session_type not in { + SessionType.LIBRESPOT, + SessionType.WEB, + }: + session_type_skipped = True + continue + session_type_skipped = False + stream_info = await self._get_stream_info_web( + media_id=media_id, + media_type=media_type, + skip_pssh=skip_pssh, + audio_quality=audio_quality, + ) elif audio_quality.ogg: if self.api.session_type == SessionType.LIBRESPOT: + session_type_skipped = False stream_info = await self._get_stream_info_librespot( media_id=media_id, media_type=media_type, audio_quality=audio_quality, ) - else: - raise VotifyMediaFormatNotAvailableForSessionTypeException( + elif self.api.session_type == SessionType.DESKTOP: + session_type_skipped = False + stream_info = await self._get_stream_info_desktop( media_id=media_id, - session_type=self.api.session_type, + media_type=media_type, + audio_quality=audio_quality, ) + else: + session_type_skipped = True + continue + + elif audio_quality.flac: + if self.api.session_type == SessionType.DESKTOP: + session_type_skipped = False + stream_info = await self._get_stream_info_desktop( + media_id=media_id, + media_type=media_type, + audio_quality=audio_quality, + ) + else: + session_type_skipped = True + continue if stream_info: return stream_info + if session_type_skipped: + raise VotifyMediaFormatNotAvailableForSessionTypeException( + media_id=media_id, + session_type=self.api.session_type, + ) + raise VotifyMediaFormatNotAvailableException( media_id=media_id, ) @@ -188,6 +227,70 @@ class SpotifyAudioInterface(SpotifyBaseInterface): return stream_info + async def _get_stream_info_desktop( + self, + media_id: str, + media_type: str, + audio_quality: AudioQuality, + ) -> StreamInfoAv | None: + extended_metadata_request = BatchedEntityRequest( + header=BatchedEntityRequestHeader(), + entity_request=[ + EntityRequest( + entity_uri=f"spotify:{media_type}:{media_id}", + query=[ + ExtensionQuery(extension_kind=ExtensionKind.TRACK_V4), + ExtensionQuery(extension_kind=ExtensionKind.AUDIO_FILES), + ], + ), + ], + ) + extended_metadata = await self.api.get_extended_metadata( + extended_metadata_request + ) + audio_files_data_array = next( + ( + ext + for ext in extended_metadata.extended_metadata + if ext.extension_kind == ExtensionKind.AUDIO_FILES + ), + None, + ) + if not audio_files_data_array: + return + + audio_files_data_bytes = audio_files_data_array.extension_data[ + 0 + ].extension_data.value + audio_files = AudioFilesExtensionResponse() + audio_files.ParseFromString(audio_files_data_bytes) + + audio_file_info = next( + ( + (extended_audio_file, extended_audio_file.average_bitrate) + for extended_audio_file in audio_files.files + if extended_audio_file.file.format == int(audio_quality.format_id) + ), + None, + ) + if not audio_file_info: + return + + stream_url = await self._get_stream_url( + audio_quality.format_id, + audio_file_info[0].file.file_id.hex(), + ) + + return StreamInfoAv( + audio_track=StreamInfo( + stream_url=stream_url, + widevine_pssh=None, + file_id=audio_file_info[0].file.file_id, + file_format=audio_quality.file_format, + actual_file_format=audio_quality.actual_file_format, + ), + ) + async def get_widevine_decryption_key(self, pssh: str) -> DecryptionKey: return await self._get_widevine_decryption_key(pssh, "audio") @@ -213,6 +316,28 @@ class SpotifyAudioInterface(SpotifyBaseInterface): return decryption_key + async def _get_desktop_decryption_key(self, file_id: bytes) -> DecryptionKey: + if not self.key_emu: + raise VotifyNoKeyEmuException() + + request = PlayPlayLicenseRequest( + version=5, + token=PLAYPLAY_TOKEN.VALUE, + interactivity=Interactivity.INTERACTIVE, + content_type=ContentType.AUDIO_TRACK, + ) + license = await self.api.get_playplay_license( + file_id=file_id.hex(), + request=request, + ) + decryption_key = self.key_emu.get_aes_key( + obfuscated_key=license.obfuscated_key, + content_id=file_id[: EMULATOR_SIZES.CONTENT_ID], + ) + return DecryptionKey( + decryption_key=decryption_key, + ) + async def get_decryption_key( self, stream_info: StreamInfoAv, @@ -233,6 +358,13 @@ class SpotifyAudioInterface(SpotifyBaseInterface): media_id=media_id, file_id=stream_info.audio_track.file_id, ) + elif ( + stream_info.audio_track.file_id + and self.api.session_type == SessionType.DESKTOP + ): + return await self._get_desktop_decryption_key( + file_id=stream_info.audio_track.file_id, + ) else: raise Exception("No method available to get decryption key") diff --git a/votify/interface/exceptions.py b/votify/interface/exceptions.py index adc6e68..b070f5b 100644 --- a/votify/interface/exceptions.py +++ b/votify/interface/exceptions.py @@ -13,6 +13,13 @@ class VotifyNoCdmException(VotifyInterfaceException): super().__init__("Content requires a CDM but no .wvd file was provided") +class VotifyNoKeyEmuException(VotifyInterfaceException): + def __init__(self): + super().__init__( + "Content requires decryption but no Spotify DLL file was provided" + ) + + class VotifyUrlParseException(VotifyInterfaceException): def __init__(self, url: str): super().__init__(f"Failed to parse Spotify URL: {url}")