Add desktop audio stream & decryption support

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

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}")