music-assistant-server

1"""
2Streaming, decryption, and playback callbacks for the Deezer provider.
3
4Handles stream URL resolution, Blowfish decryption for track audio,
5radio/podcast stream details, and listen logging callbacks.
6"""
7
8from __future__ import annotations
9
10import hashlib
11import uuid
12from collections.abc import AsyncGenerator
13from datetime import datetime
14from math import ceil
15from typing import TYPE_CHECKING, NoReturn
16
17from aiohttp import ClientTimeout
18from Crypto.Cipher import Blowfish
19from music_assistant_models.enums import ContentType, MediaType, StreamType
20from music_assistant_models.errors import AudioError, MediaNotFoundError
21from music_assistant_models.media_items import AudioFormat, MediaItemType
22from music_assistant_models.streamdetails import StreamDetails
23
24from music_assistant.helpers.app_vars import app_var
25from music_assistant.helpers.datetime import utc_timestamp
26
27from .gw_client import DeezerGWError
28from .helpers import fetch_all_audiobook_chapter_edges, fetch_all_bookmarks
29
30if TYPE_CHECKING:
31    from .provider import DeezerProvider
32
33
34class DeezerStreamingManager:
35    """Handles streaming, decryption, and playback lifecycle for Deezer."""
36
37    def __init__(self, provider: DeezerProvider) -> None:
38        """Initialize streaming manager."""
39        self.provider = provider
40        self.mass = provider.mass
41        self.instance_id = provider.instance_id
42        self.domain = provider.domain
43        self.logger = provider.logger
44
45    # -- Resume position --
46
47    async def get_resume_position(
48        self, item_id: str, media_type: MediaType
49    ) -> tuple[bool, int, datetime | None]:
50        """
51        Get the resume position for a podcast episode.
52
53        :param item_id: The provider-specific episode ID.
54        :param media_type: The media type (only PODCAST_EPISODE is supported).
55        :returns: Tuple of (fully_played, resume_position_ms, timestamp).
56        """
57        if media_type != MediaType.PODCAST_EPISODE:
58            return (False, 0, None)
59        bookmarks = await fetch_all_bookmarks(self.provider.gql_client)
60        if item_id in bookmarks:
61            is_played, position_ms = bookmarks[item_id]
62            return (is_played, position_ms, None)
63        return (False, 0, None)
64
65    # -- Playback callbacks --
66
67    async def on_played(
68        self,
69        media_type: MediaType,
70        prov_item_id: str,
71        fully_played: bool,
72        position: int,
73        media_item: MediaItemType,
74        is_playing: bool = False,
75    ) -> None:
76        """
77        Handle callback when a podcast episode has been played or is playing.
78
79        Syncs playback progress back to Deezer's bookmark/play-state system.
80        Only handles podcast episodes — Deezer's Pipe API has no track listen logging.
81        """
82        if media_type != MediaType.PODCAST_EPISODE:
83            return
84        if fully_played:
85            await self.provider.gql_client.mark_as_played_podcast_episode(episode_id=prov_item_id)
86        elif position == 0 and not is_playing:
87            await self.provider.gql_client.mark_as_not_played_podcast_episode(
88                episode_id=prov_item_id
89            )
90        elif is_playing or position > 0:
91            await self.provider.gql_client.bookmark_podcast_episode(
92                episode_id=prov_item_id, offset=position
93            )
94
95    # -- Stream details --
96
97    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
98        """Return the content details for the given track when it will be streamed."""
99        if media_type == MediaType.RADIO:
100            return await self._get_radio_stream_details(item_id)
101        if media_type == MediaType.PODCAST_EPISODE:
102            return await self._get_podcast_episode_stream_details(item_id)
103        if media_type == MediaType.AUDIOBOOK:
104            return await self._get_audiobook_stream_details(item_id)
105        return await self._get_track_stream_details(item_id)
106
107    async def _get_track_stream_details(self, item_id: str) -> StreamDetails:
108        """Return stream details for a regular Deezer track."""
109        try:
110            url_details, song_data = await self.provider.gw_client.get_deezer_track_urls(item_id)
111        except DeezerGWError as err:
112            _raise_stream_error(err, item_id, "Track")
113        url = url_details["sources"][0]["url"]
114        size_key = f"FILESIZE_{url_details['format']}"
115        size = int(song_data.get(size_key) or song_data.get("FILESIZE_MP3_MISC") or 0)
116        return StreamDetails(
117            item_id=item_id,
118            provider=self.instance_id,
119            audio_format=AudioFormat(
120                content_type=ContentType.try_parse(url_details["format"].split("_")[0])
121            ),
122            stream_type=StreamType.CUSTOM,
123            duration=int(song_data["DURATION"]),
124            data={
125                "url": url,
126                "format": url_details["format"],
127                "track_id": str(song_data["SNG_ID"]),
128            },
129            size=size,
130            can_seek=True,
131            allow_seek=True,
132        )
133
134    async def _get_audiobook_stream_details(self, item_id: str) -> StreamDetails:
135        """
136        Return stream details for a Deezer audiobook.
137
138        Resolves all chapter IDs and durations. Each chapter is streamed
139        as a regular encrypted track via get_audio_stream.
140        """
141        all_edges = await fetch_all_audiobook_chapter_edges(self.provider.gql_client, item_id)
142
143        chapter_ids: list[str] = []
144        chapter_durations_ms: list[int] = []
145        for edge in all_edges:
146            if edge.node is None:
147                continue
148            chapter_ids.append(edge.node.id)
149            chapter_durations_ms.append(edge.node.duration * 1000)
150
151        if not chapter_ids:
152            raise MediaNotFoundError(f"No chapters found for audiobook {item_id}")
153
154        # Probe the first chapter to determine audio format
155        try:
156            first_url_details, _ = await self.provider.gw_client.get_deezer_track_urls(
157                chapter_ids[0]
158            )
159        except DeezerGWError as err:
160            _raise_stream_error(err, item_id, "Audiobook")
161        total_duration = sum(chapter_durations_ms) // 1000
162
163        return StreamDetails(
164            item_id=item_id,
165            provider=self.instance_id,
166            media_type=MediaType.AUDIOBOOK,
167            audio_format=AudioFormat(
168                content_type=ContentType.try_parse(first_url_details["format"].split("_")[0])
169            ),
170            stream_type=StreamType.CUSTOM,
171            duration=total_duration,
172            data={
173                "chapter_ids": chapter_ids,
174                "chapter_durations_ms": chapter_durations_ms,
175            },
176            can_seek=True,
177            allow_seek=True,
178        )
179
180    async def _get_radio_stream_details(self, item_id: str) -> StreamDetails:
181        """Return stream details for a Deezer livestream (radio station)."""
182        result = await self.provider.gql_client.get_livestream(livestream_id=item_id)
183        if result is None or not result.media:
184            raise MediaNotFoundError(f"Radio {item_id} has no stream URL")
185        # Prefer HLS, then AAC, then MP3
186        best_media = result.media[0]
187        for media in result.media:
188            if media.codec and media.codec.type_ == "hls":
189                best_media = media
190                break
191        content_type = ContentType.UNKNOWN
192        if best_media.codec:
193            content_type = ContentType.try_parse(best_media.codec.type_)
194        return StreamDetails(
195            provider=self.instance_id,
196            item_id=item_id,
197            audio_format=AudioFormat(
198                content_type=content_type,
199                bit_rate=best_media.codec.bitrate if best_media.codec else None,
200            ),
201            media_type=MediaType.RADIO,
202            stream_type=StreamType.HTTP,
203            path=best_media.url,
204            can_seek=False,
205            allow_seek=False,
206        )
207
208    async def _get_podcast_episode_stream_details(self, item_id: str) -> StreamDetails:
209        """Return stream details for a Deezer podcast episode."""
210        result = await self.provider.gql_client.get_podcast_episode(podcast_episode_id=item_id)
211        if result is None or not result.media:
212            raise MediaNotFoundError(f"Podcast episode {item_id} has no stream URL")
213        content_type = ContentType.UNKNOWN
214        if result.media.codec:
215            content_type = ContentType.try_parse(result.media.codec.type_)
216        return StreamDetails(
217            provider=self.instance_id,
218            item_id=item_id,
219            audio_format=AudioFormat(
220                content_type=content_type,
221                bit_rate=result.media.codec.bitrate if result.media.codec else None,
222            ),
223            media_type=MediaType.PODCAST_EPISODE,
224            stream_type=StreamType.HTTP,
225            path=result.media.url,
226            duration=result.duration,
227            can_seek=True,
228            allow_seek=True,
229        )
230
231    # -- Audio stream (Blowfish decryption) --
232
233    async def get_audio_stream(
234        self, streamdetails: StreamDetails, seek_position: int = 0
235    ) -> AsyncGenerator[bytes]:
236        """Return the audio stream for the provider item."""
237        if streamdetails.media_type == MediaType.AUDIOBOOK and isinstance(streamdetails.data, dict):
238            async for chunk in self._stream_audiobook_chapters(streamdetails, seek_position):
239                yield chunk
240            return
241        async for chunk in self._stream_encrypted_track(streamdetails, seek_position):
242            yield chunk
243
244    async def _stream_audiobook_chapters(
245        self, streamdetails: StreamDetails, seek_position: int = 0
246    ) -> AsyncGenerator[bytes]:
247        """Stream audiobook by iterating through encrypted chapter tracks."""
248        chapter_ids: list[str] = streamdetails.data["chapter_ids"]
249        chapter_durations_ms: list[int] = streamdetails.data["chapter_durations_ms"]
250
251        # Resolve which chapter to start from based on seek_position
252        start_chapter = 0
253        chapter_seek = 0
254        if seek_position > 0:
255            seek_ms = seek_position * 1000
256            accumulated_ms = 0
257            for i, dur_ms in enumerate(chapter_durations_ms):
258                if accumulated_ms + dur_ms > seek_ms:
259                    start_chapter = i
260                    chapter_seek = (seek_ms - accumulated_ms) // 1000
261                    break
262                accumulated_ms += dur_ms
263            else:
264                start_chapter = max(len(chapter_ids) - 1, 0)
265                chapter_seek = 0
266
267        prev_chapter: StreamDetails | None = None
268        current_chapter: StreamDetails | None = None
269        try:
270            for i in range(start_chapter, len(chapter_ids)):
271                chapter_id = chapter_ids[i]
272                try:
273                    url_details, song_data = await self.provider.gw_client.get_deezer_track_urls(
274                        chapter_id
275                    )
276                except DeezerGWError, MediaNotFoundError, KeyError:
277                    self.logger.warning("Failed to get URL for audiobook chapter %s", chapter_id)
278                    continue
279                url = url_details["sources"][0]["url"]
280                size_key = f"FILESIZE_{url_details['format']}"
281                size = int(song_data.get(size_key) or song_data.get("FILESIZE_MP3_MISC") or 0)
282                duration = int(song_data["DURATION"])
283                chapter_details = StreamDetails(
284                    item_id=chapter_id,
285                    provider=self.instance_id,
286                    audio_format=streamdetails.audio_format,
287                    stream_type=StreamType.CUSTOM,
288                    duration=duration,
289                    data={
290                        "url": url,
291                        "format": url_details["format"],
292                        "track_id": str(song_data["SNG_ID"]),
293                    },
294                    size=size,
295                )
296                # Log the previous chapter as fully played before starting the next
297                if prev_chapter and "start_ts" in prev_chapter.data:
298                    self.mass.create_task(
299                        self.provider.gw_client.log_listen(last_track=prev_chapter)
300                    )
301                current_chapter = chapter_details
302                seek = chapter_seek if i == start_chapter else 0
303                async for chunk in self._stream_encrypted_track(chapter_details, seek):
304                    yield chunk
305                prev_chapter = chapter_details
306                current_chapter = None
307        finally:
308            # Log the last chapter that was playing (completed or cancelled)
309            last = current_chapter or prev_chapter
310            if last and "start_ts" in last.data:
311                self.mass.create_task(self.provider.gw_client.log_listen(last_track=last))
312
313    async def _stream_encrypted_track(
314        self, streamdetails: StreamDetails, seek_position: int = 0
315    ) -> AsyncGenerator[bytes]:
316        """Stream and decrypt a single encrypted Deezer track."""
317        blowfish_key = self._get_blowfish_key(streamdetails.data["track_id"])
318        chunk_index = 0
319        timeout = ClientTimeout(total=None, connect=30, sock_read=600)
320        headers: dict[str, str] = {}
321
322        # Seek by skipping chunks (Range header causes malformed audio)
323        if seek_position and streamdetails.size and streamdetails.duration:
324            chunk_count = ceil(streamdetails.size / 2048)
325            skip_chunks = int(chunk_count / streamdetails.duration) * seek_position
326        else:
327            skip_chunks = 0
328
329        buffer = bytearray()
330        streamdetails.data["start_ts"] = utc_timestamp()
331        streamdetails.data["stream_id"] = uuid.uuid1()
332        self.mass.create_task(self.provider.gw_client.log_listen(next_track=streamdetails.item_id))
333        async with self.mass.http_session.get(
334            streamdetails.data["url"], headers=headers, timeout=timeout
335        ) as resp:
336            if resp.status != 200:
337                raise MediaNotFoundError(
338                    f"Failed to stream track {streamdetails.item_id}: HTTP {resp.status}"
339                )
340            async for chunk in resp.content.iter_chunked(2048):
341                buffer += chunk
342                if len(buffer) >= 2048:
343                    if chunk_index >= skip_chunks or chunk_index == 0:
344                        if chunk_index % 3 > 0:
345                            yield bytes(buffer[:2048])
346                        else:
347                            yield self._decrypt_chunk(bytes(buffer[:2048]), blowfish_key)
348
349                    chunk_index += 1
350                    del buffer[:2048]
351        yield bytes(buffer)
352
353    async def on_streamed(self, streamdetails: StreamDetails) -> None:
354        """Handle callback when an item completed streaming."""
355        if not isinstance(streamdetails.data, dict) or "start_ts" not in streamdetails.data:
356            return
357        await self.provider.gw_client.log_listen(last_track=streamdetails)
358
359    # -- Decryption helpers --
360
361    @staticmethod
362    def _md5(data: str, data_type: str = "ascii") -> str:
363        md5sum = hashlib.md5()
364        md5sum.update(data.encode(data_type))
365        return md5sum.hexdigest()
366
367    def _get_blowfish_key(self, track_id: str) -> str:
368        """Get blowfish key to decrypt a chunk of a track."""
369        secret = app_var("deezer_decrypt_key")
370        id_md5 = self._md5(track_id)
371        return "".join(
372            chr(ord(id_md5[i]) ^ ord(id_md5[i + 16]) ^ ord(secret[i])) for i in range(16)
373        )
374
375    @staticmethod
376    def _decrypt_chunk(chunk: bytes, blowfish_key: str) -> bytes:
377        """Decrypt a given chunk using the blow fish key."""
378        cipher = Blowfish.new(
379            blowfish_key.encode("ascii"),
380            Blowfish.MODE_CBC,
381            b"\x00\x01\x02\x03\x04\x05\x06\x07",
382        )
383        return cipher.decrypt(chunk)  # type: ignore[no-any-return,unused-ignore]
384
385
386def _raise_stream_error(err: DeezerGWError, item_id: str, label: str) -> NoReturn:
387    """
388    Translate a DeezerGWError into the appropriate streaming error.
389
390    :param err: The GW error raised while resolving the stream URL.
391    :param item_id: The provider-specific item ID, used in the error message.
392    :param label: Human-readable item kind (e.g. "Track", "Audiobook").
393    """
394    api_errors = err.args[1] if len(err.args) > 1 else []
395    # Code 2002 means the item is genuinely unavailable (region/plan rights), so it
396    # maps to MediaNotFoundError. Any other GW failure is treated as a transient
397    # AudioError. Both are caught by the queue's skip-on-error path.
398    if isinstance(api_errors, list) and api_errors and api_errors[0].get("code") == 2002:
399        raise MediaNotFoundError(f"{label} {item_id} is not available on Deezer") from err
400    raise AudioError(str(err)) from err
401