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