music-assistant-server

1"""Streaming operations for Yandex Music."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from typing import TYPE_CHECKING, Any, Final
8
9import aiohttp
10from aiohttp import ClientPayloadError, ServerDisconnectedError
11from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
12from music_assistant_models.enums import ContentType, StreamType
13from music_assistant_models.errors import MediaNotFoundError
14from music_assistant_models.media_items import AudioFormat
15from music_assistant_models.streamdetails import StreamDetails
16
17from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
18
19from .constants import (
20    CONF_CODECS,
21    CONF_QUALITY,
22    CONF_TRANSPORT,
23    QUALITY_BALANCED,
24    QUALITY_EFFICIENT,
25    QUALITY_FILE_INFO_PARAMS,
26    QUALITY_HIGH,
27    QUALITY_SUPERB,
28    RADIO_TRACK_ID_SEP,
29    TRANSPORT_RAW,
30)
31
32if TYPE_CHECKING:
33    from yandex_music import DownloadInfo
34
35    from .provider import YandexMusicProvider
36
37
38# Windowed-stream tuning constants
39_CHUNK_SIZE = 16384  # smaller than default 65536 for faster first-byte after retry
40_STREAM_TIMEOUT = aiohttp.ClientTimeout(total=None, sock_read=30)
41# Yandex CDN drops TCP connections for slow consumers (observed at ~45s for raw transport
42# at real-time playback rate ~200 KB/s). By capping each Range request to 4 MB we download
43# each window quickly, preventing CDN drops for both raw and encrypted transports.
44_RANGE_WINDOW = 4 * 1024 * 1024  # 4 MB per Range request
45# AES-CTR block size in bytes (used for block-aligned Range requests in encrypted transport)
46_AES_BLOCK_SIZE = 16
47# Flat short delays for TCP drops (network glitches within a 4 MB window)
48_TCP_DROP_DELAYS = (0.5, 1.0, 2.0)
49# Exponential delays for true network stalls (read timeout)
50_STALL_DELAYS = (2.0, 4.0, 8.0)
51
52# Normalize Yandex codec names to MA ContentType values
53_CODEC_ALIASES: Final[dict[str, str]] = {
54    "he-aac": "aac",
55    "mpeg": "mp3",
56}
57
58
59class YandexMusicStreamingManager:
60    """Manages Yandex Music streaming operations."""
61
62    def __init__(self, provider: YandexMusicProvider) -> None:
63        """
64        Initialize streaming manager.
65
66        :param provider: The Yandex Music provider instance.
67        """
68        self.provider = provider
69        self.client = provider.client
70        self.mass = provider.mass
71        self.logger = provider.logger
72
73    def _track_id_from_item_id(self, item_id: str) -> str:
74        """Extract API track ID from item_id (may be track_id@station_id for My Wave)."""
75        if RADIO_TRACK_ID_SEP in item_id:
76            return item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
77        return item_id
78
79    async def get_stream_details(self, item_id: str) -> StreamDetails:
80        """
81        Get stream details for a track.
82
83        Uses the unified /get-file-info endpoint for all quality tiers.
84        Falls back to /tracks/{id}/download-info if get-file-info fails.
85
86        :param item_id: Track ID or composite track_id@station_id for My Wave.
87        :return: StreamDetails for the track (item_id preserved for on_streamed).
88        :raises MediaNotFoundError: If stream URL cannot be obtained.
89        """
90        track_id = self._track_id_from_item_id(item_id)
91        track = await self.provider.get_track(item_id)
92        if not track:
93            raise MediaNotFoundError(f"Track {item_id} not found")
94
95        quality = (
96            str(self.provider.config.get_value(CONF_QUALITY) or QUALITY_BALANCED).strip().lower()
97        )
98        transport = (
99            str(self.provider.config.get_value(CONF_TRANSPORT) or TRANSPORT_RAW).strip().lower()
100        )
101
102        # Backward compatibility: old "lossless" config value
103        if quality == "lossless":
104            quality = QUALITY_SUPERB
105
106        fi_params = QUALITY_FILE_INFO_PARAMS.get(
107            quality, QUALITY_FILE_INFO_PARAMS[QUALITY_BALANCED]
108        )
109
110        # Allow advanced users to override codecs
111        codecs_override = str(self.provider.config.get_value(CONF_CODECS) or "").strip()
112        codecs = codecs_override or fi_params["codecs"]
113
114        self.logger.debug(
115            "Requesting stream for track %s: quality=%s, transport=%s, codecs=%s",
116            track_id,
117            quality,
118            transport,
119            codecs,
120        )
121
122        file_info = await self.client.get_track_file_info(
123            track_id,
124            quality=fi_params["quality"],
125            codecs=codecs,
126            transport=transport,
127        )
128
129        if file_info and file_info.get("url"):
130            url = file_info["url"]
131            codec = file_info.get("codec") or ""
132            needs_decryption = file_info.get("needs_decryption", False)
133
134            # Gather audio params: API response first, then probe container
135            bit_rate = file_info.get("bitrate") or file_info.get("bitrate_in_kbps") or 0
136            sample_rate = file_info.get("sample_rate") or 0
137            bit_depth = file_info.get("bit_depth") or 0
138
139            if (not sample_rate or not bit_depth) and not needs_decryption:
140                # Probe raw stream headers for real sample_rate/bit_depth
141                probed_sr, probed_bd = await self._probe_stream_params(url, codec)
142                sample_rate = sample_rate or probed_sr
143                bit_depth = bit_depth or probed_bd
144
145            self.logger.debug(
146                "Audio params for track %s: codec=%s, bit_rate=%s, sample_rate=%s, bit_depth=%s",
147                track_id,
148                codec,
149                bit_rate,
150                sample_rate,
151                bit_depth,
152            )
153
154            audio_format = self._build_audio_format(
155                codec,
156                bit_rate=bit_rate,
157                sample_rate=sample_rate,
158                bit_depth=bit_depth,
159            )
160
161            # Always use StreamType.CUSTOM with windowed Range requests to prevent CDN drops.
162            # can_seek=True only for codecs where bitrate * time yields a decodable byte
163            # offset — i.e. raw MP3. MP4-container codecs (aac-mp4, flac-mp4) need the
164            # ftyp/moov init atoms at the file start, so byte-offset seeks land in mdat
165            # with no codec config and produce undecodable data. Raw FLAC frames aren't
166            # fixed-size either, so byte-rate math doesn't land on a frame boundary.
167            # allow_seek=True lets ffmpeg handle time-based seeking via -ss in those cases.
168            byte_seekable = codec.lower() in ("mp3", "mpeg")
169            can_seek = not needs_decryption and bit_rate > 0 and byte_seekable
170            data: dict[str, Any] = {
171                "url": url,
172                "codec": codec,
173                "transport": transport,
174                "bit_rate": bit_rate,
175                # Stored for URL refresh on 4xx:
176                "fi_quality": fi_params["quality"],
177                "fi_codecs": codecs,
178            }
179            if needs_decryption and "key" in file_info:
180                data["decryption_key"] = file_info["key"]
181
182            return StreamDetails(
183                item_id=item_id,
184                provider=self.provider.instance_id,
185                audio_format=audio_format,
186                stream_type=StreamType.CUSTOM,
187                duration=track.duration,
188                data=data,
189                can_seek=can_seek,
190                allow_seek=not needs_decryption,
191            )
192
193        # Fallback: /tracks/{id}/download-info (defensive, should rarely trigger)
194        self.logger.warning(
195            "get-file-info failed for track %s, falling back to download-info", track_id
196        )
197        download_infos = await self.client.get_track_download_info(track_id, get_direct_links=True)
198        if not download_infos:
199            raise MediaNotFoundError(f"No stream info available for track {item_id}")
200
201        selected_info = self._select_best_quality(download_infos, quality)
202        if not selected_info or not selected_info.direct_link:
203            raise MediaNotFoundError(f"No stream URL available for track {item_id}")
204
205        self.logger.debug(
206            "Fallback stream for track %s: codec=%s, bitrate=%s",
207            track_id,
208            getattr(selected_info, "codec", None),
209            getattr(selected_info, "bitrate_in_kbps", None),
210        )
211
212        return StreamDetails(
213            item_id=item_id,
214            provider=self.provider.instance_id,
215            audio_format=self._build_audio_format(
216                selected_info.codec, bit_rate=selected_info.bitrate_in_kbps or 0
217            ),
218            stream_type=StreamType.HTTP,
219            duration=track.duration,
220            path=selected_info.direct_link,
221            can_seek=True,
222            allow_seek=True,
223            expiration=50,  # download-info direct links expire after ~60s
224        )
225
226    def _select_best_quality(
227        self, download_infos: list[Any], preferred_quality: str | None
228    ) -> DownloadInfo | None:
229        """
230        Select the best quality download info based on user preference.
231
232        Used as fallback when get-file-info is unavailable.
233
234        :param download_infos: List of DownloadInfo objects.
235        :param preferred_quality: User's quality preference (efficient/high/balanced/superb).
236        :return: Best matching DownloadInfo or None.
237        """
238        if not download_infos:
239            return None
240
241        preferred_normalized = (preferred_quality or "").strip().lower()
242
243        # Sort by bitrate descending
244        sorted_infos = sorted(
245            download_infos,
246            key=lambda x: x.bitrate_in_kbps or 0,
247            reverse=True,
248        )
249
250        # Superb: Prefer FLAC. The legacy "lossless" alias still maps to Superb,
251        # but we use an exact-match set so a stray value like "lossless_foo"
252        # doesn't sneak in.
253        if preferred_normalized in {QUALITY_SUPERB, "lossless"}:
254            for codec in ("flac-mp4", "flac"):
255                for info in sorted_infos:
256                    if info.codec and info.codec.lower() == codec:
257                        return info
258            self.logger.warning(
259                "Superb quality (FLAC) requested but not available; using best available"
260            )
261            return sorted_infos[0]
262
263        # Efficient: Prefer lowest bitrate AAC/MP3
264        if preferred_normalized == QUALITY_EFFICIENT:
265            sorted_infos_asc = sorted(
266                download_infos,
267                # ``or float('inf')`` (rather than the previous ``or 999``) makes
268                # the sentinel unambiguous: 999 kbps is conceivably a real
269                # bitrate, but no real Yandex stream reports infinity. Both
270                # ``None`` and ``0`` (which Yandex emits for lossless FLAC) are
271                # falsy and rank last so "efficient" never picks a lossless
272                # stream over a known low-bitrate AAC.
273                key=lambda x: x.bitrate_in_kbps or float("inf"),
274            )
275            for codec in ("aac-mp4", "aac", "he-aac-mp4", "he-aac", "mp3"):
276                for info in sorted_infos_asc:
277                    if info.codec and info.codec.lower() == codec:
278                        return info
279            return sorted_infos_asc[0]
280
281        # High: Prefer high bitrate MP3 (~320kbps)
282        if preferred_normalized == QUALITY_HIGH:
283            high_quality_mp3 = [
284                info
285                for info in sorted_infos
286                if info.codec
287                and info.codec.lower() == "mp3"
288                and info.bitrate_in_kbps
289                and info.bitrate_in_kbps >= 256
290            ]
291            if high_quality_mp3:
292                return high_quality_mp3[0]
293
294            for info in sorted_infos:
295                if info.codec and info.codec.lower() == "mp3":
296                    return info
297
298            for info in sorted_infos:
299                if info.codec and info.codec.lower() not in ("flac", "flac-mp4"):
300                    return info
301
302            return sorted_infos[0]
303
304        # Balanced (default): Prefer ~192kbps AAC
305        balanced_infos = [
306            info
307            for info in sorted_infos
308            if info.bitrate_in_kbps and 128 <= info.bitrate_in_kbps <= 256
309        ]
310        if balanced_infos:
311            for codec in ("aac-mp4", "aac", "he-aac-mp4", "he-aac", "mp3"):
312                for info in balanced_infos:
313                    if info.codec and info.codec.lower() == codec:
314                        return info
315            return balanced_infos[0]
316
317        return sorted_infos[0] if sorted_infos else None
318
319    def _get_content_type(self, codec: str | None) -> tuple[ContentType, ContentType]:
320        """
321        Determine content_type and codec_type from Yandex API codec string.
322
323        Parses the codec string automatically:
324        - Simple codecs ("flac", "mp3", "aac") → (ContentType.<codec>, UNKNOWN)
325        - Compound "codec-container" ("flac-mp4", "aac-mp4") →
326          (ContentType.<codec>, ContentType.<codec>)
327
328        content_type always reflects the audio codec (not the container),
329        so MA's is_lossless() correctly identifies lossless streams and
330        ffmpeg gets the right decoder name via codec_type.
331
332        :param codec: Codec string from Yandex API (e.g. "flac-mp4", "mp3").
333        :return: Tuple of (content_type, codec_type).
334        """
335        if not codec:
336            return ContentType.UNKNOWN, ContentType.UNKNOWN
337
338        codec_lower = codec.lower()
339
340        # Strip container suffix: "flac-mp4" → "flac", "he-aac-mp4" → "he-aac"
341        has_container = codec_lower.endswith("-mp4")
342        audio_part = codec_lower[:-4] if has_container else codec_lower
343
344        # Normalize aliases (he-aac → aac, mpeg → mp3)
345        audio_part = _CODEC_ALIASES.get(audio_part, audio_part)
346
347        try:
348            content_type = ContentType(audio_part)
349        except ValueError:
350            self.logger.debug("Unknown codec from Yandex API: %s", codec)
351            return ContentType.UNKNOWN, ContentType.UNKNOWN
352
353        # For compound formats, set codec_type so ffmpeg knows the decoder
354        codec_type = content_type if has_container else ContentType.UNKNOWN
355        return content_type, codec_type
356
357    def _build_audio_format(
358        self,
359        codec: str | None,
360        *,
361        bit_rate: int = 0,
362        sample_rate: int = 0,
363        bit_depth: int = 0,
364    ) -> AudioFormat:
365        """
366        Build AudioFormat from codec string and optional stream metadata.
367
368        Values of 0 mean "unknown — let MA/ffmpeg detect from the actual stream".
369        Pass real values from the API response when available.
370
371        :param codec: Codec string from Yandex API.
372        :param bit_rate: Bitrate in kbps (0 = unknown).
373        :param sample_rate: Sample rate in Hz (0 = unknown, detect from stream).
374        :param bit_depth: Bit depth (0 = unknown, detect from stream).
375        :return: Configured AudioFormat instance.
376        """
377        content_type, codec_type = self._get_content_type(codec)
378        kwargs: dict[str, Any] = {
379            "content_type": content_type,
380            "codec_type": codec_type,
381        }
382        # Only pass non-zero values; AudioFormat defaults to 44100/16 which
383        # MA/ffmpeg rely on.  Passing 0 would override those defaults.
384        if bit_rate:
385            kwargs["bit_rate"] = bit_rate
386        if sample_rate:
387            kwargs["sample_rate"] = sample_rate
388        if bit_depth:
389            kwargs["bit_depth"] = bit_depth
390        return AudioFormat(**kwargs)
391
392    @staticmethod
393    def _parse_flac_streaminfo(header: bytes) -> tuple[int, int]:
394        """
395        Extract sample_rate and bit_depth from FLAC STREAMINFO block.
396
397        FLAC format: 4-byte magic "fLaC", then metadata blocks.
398        STREAMINFO is always the first block (type 0), 34 bytes payload.
399        Bytes 10-17 of STREAMINFO contain sample_rate (20 bits),
400        channels (3 bits), bit_depth (5 bits), total samples (36 bits).
401
402        :param header: First 42+ bytes of the FLAC stream.
403        :return: (sample_rate, bit_depth) or (0, 0) on parse failure.
404        """
405        if len(header) < 42 or header[:4] != b"fLaC":
406            return 0, 0
407        # STREAMINFO payload: 4 magic + 4 block header = 8 byte offset, 34 bytes long
408        # Bytes 10-13 of payload: sample_rate(20) | channels(3) | bps(5) | total(4 high)
409        payload = header[8:]  # skip "fLaC" + block header
410        if len(payload) < 34:
411            return 0, 0
412        val = int.from_bytes(payload[10:14], "big")
413        sample_rate = (val >> 12) & 0xFFFFF
414        bit_depth = ((val >> 4) & 0x1F) + 1
415        return sample_rate, bit_depth
416
417    @staticmethod
418    def _parse_mp4_audio_params(header: bytes) -> tuple[int, int]:
419        """
420        Extract sample_rate and bit_depth from MP4/fMP4 container.
421
422        Scans for the 'dfLa' (FLAC-in-MP4) box, or falls back to parsing
423        the AudioSampleEntry in an 'mp4a' box to read sample size and sample rate.
424
425        :param header: First 8-32 KB of the MP4 stream.
426        :return: (sample_rate, bit_depth) or (0, 0) if not found.
427        """
428        # Quick scan for dfLa box (FLAC-in-MP4: contains FLAC STREAMINFO)
429        dfl_pos = header.find(b"dfLa")
430        if dfl_pos >= 4:
431            # dfLa box layout after "dfLa" type:
432            #   4 bytes version/flags
433            #   4 bytes STREAMINFO block header (type byte + 3-byte length)
434            #   34 bytes STREAMINFO payload
435            payload_start = dfl_pos + 4 + 4 + 4  # after type + version/flags + block header
436            payload = header[payload_start:]
437            if len(payload) >= 34:
438                val = int.from_bytes(payload[10:14], "big")
439                sample_rate = (val >> 12) & 0xFFFFF
440                bit_depth = ((val >> 4) & 0x1F) + 1
441                if 8000 <= sample_rate <= 384000 and 1 <= bit_depth <= 32:
442                    return sample_rate, bit_depth
443
444        # Scan for mp4a AudioSampleEntry (AAC/generic audio in MP4)
445        mp4a_pos = header.find(b"mp4a")
446        if mp4a_pos >= 4:
447            # AudioSampleEntry: 4-byte size, "mp4a", 6 reserved, 2 data_ref,
448            # 2 version, 2 revision, 4 vendor, 2 channels, 2 sample_size,
449            # 2 compression_id, 2 packet_size, 4 sample_rate (16.16 fixed-point)
450            entry_start = mp4a_pos + 4  # after "mp4a"
451            entry = header[entry_start:]
452            if len(entry) >= 28:
453                sample_size = int.from_bytes(entry[18:20], "big")
454                sr_fixed = int.from_bytes(entry[24:28], "big")
455                sample_rate = sr_fixed >> 16
456                bit_depth = max(0, sample_size)
457                if 8000 <= sample_rate <= 384000:
458                    return sample_rate, bit_depth
459
460        return 0, 0
461
462    async def _probe_stream_params(self, url: str, codec: str) -> tuple[int, int]:
463        """
464        Probe audio params by reading the first bytes of the stream.
465
466        Makes a small Range request to read container/stream headers,
467        then parses FLAC STREAMINFO or MP4 box structure.
468
469        :param url: Stream URL.
470        :param codec: Codec string from API (e.g. "flac-mp4", "flac").
471        :return: (sample_rate, bit_depth) or (0, 0) if probing fails.
472        """
473        codec_lower = (codec or "").lower()
474        # Determine how many bytes to read and which parser to use
475        if codec_lower == "flac":
476            probe_size = 64
477            parser = self._parse_flac_streaminfo
478        elif "-mp4" in codec_lower:
479            probe_size = 32768  # MP4 moov/stsd can be further in
480            parser = self._parse_mp4_audio_params
481        else:
482            return 0, 0  # lossy without container — let MA detect
483
484        try:
485            headers = {"Range": f"bytes=0-{probe_size - 1}"}
486            async with self.mass.http_session.get(
487                url,
488                headers=headers,
489                timeout=aiohttp.ClientTimeout(total=10),
490            ) as resp:
491                if resp.status not in (200, 206):
492                    return 0, 0
493                header_bytes = await resp.content.read(probe_size)
494                self.logger.debug("Probe read %d bytes for codec=%s", len(header_bytes), codec)
495                result = parser(header_bytes)
496                self.logger.debug("Probe result: sample_rate=%d, bit_depth=%d", *result)
497                return result
498        except asyncio.CancelledError:
499            raise
500        except Exception:
501            self.logger.debug("Stream probe failed for codec=%s", codec)
502            return 0, 0
503
504    async def _refresh_stream_url(
505        self,
506        streamdetails: StreamDetails,
507        http_status: int,
508        bytes_yielded: int,
509        attempt: int,
510        max_retries: int,
511    ) -> bool:
512        """
513        Re-fetch an expired stream URL (works for both raw and encraw).
514
515        Updates streamdetails.data in-place with new URL (and key for encraw).
516
517        :return: True on success, False if retries exhausted.
518        """
519        if attempt >= max_retries:
520            return False
521        data = streamdetails.data
522        track_id = self._track_id_from_item_id(streamdetails.item_id)
523        self.logger.warning(
524            "Stream URL expired (HTTP %d) at %d bytes (attempt %d/%d) — re-fetching",
525            http_status,
526            bytes_yielded,
527            attempt + 1,
528            max_retries,
529        )
530        token = BYPASS_THROTTLER.set(True)
531        try:
532            file_info = await self.client.get_track_file_info(
533                track_id,
534                quality=data["fi_quality"],
535                codecs=data["fi_codecs"],
536                transport=data.get("transport", TRANSPORT_RAW),
537            )
538        finally:
539            BYPASS_THROTTLER.reset(token)
540        if file_info and file_info.get("url"):
541            data["url"] = file_info["url"]
542            if "decryption_key" in data and file_info.get("key"):
543                data["decryption_key"] = file_info["key"]
544            return True
545        return False
546
547    async def _decrypt_response_stream(
548        self,
549        response: Any,
550        key_bytes: bytes,
551        block_size: int,
552        bytes_delivered: int,
553    ) -> AsyncGenerator[bytes]:
554        """
555        Decrypt one HTTP response and yield plaintext chunks.
556
557        Aligns the AES-CTR counter to the correct block for resumption.
558        If the server ignores a Range header (200 instead of 206), resets the
559        counter to 0 and skips the already-delivered prefix transparently.
560
561        :param response: aiohttp ClientResponse (open context manager).
562        :param key_bytes: Raw AES key bytes.
563        :param block_size: AES block size (16 for CTR mode).
564        :param bytes_delivered: Total plaintext bytes already sent to the caller.
565        :return: Async generator yielding decrypted audio bytes.
566        """
567        block_start = (bytes_delivered // block_size) * block_size
568        block_skip = bytes_delivered - block_start
569
570        if block_start > 0 and response.status == 200:
571            self.logger.warning(
572                "Server ignored Range header at %d bytes (200 instead of 206)"
573                " — restarting decrypt from position 0, skipping %d already-sent bytes",
574                block_start,
575                bytes_delivered,
576            )
577            block_skip = bytes_delivered
578            block_num = (0).to_bytes(block_size, "big")
579        else:
580            block_num = (block_start // block_size).to_bytes(block_size, "big")
581
582        decryptor = Cipher(algorithms.AES(key_bytes), modes.CTR(block_num)).decryptor()
583        carry_skip = block_skip
584        async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
585            decrypted = decryptor.update(chunk)
586            if carry_skip > 0:
587                skip = min(carry_skip, len(decrypted))
588                decrypted = decrypted[skip:]
589                carry_skip -= skip
590            if decrypted:
591                yield decrypted
592        final = decryptor.finalize()
593        if final:
594            yield final
595
596    def _handle_stream_error(
597        self,
598        err: Exception,
599        attempt: int,
600        max_retries: int,
601        bytes_yielded: int,
602        delays: tuple[float, ...],
603        label: str,
604    ) -> tuple[int, float]:
605        """
606        Increment retry counter, log a warning, or raise if retries are exhausted.
607
608        :param err: The exception that caused the retry.
609        :param attempt: Current retry attempt count (0-based).
610        :param max_retries: Maximum number of retries allowed.
611        :param bytes_yielded: Bytes delivered so far (for log context).
612        :param delays: Backoff delay sequence to pick from.
613        :param label: Short verb describing the failure (e.g. "dropped", "stalled").
614        :return: (new_attempt, retry_delay) tuple when retrying.
615        :raises MediaNotFoundError: When attempt count exceeds max_retries.
616        """
617        delay = delays[min(attempt, len(delays) - 1)]
618        attempt += 1
619        if attempt <= max_retries:
620            self.logger.warning(
621                "Stream %s at %d bytes (attempt %d/%d) — retrying",
622                label,
623                bytes_yielded,
624                attempt,
625                max_retries,
626            )
627            return attempt, delay
628        raise MediaNotFoundError(f"Stream {label} after retries were exhausted") from err
629
630    @staticmethod
631    def _is_content_range_eof(headers: Any, window_end: int) -> bool:
632        """
633        Return True when Content-Range indicates *window_end* reached the last file byte.
634
635        Parses ``Content-Range: bytes start-end/total`` and checks whether
636        ``window_end >= total - 1``.  Returns False on any malformed header so
637        the caller falls back to the next window safely.
638        """
639        content_range = headers.get("Content-Range", "")
640        if not content_range.startswith("bytes "):
641            return False
642        try:
643            _, range_spec = content_range.split(" ", 1)
644            _, total_str = range_spec.split("/", 1)
645            total_str = total_str.strip()
646            return total_str.isdigit() and window_end >= int(total_str) - 1
647        except ValueError:
648            return False
649
650    async def _iter_raw_response(
651        self,
652        response: Any,
653        bytes_delivered: int,
654        block_start: int,
655    ) -> AsyncGenerator[bytes]:
656        """
657        Yield raw (unencrypted) chunks from one HTTP response.
658
659        If the server ignored the Range header (200 instead of 206), skips the
660        already-delivered prefix transparently.
661
662        :param response: aiohttp ClientResponse (open context manager).
663        :param bytes_delivered: Total bytes already sent to the caller.
664        :param block_start: Requested Range start offset.
665        :return: Async generator yielding raw audio bytes.
666        """
667        range_ignored = response.status == 200 and block_start > 0
668        skip_bytes = bytes_delivered if range_ignored else 0
669        async for raw_chunk in response.content.iter_chunked(_CHUNK_SIZE):
670            if skip_bytes > 0:
671                if len(raw_chunk) <= skip_bytes:
672                    skip_bytes -= len(raw_chunk)
673                    continue
674                raw_chunk = raw_chunk[skip_bytes:]  # noqa: PLW2901
675                skip_bytes = 0
676            if raw_chunk:
677                yield raw_chunk
678
679    async def _handle_expired_url(
680        self,
681        streamdetails: StreamDetails,
682        response_status: int,
683        bytes_yielded: int,
684        attempt: int,
685        max_retries: int,
686    ) -> bytes:
687        """
688        Handle URL expiry (401/403/410) by refreshing and returning updated key.
689
690        :return: Updated AES key bytes for encrypted streams, or empty ``bytes``
691            for raw streams.
692        :raises MediaNotFoundError: When refresh fails or retries are exhausted.
693        """
694        if not await self._refresh_stream_url(
695            streamdetails,
696            response_status,
697            bytes_yielded,
698            attempt,
699            max_retries,
700        ):
701            raise MediaNotFoundError(
702                f"Stream URL expired (HTTP {response_status}) after retries exhausted"
703            )
704        data = streamdetails.data
705        if "decryption_key" in data:
706            try:
707                return bytes.fromhex(data["decryption_key"])
708            except ValueError as err:
709                raise MediaNotFoundError(f"Invalid decryption key: {err}") from err
710        return b""
711
712    @staticmethod
713    def _validate_encryption_key(data: dict[str, Any]) -> tuple[bool, bytes | None]:
714        """
715        Validate and extract encryption parameters from stream data.
716
717        :return: (is_encrypted, key_bytes) tuple.
718        :raises MediaNotFoundError: If AES key length is invalid.
719        """
720        if "decryption_key" not in data:
721            return False, None
722        try:
723            key_bytes = bytes.fromhex(data["decryption_key"])
724        except ValueError as err:
725            raise MediaNotFoundError(f"Invalid decryption key: {err}") from err
726        if len(key_bytes) not in (16, 24, 32):
727            raise MediaNotFoundError(f"Unsupported AES key length: {len(key_bytes)} bytes")
728        return True, key_bytes
729
730    def _calculate_seek_offset(
731        self, data: dict[str, Any], seek_position: int, is_encrypted: bool
732    ) -> int:
733        """
734        Calculate initial byte offset for raw transport seeking.
735
736        :param data: Stream data dict (must contain 'bit_rate' in kbps).
737        :param seek_position: Seek offset in seconds.
738        :param is_encrypted: Whether the stream uses AES encryption.
739        :return: Byte offset to start streaming from (0 if not applicable).
740        """
741        if seek_position <= 0 or is_encrypted:
742            return 0
743        bit_rate = data.get("bit_rate") or 0
744        if not bit_rate:
745            return 0
746        byte_offset = seek_position * bit_rate * 1000 // 8
747        self.logger.debug(
748            "Seeking to %ds: byte offset %d (bitrate %d kbps)",
749            seek_position,
750            byte_offset,
751            bit_rate,
752        )
753        return byte_offset
754
755    async def get_audio_stream(
756        self, streamdetails: StreamDetails, seek_position: int = 0
757    ) -> AsyncGenerator[bytes]:
758        """
759        Return the audio stream via windowed Range requests.
760
761        Handles both raw (direct) and encraw (AES-CTR encrypted) transports.
762        Downloads in windowed Range requests of _RANGE_WINDOW bytes each to prevent
763        Yandex CDN from dropping slow-consumer TCP connections.
764
765        On connection drop: flat short backoff (0.5s/1.0s/2.0s).
766        On read stall: exponential backoff (2s/4s/8s).
767        On URL expiry (HTTP 4xx): re-fetches URL and resumes from bytes_yielded.
768        Retry counter resets after each successful window.
769
770        :param streamdetails: Stream details with URL (and optional decryption key).
771        :param seek_position: Seek offset in seconds for raw transport (0 = from start).
772        :return: Async generator yielding audio bytes.
773        """
774        data = streamdetails.data
775        is_encrypted, key_bytes = self._validate_encryption_key(data)
776        initial_byte_offset = self._calculate_seek_offset(data, seek_position, is_encrypted)
777
778        max_retries = 6
779        bytes_yielded = initial_byte_offset
780        attempt = 0
781        retry_delay: float = 0.0
782
783        while True:
784            if attempt > 0:
785                await asyncio.sleep(retry_delay)
786
787            block_start = (
788                (bytes_yielded // _AES_BLOCK_SIZE) * _AES_BLOCK_SIZE
789                if is_encrypted
790                else bytes_yielded
791            )
792            window_end = block_start + _RANGE_WINDOW - 1
793
794            try:
795                async with self.mass.http_session.get(
796                    data["url"],
797                    headers={"Range": f"bytes={block_start}-{window_end}"},
798                    timeout=_STREAM_TIMEOUT,
799                ) as response:
800                    if response.status in (401, 403, 410):
801                        new_key = await self._handle_expired_url(
802                            streamdetails,
803                            response.status,
804                            bytes_yielded,
805                            attempt,
806                            max_retries,
807                        )
808                        if is_encrypted:
809                            key_bytes = new_key
810                        attempt += 1
811                        retry_delay = 0.0
812                        continue
813                    try:
814                        response.raise_for_status()
815                    except Exception as err:
816                        # Do not embed err.__str__ — aiohttp's ClientResponseError
817                        # includes the signed CDN URL, which carries an expiring
818                        # signature that should not reach logs or the frontend.
819                        raise MediaNotFoundError(
820                            f"Failed to fetch stream: HTTP {response.status}"
821                        ) from err
822
823                    bytes_before = bytes_yielded
824                    if is_encrypted:
825                        if key_bytes is None:
826                            raise MediaNotFoundError("Missing decryption key")
827                        block_skip = bytes_before - block_start
828                        async for chunk in self._decrypt_response_stream(
829                            response,
830                            key_bytes,
831                            _AES_BLOCK_SIZE,
832                            bytes_yielded,
833                        ):
834                            bytes_yielded += len(chunk)
835                            yield chunk
836                    else:
837                        range_ignored = response.status == 200 and block_start > 0
838                        block_skip = bytes_before if range_ignored else 0
839                        async for chunk in self._iter_raw_response(
840                            response,
841                            bytes_before,
842                            block_start,
843                        ):
844                            bytes_yielded += len(chunk)
845                            yield chunk
846
847                    received = (bytes_yielded - bytes_before) + block_skip
848                    if response.status == 200 or received < _RANGE_WINDOW:
849                        return
850                    if self._is_content_range_eof(response.headers, window_end):
851                        return
852                    attempt = 0
853                    retry_delay = 0.0
854
855            except asyncio.CancelledError:
856                raise
857            except (ClientPayloadError, ServerDisconnectedError) as err:
858                attempt, retry_delay = self._handle_stream_error(
859                    err,
860                    attempt,
861                    max_retries,
862                    bytes_yielded,
863                    _TCP_DROP_DELAYS,
864                    "dropped",
865                )
866            except TimeoutError as err:
867                attempt, retry_delay = self._handle_stream_error(
868                    err,
869                    attempt,
870                    max_retries,
871                    bytes_yielded,
872                    _STALL_DELAYS,
873                    "stalled",
874                )
875