/
/
/
1"""
2Storytel provider helper utilities.
3
4Lightweight async client helpers used by the Storytel provider for
5interacting with Storytel API endpoints.
6"""
7
8from __future__ import annotations
9
10import logging
11from asyncio import Lock, Task, TaskGroup
12from collections.abc import Callable, Coroutine
13from dataclasses import dataclass
14from datetime import UTC, datetime
15from json import JSONDecodeError
16from typing import TYPE_CHECKING, Any, TypeVar, cast
17from urllib.parse import quote
18
19from aiohttp.client_exceptions import ClientError, ContentTypeError
20from Crypto.Cipher import AES
21from Crypto.Util.Padding import pad
22from music_assistant_models.enums import (
23 ContentType,
24 ImageType,
25 MediaType,
26 StreamType,
27)
28from music_assistant_models.errors import (
29 InvalidDataError,
30 LoginFailed,
31 MediaNotFoundError,
32 ProviderUnavailableError,
33 UnplayableMediaError,
34 UnsupportedFeaturedException,
35)
36from music_assistant_models.media_items import (
37 Audiobook,
38 AudioFormat,
39 MediaItemChapter,
40 MediaItemImage,
41 MediaItemType,
42 Podcast,
43 PodcastEpisode,
44 ProviderMapping,
45 RecommendationFolder,
46 UniqueList,
47)
48from music_assistant_models.streamdetails import StreamDetails
49from yarl import URL
50
51from .constants import (
52 API_DEFAULT_RESOURCE_VERSION,
53 API_ENCRYPTION_IV,
54 API_ENCRYPTION_KEY,
55 API_HEADER_CONTENT_TYPE_BOOK_DETAILS,
56 API_HEADER_CONTENT_TYPE_EXPLORE,
57 API_HEADER_CONTENT_TYPE_LIBRARY_DELTA,
58 API_HEADER_CONTENT_TYPE_SEARCH,
59 API_HEADER_STORYTEL_MEDIA_ACCEPT,
60 API_HEADER_STORYTEL_MEDIA_FORMATS,
61 URL_BOOKMARK_GET,
62 URL_BOOKMARK_SET,
63 URL_CONSUMABLE_DETAILS,
64 URL_CONSUMABLE_DOWNLOAD_NO_RANGE,
65 URL_FRONTPAGE,
66 URL_LIBRARY_MANAGEMENT,
67 URL_LOGIN,
68 URL_PLAYBACK_BOOK_DETAILS,
69 URL_PODCAST_DETAILS,
70 URL_REVALIDATE,
71 URL_SEARCH,
72)
73
74if TYPE_CHECKING:
75 from aiohttp import ClientResponse, ClientSession
76
77 from music_assistant.providers.storytel import Storytel
78
79
80# Generic type for async fetch functions used by _fetch_search_page
81T = TypeVar("T")
82
83
84@dataclass
85class StorytelAuth:
86 """
87 Authentication tokens container for Storytel API.
88
89 Holds the JWT and the single-sign token used for Storytel requests.
90 """
91
92 jwt: str
93 single_sign_token: str
94
95
96class StorytelHelper:
97 """Async client for the Storytel API, for endpoints needed by the provider."""
98
99 _KEY = API_ENCRYPTION_KEY
100 _IV = API_ENCRYPTION_IV
101 # Query params carrying login credentials (see URL_LOGIN), redacted before logging.
102 _SENSITIVE_QUERY_PARAMS = frozenset({"uid", "pwd"})
103
104 def __init__(
105 self,
106 session: ClientSession,
107 provider_instance: Storytel,
108 provider_id: str,
109 provider_domain: str,
110 kids_mode: bool = False,
111 languages: dict[str, str] | None = None,
112 logger: logging.Logger | None = None,
113 ) -> None:
114 """
115 Initialize the StorytelHelper.
116
117 :param session: aiohttp ClientSession used for requests.
118 :param provider_instance: parent provider instance.
119 :param provider_id: provider instance id.
120 :param provider_domain: provider domain string.
121 :param kids_mode: whether to request kids-mode content.
122 :param languages: optional language mapping for queries.
123 :param logger: optional logger to use.
124 """
125 self._session = session
126 self._auth: StorytelAuth | None = None
127 self.provider_instance = provider_instance
128 self.provider_id = provider_id
129 self.provider_domain = provider_domain
130 self._kids_mode = kids_mode
131 self._languages = languages or {}
132 self._resource_version: str | None = None
133 self._revalidate_lock = Lock()
134 self.logger = logger or logging.getLogger("storytel_helper")
135
136 @property
137 def authorized(self) -> bool:
138 """Return whether the provider is currently authorized."""
139 return bool(self._auth and self._auth.jwt and self._auth.single_sign_token)
140
141 @property
142 def session(self) -> ClientSession:
143 """Return the aiohttp session used for Storytel requests."""
144 return self._session
145
146 @property
147 def languages_query(self) -> str:
148 """Return comma-separated ISO language codes for API queries."""
149 if not self._languages:
150 return "en"
151 iso_values = sorted(self._languages.values())
152 return ",".join(iso_values)
153
154 @property
155 def kids_mode_query(self) -> str:
156 """Return the Storytel kids-mode query value."""
157 return "true" if self._kids_mode else "false"
158
159 @property
160 def resource_version(self) -> str:
161 """Return the current Storytel resource version or the default fallback."""
162 if self._resource_version:
163 return self._resource_version
164 return API_DEFAULT_RESOURCE_VERSION
165
166 @resource_version.setter
167 def resource_version(self, value: str) -> None:
168 self._resource_version = value
169
170 def headers_api(self) -> dict[str, str]:
171 """Build the standard API headers used for Storytel requests."""
172 headers: dict[str, str] = {
173 "Accept": "application/json",
174 "User-Agent": "MusicAssistant-Storytel/1.0",
175 }
176 if self._auth:
177 headers["authorization"] = f"bearer {self._auth.jwt}"
178 return headers
179
180 async def raise_for_status(self, resp: ClientResponse) -> None:
181 """Raise a provider exception for non-success Storytel responses."""
182 if 200 <= resp.status < 300:
183 return
184 try:
185 resp_json = await resp.json()
186 resp_message = resp_json.get("message") or ""
187 except ContentTypeError, JSONDecodeError:
188 resp_message = await resp.text() or "<no response>"
189 if resp.status in (401, 403):
190 raise LoginFailed(f"Unauthorized ({resp.status}): {resp_message}")
191 self.logger.warning(
192 "Failed Storytel API request with status %s: %s", resp.status, resp_message
193 )
194 raise ProviderUnavailableError(f"Storytel HTTP {resp.status}: {resp_message}")
195
196 async def login(self, username: str, password: str) -> StorytelAuth:
197 """
198 Authenticate with the Storytel API.
199
200 :param username: the username.
201 :param password: the password.
202 """
203 enc = self._encrypt_password_hex(password)
204 url = URL_LOGIN.replace("{UID}", quote(username, safe="")).replace(
205 "{PASSWORD}", quote(enc, safe="")
206 )
207 data = await self._request_json("GET", url)
208 acc = data.get("accountInfo") or {}
209 jwt = acc.get("jwt")
210 sst = acc.get("singleSignToken")
211 if not jwt or not sst:
212 # API returns a message on login errors
213 msg = data.get("message") or "Invalid credentials"
214 raise LoginFailed(msg)
215 self._auth = StorytelAuth(jwt=jwt, single_sign_token=sst)
216 return self._auth
217
218 async def revalidate_account(self) -> StorytelAuth:
219 """Revalidate the Storytel account using the single sign token."""
220 if not self._auth or not self._auth.single_sign_token:
221 raise LoginFailed("No single sign token")
222 current_token = self._auth.single_sign_token
223 async with self._revalidate_lock:
224 if not self._auth or self._auth.single_sign_token != current_token:
225 return self._auth
226 try:
227 data = await self._request_json(
228 "POST", URL_REVALIDATE, json_data={"token": current_token}
229 )
230 except LoginFailed as err:
231 self.logger.warning("Storytel account revalidation failed: %s", err)
232 raise ProviderUnavailableError(
233 "Storytel account revalidation failed, token may be expired. Please login again."
234 ) from err
235 acc = data.get("accountInfo") or {}
236 jwt = acc.get("jwt")
237 sst = acc.get("singleSignToken")
238 if not jwt or not sst:
239 msg = data.get("message") or "Revalidation failed"
240 raise LoginFailed(msg)
241 self._auth = StorytelAuth(jwt=jwt, single_sign_token=sst)
242 return self._auth
243
244 async def get_library(self) -> tuple[dict[str, Any], dict[str, Any]]:
245 """
246 Fetch the user's library, including both bookshelf items and followed items (e.g. podcasts).
247
248 Returns a tuple of (library_items, following_items) where each is a dict keyed by consumableId.
249 """
250 url = URL_LIBRARY_MANAGEMENT
251 headers = self.headers_api()
252 headers["Accept"] = API_HEADER_CONTENT_TYPE_LIBRARY_DELTA
253
254 data = await self._request_json("POST", url, headers=headers, json_data={})
255
256 library_items = data.get("items") or {}
257 following_items = data.get("followingItems") or {}
258
259 # Pre-process library_items such that they only include items where model.resultType is "book". This is needed as the library endpoint returns both books and consuming podcasts, and we want to keep them separate for now.
260 library_items = {
261 k: v
262 for k, v in library_items.items()
263 if (model := v.get("model") or {})
264 and self._is_audiobook(model)
265 and self._abook_is_released(model.get("formats") or [{}])
266 }
267
268 return library_items, following_items
269
270 async def get_consumable_details(self, consumable_id: str) -> dict[str, Any]:
271 """
272 Fetch consumable details from the Storytel API.
273
274 :param consumable_id: the consumable id.
275 """
276 url = URL_CONSUMABLE_DETAILS.replace("{CONSUMABLE_ID}", consumable_id)
277 headers = self.headers_api()
278 headers["Accept"] = API_HEADER_CONTENT_TYPE_BOOK_DETAILS
279 return await self._request_json("GET", url, headers=headers)
280
281 async def get_bookmark(self, consumable_id: str) -> dict[str, Any] | None:
282 """
283 Fetch the bookmark for a consumable from the Storytel API.
284
285 :param consumable_id: the consumable id.
286 """
287 url = URL_BOOKMARK_GET.replace("{CONSUMABLE_ID}", consumable_id)
288 data = await self._request_json("GET", url, headers=self.headers_api())
289 bookmarks = data.get("bookmarks") or []
290 # Only abook bookmark
291 for bm in bookmarks:
292 if bm.get("type") == "abook":
293 return cast("dict[str, Any]", bm)
294 return None
295
296 async def set_bookmark(
297 self, consumable_id: str, position: int, kids_mode: bool = False
298 ) -> dict[str, Any]:
299 """
300 Set the bookmark for a consumable.
301
302 :param consumable_id: the consumable id.
303 :param position: the position in seconds.
304 :param kids_mode: True if kids mode is enabled.
305 """
306 payload = {
307 "consumableId": consumable_id,
308 "kidsMode": kids_mode,
309 # MA provides seconds; Storytel expects milliseconds
310 "position": int(position * 1000),
311 "secondsSinceCreated": 0,
312 "type": "abook",
313 }
314 headers = {**self.headers_api(), "content-type": "application/json"}
315 return await self._request_json(
316 "POST", URL_BOOKMARK_SET, headers=headers, json_data=payload
317 )
318
319 async def fetch_resource_version(self) -> None:
320 """Fetch and cache the resource version for the Storytel API."""
321 url = URL_LIBRARY_MANAGEMENT
322 headers = self.headers_api()
323 headers["Accept"] = API_HEADER_CONTENT_TYPE_LIBRARY_DELTA
324 request_data: dict[str, Any] = {
325 "resourceVersion": self.resource_version,
326 "followingItems": {},
327 "items": {},
328 }
329
330 response_data = await self._request_json(
331 "POST", url, headers=headers, json_data=request_data
332 )
333
334 resource_version = response_data.get("resourceVersion", "")
335
336 if resource_version != "":
337 self.logger.debug("Fetched Storytel resource version: %s", resource_version)
338 self.resource_version = resource_version
339 else:
340 self.logger.debug("No resource version found in Storytel response.")
341 self.resource_version = API_DEFAULT_RESOURCE_VERSION
342
343 async def add_to_bookshelf(self, consumable_id: str, item: MediaItemType) -> bool:
344 """
345 Add an audiobook or podcast to the user's bookshelf.
346
347 :param consumable_id: the consumable id.
348 :param item: the media item to add.
349 """
350 url = URL_LIBRARY_MANAGEMENT
351 headers = self.headers_api()
352 headers["Accept"] = API_HEADER_CONTENT_TYPE_LIBRARY_DELTA
353 request_data: dict[str, Any] = {
354 "resourceVersion": self.resource_version,
355 "followingItems": {},
356 "items": {},
357 }
358 if item.media_type == MediaType.PODCAST:
359 cast("dict[str, Any]", request_data["followingItems"])["podcast-" + consumable_id] = {
360 "id": consumable_id,
361 "action": "SET",
362 "resultType": "podcast",
363 "state": "DO_NOT_NOTIFY",
364 "millisecondsSinceEvent": 10,
365 }
366 elif item.media_type == MediaType.AUDIOBOOK:
367 cast("dict[str, Any]", request_data["items"])[consumable_id] = {
368 "action": "SET",
369 "state": "WILL_CONSUME",
370 "millisecondsSinceEvent": 10,
371 }
372
373 response_data = await self._request_json(
374 "POST", url, headers=headers, json_data=request_data
375 )
376
377 following_items = response_data.get("followingItems") or {}
378 library_items = response_data.get("items") or {}
379 success = False
380 if item.media_type == MediaType.PODCAST:
381 podcast_item = following_items.get("podcast-" + consumable_id) or {}
382 success = (
383 podcast_item.get("action") == "SET"
384 and (podcast_item.get("model") or {}).get("state") == "DO_NOT_NOTIFY"
385 )
386 elif item.media_type == MediaType.AUDIOBOOK:
387 audiobook_item = library_items.get(consumable_id) or {}
388 success = (
389 audiobook_item.get("action") == "SET"
390 and (audiobook_item.get("model") or {}).get("state") == "WILL_CONSUME"
391 )
392
393 if success:
394 self.logger.debug("Added %s %s to bookshelf.", item.media_type.value, consumable_id)
395 self.resource_version = response_data.get("resourceVersion", self.resource_version)
396 else:
397 self.logger.debug(
398 "Failed to add %s %s to bookshelf.", item.media_type.value, consumable_id
399 )
400
401 return success
402
403 async def remove_from_bookshelf(self, consumable_id: str, media_type: MediaType) -> bool:
404 """
405 Remove an audiobook or podcast from the user's bookshelf.
406
407 :param consumable_id: the consumable id.
408 :param media_type: the media type.
409 """
410 url = URL_LIBRARY_MANAGEMENT
411 headers = self.headers_api()
412 headers["Accept"] = API_HEADER_CONTENT_TYPE_LIBRARY_DELTA
413 request_data: dict[str, Any] = {
414 "resourceVersion": self.resource_version,
415 "followingItems": {},
416 "items": {},
417 }
418 if media_type == MediaType.PODCAST:
419 cast("dict[str, Any]", request_data["followingItems"])["podcast-" + consumable_id] = {
420 "id": consumable_id,
421 "millisecondsSinceEvent": 10,
422 "resultType": "podcast",
423 "action": "DELETE",
424 }
425 elif media_type == MediaType.AUDIOBOOK:
426 cast("dict[str, Any]", request_data["items"])[consumable_id] = {
427 "action": "DELETE",
428 "millisecondsSinceEvent": 10,
429 }
430
431 response_data = await self._request_json(
432 "POST", url, headers=headers, json_data=request_data
433 )
434
435 following_items = response_data.get("followingItems") or {}
436 library_items = response_data.get("items") or {}
437 success = False
438 if media_type == MediaType.PODCAST:
439 podcast_item = following_items.get("podcast-" + consumable_id) or {}
440 success = podcast_item.get("action") == "DELETE"
441 elif media_type == MediaType.AUDIOBOOK:
442 audiobook_item = library_items.get(consumable_id) or {}
443 success = audiobook_item.get("action") == "DELETE"
444
445 if success:
446 self.logger.debug("Removed %s %s from bookshelf.", media_type.value, consumable_id)
447 self.resource_version = response_data.get("resourceVersion", self.resource_version)
448 else:
449 self.logger.debug(
450 "Failed to remove %s %s from bookshelf.", media_type.value, consumable_id
451 )
452
453 return success
454
455 def parse_podcast(self, podcast_data: dict[str, Any]) -> Podcast:
456 """Parse Storytel podcast data to Music Assistant Podcast."""
457 list_metadata = podcast_data.get("listMetadata") or {}
458 media_type = list_metadata.get("type") or ""
459 if media_type != "podcast":
460 self.logger.debug(
461 "Skipping non-podcast item of type %s: %s", media_type, podcast_data.get("id")
462 )
463 raise InvalidDataError(f"Item {podcast_data.get('id')} is not a podcast")
464
465 consumable_id = podcast_data.get("id") or ""
466 title = podcast_data.get("title") or "Unknown"
467 episode_count = podcast_data.get("totalCount") or 0
468 hosts = parse_podcast_hosts(list_metadata)
469
470 podcast = Podcast(
471 item_id=consumable_id,
472 provider=self.provider_id,
473 name=title,
474 provider_mappings={
475 ProviderMapping(
476 item_id=consumable_id,
477 provider_domain=self.provider_domain,
478 provider_instance=self.provider_id,
479 )
480 },
481 favorite=False,
482 total_episodes=episode_count,
483 )
484
485 cover_url = (list_metadata.get("imageUrl") or {}).get("url") or ""
486 if cover_url:
487 podcast.metadata.images = UniqueList(
488 [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=self.provider_id)]
489 )
490 podcast.metadata.description = list_metadata.get("description") or ""
491 language = list_metadata.get("language") or ""
492 if language:
493 podcast.metadata.languages = UniqueList([language])
494 genre = list_metadata.get("genre") or ""
495 if genre:
496 podcast.metadata.genres = {genre}
497 podcast.metadata.performers = set(hosts)
498 latest_episode_date_text = (list_metadata.get("followingInfo") or {}).get(
499 "newestItemReleaseDate"
500 ) or ""
501 if latest_episode_date_text:
502 podcast.metadata.release_date = datetime.fromisoformat(
503 latest_episode_date_text
504 ).astimezone(UTC)
505
506 return podcast
507
508 async def get_podcast_episodes(
509 self,
510 prov_podcast_id: str,
511 total_episodes: int | None = None,
512 include_languages: str | None = None,
513 ) -> list[dict[str, Any]]:
514 """
515 Get all podcast episodes for a specific podcast.
516
517 :param prov_podcast_id: the provider podcast id.
518 :param total_episodes: the total number of episodes for the podcast.
519 :param include_languages: optional comma-separated language filter for the episode list.
520 """
521 languages_query = include_languages or self.languages_query
522
523 async def fetch_page(token: str = "") -> dict[str, Any]:
524 url = URL_PODCAST_DETAILS.replace("{CONSUMABLE_ID}", prov_podcast_id)
525 url += (
526 "?configVariant=voice-switcher-enabled"
527 "&includeFormats=ebook%2Cabook%2Cpodcast"
528 f"&includeLanguages={quote(languages_query, safe='')}"
529 f"&kidsMode={self.kids_mode_query}"
530 "&orderBy=default"
531 )
532 if token != "":
533 url += f"&nextPageToken={quote(token, safe='')}"
534 headers = self.headers_api()
535 headers["Accept"] = API_HEADER_CONTENT_TYPE_EXPLORE
536 return await self._request_json("GET", url, headers=headers)
537
538 async def fetch_items(token: str) -> list[dict[str, Any]]:
539 page_data = await fetch_page(token)
540 return page_data.get("items") or []
541
542 async def fetch_pages(page_tokens: list[str]) -> list[dict[str, Any]]:
543 page_results: list[Task[list[dict[str, Any]]]] = []
544
545 async with TaskGroup() as tg:
546 for token in page_tokens:
547 page_results.append(tg.create_task(fetch_items(token)))
548
549 results: list[dict[str, Any]] = []
550 for task in page_results:
551 results.extend(task.result())
552 return results
553
554 if not total_episodes or total_episodes <= 0:
555 return []
556
557 page_size = 10
558 page_tokens = [str(page_offset) for page_offset in range(0, total_episodes, page_size)]
559 if not page_tokens:
560 return []
561 return await fetch_pages(page_tokens)
562
563 async def parse_media_item(self, item_data: dict[str, Any]) -> MediaItemType:
564 """Parse a media item from Storytel API to the appropriate Music Assistant media item type."""
565 item_type = item_data.get("type") or ""
566 media_item: Audiobook | PodcastEpisode
567
568 if item_type == "detailedPodcastEpisode":
569 media_item = await self._parse_podcast_episode_item(item_data)
570 elif item_type == "detailedBook":
571 media_item = await self._parse_audiobook_item(item_data)
572 else:
573 self.logger.warning("Unsupported media item type for parsing: %s", item_type)
574 raise UnsupportedFeaturedException(f"Unsupported media item type: {item_type}")
575
576 return self._apply_media_item_metadata(media_item, item_data)
577
578 async def get_podcast_details(self, consumable_id: str) -> dict[str, Any]:
579 """
580 Get details of a podcast from the Storytel API.
581
582 :param consumable_id: the consumable id.
583 """
584 url = URL_PODCAST_DETAILS.replace("{CONSUMABLE_ID}", consumable_id)
585 headers = self.headers_api()
586 headers["Accept"] = API_HEADER_CONTENT_TYPE_EXPLORE
587 return await self._request_json("GET", url, headers=headers)
588
589 async def search_podcasts(
590 self, query: str, limit: int = 10, page_token: str = "", results_count: int = 0
591 ) -> list[Podcast]:
592 """
593 Search for podcasts matching the query.
594
595 :param query: the search query.
596 :param limit: the maximum number of results.
597 :param page_token: the page token for pagination.
598 :param results_count: the current count of results.
599 """
600 max_pages = 10
601
602 def filter_podcasts(results: list[dict[str, Any]]) -> list[dict[str, Any]]:
603 """Filter results to extract podcast result payloads."""
604 return [result for result in results if result.get("resultType") == "podcast"]
605
606 async def fetch_page(search_page_token: str) -> tuple[list[Podcast], int, int]:
607 return await self._fetch_search_page(
608 query,
609 "podcast_shows",
610 search_page_token,
611 filter_podcasts,
612 self._parse_search_podcast_item,
613 )
614
615 podcasts: list[Podcast] = []
616 current_page_token = page_token
617 current_results_count = max(results_count, 0)
618
619 page_podcasts, total_results, next_page_offset = await fetch_page(current_page_token)
620 podcasts.extend(page_podcasts)
621 current_results_count += len(page_podcasts)
622
623 if next_page_offset <= 0 or next_page_offset >= total_results:
624 return podcasts
625
626 page_size = max(
627 next_page_offset - (int(current_page_token) if current_page_token else 0), 1
628 )
629 page_batch_size = 5
630 pages_fetched = 1
631
632 while (
633 current_results_count < limit
634 and next_page_offset < total_results
635 and pages_fetched < max_pages
636 ):
637 batch_tokens = [
638 str(page_offset)
639 for page_offset in range(
640 next_page_offset,
641 total_results,
642 page_size,
643 )
644 ][:page_batch_size]
645 if not batch_tokens:
646 break
647
648 batch_results: list[list[Podcast]] = []
649 async with TaskGroup() as tg:
650 batch_tasks = [tg.create_task(fetch_page(token)) for token in batch_tokens]
651 for task in batch_tasks:
652 page_podcasts, _, _ = task.result()
653 batch_results.append(page_podcasts)
654
655 for page_podcasts in batch_results:
656 podcasts.extend(page_podcasts)
657 current_results_count += len(page_podcasts)
658
659 pages_fetched += len(batch_tokens)
660 next_page_offset = int(batch_tokens[-1]) + page_size
661
662 return podcasts
663
664 async def search_audiobooks(
665 self, query: str, limit: int = 10, page_token: str = "", results_count: int = 0
666 ) -> list[Audiobook]:
667 """
668 Search for audiobooks matching the query.
669
670 :param query: the search query.
671 :param limit: the maximum number of results.
672 :param page_token: the page token for pagination.
673 :param results_count: the current count of results.
674 """
675 max_pages = 10
676
677 def filter_audiobooks(results: list[dict[str, Any]]) -> list[dict[str, Any]]:
678 """Filter results to extract audiobook result payloads."""
679 return [
680 result
681 for result in results
682 if result.get("resultType") == "book"
683 and any(
684 isinstance(book_format, dict) and book_format.get("type") == "abook"
685 for book_format in result.get("formats", [])
686 )
687 ]
688
689 async def fetch_page(search_page_token: str) -> tuple[list[Audiobook], int, int]:
690 return await self._fetch_search_page(
691 query,
692 "books",
693 search_page_token,
694 filter_audiobooks,
695 self._parse_search_audiobook_item,
696 )
697
698 audiobooks: list[Audiobook] = []
699 current_page_token = page_token
700 current_results_count = max(results_count, 0)
701
702 page_audiobooks, total_results, next_page_offset = await fetch_page(current_page_token)
703 audiobooks.extend(page_audiobooks)
704 current_results_count += len(page_audiobooks)
705
706 if next_page_offset <= 0 or next_page_offset >= total_results:
707 return audiobooks
708
709 page_size = max(
710 next_page_offset - (int(current_page_token) if current_page_token else 0), 1
711 )
712 page_batch_size = 5
713 pages_fetched = 1
714
715 while (
716 current_results_count < limit
717 and next_page_offset < total_results
718 and pages_fetched < max_pages
719 ):
720 batch_tokens = [
721 str(page_offset)
722 for page_offset in range(
723 next_page_offset,
724 total_results,
725 page_size,
726 )
727 ][:page_batch_size]
728 if not batch_tokens:
729 break
730
731 batch_results: list[list[Audiobook]] = []
732 async with TaskGroup() as tg:
733 batch_tasks = [tg.create_task(fetch_page(token)) for token in batch_tokens]
734 for task in batch_tasks:
735 page_audiobooks, _, _ = task.result()
736 batch_results.append(page_audiobooks)
737
738 for page_audiobooks in batch_results:
739 audiobooks.extend(page_audiobooks)
740 current_results_count += len(page_audiobooks)
741
742 pages_fetched += len(batch_tokens)
743 next_page_offset = int(batch_tokens[-1]) + page_size
744
745 return audiobooks
746
747 async def get_recommendations(self) -> list[RecommendationFolder]:
748 """Get audiobook and podcast recommendations for the user."""
749 chip_url = URL_FRONTPAGE
750 chip_url += (
751 "?includeFormats=abook%2Cpodcast"
752 f"&includeLanguages={quote(self.languages_query, safe='')}"
753 f"&kidsMode={self.kids_mode_query}"
754 "&onboarding=false&version=2"
755 )
756
757 chip_response = await self._fetch_explore_page(chip_url)
758 chips = chip_response.get("chips") or []
759
760 folders: list[RecommendationFolder] = []
761
762 frontpage_chip = next(
763 (chip for chip in chips if str(chip.get("id") or "").startswith("frontpage")),
764 None,
765 )
766 if frontpage_chip and (frontpage_url := str(frontpage_chip.get("url") or "").strip()):
767 frontpage_url += (
768 "?categoryIds="
769 "&configVariant=voice-switcher-enabled"
770 "&includeFormats=abook%2Cpodcast"
771 f"&includeLanguages={quote(self.languages_query, safe='')}"
772 f"&kidsMode={self.kids_mode_query}"
773 "&onboarding=false&version=2"
774 )
775 if folder := await self._recommendation_folder_from_block_url(
776 frontpage_url,
777 block_id_prefixes=("personal-recommendations_",),
778 folder_item_id=f"{self.provider_id}_recommendations",
779 folder_name="Recommended for You",
780 translation_key="recommended_for_you",
781 ):
782 folders.append(folder)
783
784 podcast_chip = next(
785 (
786 chip
787 for chip in chips
788 if "podcast" in str(chip.get("id") or "").lower()
789 and not str(chip.get("id") or "").startswith("frontpage")
790 ),
791 None,
792 )
793 if podcast_chip and (podcast_url := str(podcast_chip.get("url") or "").strip()):
794 if folder := await self._recommendation_folder_from_block_url(
795 podcast_url,
796 block_id_prefixes=("algorithmic-podcasts-for-you",),
797 folder_item_id=f"{self.provider_id}_podcast_recommendations",
798 folder_name="Recommended Podcasts",
799 translation_key="recommended_podcasts",
800 ):
801 folders.append(folder)
802
803 self.logger.debug(
804 "Storytel recommendation folders resolved: %d (chips=%d)", len(folders), len(chips)
805 )
806
807 return folders
808
809 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
810 """Resolve Storytel stream information for an audiobook or podcast episode."""
811 try:
812 stream_info_url = URL_CONSUMABLE_DOWNLOAD_NO_RANGE.replace("{CONSUMABLE_ID}", item_id)
813 headers = self.headers_api()
814 headers[API_HEADER_STORYTEL_MEDIA_ACCEPT] = API_HEADER_STORYTEL_MEDIA_FORMATS
815
816 async with self.session.get(stream_info_url, headers=headers) as stream_info_response:
817 await self.raise_for_status(stream_info_response)
818 response = await stream_info_response.json()
819 stream_url = (response.get("result") or {}).get("signedUrl") or ""
820 if stream_url == "":
821 self.logger.error("No signed URL returned for %s", item_id)
822 raise MediaNotFoundError("No signed URL returned")
823 headers.pop("authorization", None)
824 headers["range"] = "bytes=0-1"
825 current_stream_url = URL(stream_url, encoded=True)
826 resolved_stream_url = stream_url
827 stream_headers: dict[str, str] = {}
828 for _redirect_count in range(5):
829 async with self.session.get(
830 current_stream_url, headers=headers, allow_redirects=False
831 ) as stream_response:
832 if 300 <= stream_response.status < 400:
833 location = stream_response.headers.get("Location") or ""
834 if location == "":
835 await self.raise_for_status(stream_response)
836 location_url = URL(location, encoded=True)
837 current_stream_url = (
838 location_url
839 if location.startswith("http")
840 else current_stream_url.join(location_url)
841 )
842 resolved_stream_url = str(current_stream_url)
843 continue
844
845 await self.raise_for_status(stream_response)
846 stream_headers = parse_raw_headers(stream_response.raw_headers)
847 content_type, content_full_size = parse_partial_content_probe(
848 status_code=stream_response.status,
849 stream_headers=stream_headers,
850 )
851 resolved_stream_url = str(current_stream_url)
852 break
853 else:
854 raise UnplayableMediaError("Too many redirects while resolving Storytel stream")
855 except (
856 ClientError,
857 ConnectionError,
858 TimeoutError,
859 ContentTypeError,
860 JSONDecodeError,
861 LoginFailed,
862 MediaNotFoundError,
863 ProviderUnavailableError,
864 UnplayableMediaError,
865 ) as err:
866 if isinstance(err, (LoginFailed, MediaNotFoundError, UnplayableMediaError)):
867 raise
868 self.logger.error("Failed to fetch stream details for %s: %s", item_id, err)
869 raise UnplayableMediaError(f"Failed to fetch stream details: {err}") from err
870
871 mass_content_type: ContentType | None = parse_content_type(content_type)
872 content_duration_seconds: int = int(
873 float(stream_headers.get("x-amz-meta-x-durationseconds", 0))
874 )
875 if content_duration_seconds == 0:
876 if media_type == MediaType.PODCAST_EPISODE:
877 item_details: (
878 Audiobook | PodcastEpisode
879 ) = await self.provider_instance.get_podcast_episode(item_id)
880 else:
881 item_details = await self.provider_instance.get_audiobook(item_id)
882 content_duration_seconds = item_details.duration or 0
883
884 return StreamDetails(
885 provider=self.provider_id,
886 size=content_full_size,
887 item_id=item_id,
888 audio_format=AudioFormat(content_type=mass_content_type or ContentType.UNKNOWN),
889 media_type=media_type,
890 stream_type=StreamType.HTTP,
891 path=resolved_stream_url,
892 can_seek=True,
893 allow_seek=True,
894 duration=content_duration_seconds,
895 )
896
897 async def _request_json(
898 self,
899 method: str,
900 url: str,
901 *,
902 headers: dict[str, str] | None = None,
903 json_data: dict[str, Any] | None = None,
904 ) -> dict[str, Any]:
905 """
906 Perform a Storytel API request and return the parsed JSON response body.
907
908 :param method: the HTTP method to use, e.g. "GET" or "POST".
909 :param url: the request URL.
910 :param headers: optional request headers.
911 :param json_data: optional JSON request body.
912 """
913 safe_url = self._redact_url_query(url)
914 try:
915 async with self.session.request(method, url, headers=headers, json=json_data) as resp:
916 await self.raise_for_status(resp)
917 return cast("dict[str, Any]", await resp.json())
918 except (ClientError, ConnectionError, TimeoutError) as err:
919 self.logger.error("Storytel request %s %s failed: %s", method, safe_url, err)
920 raise ProviderUnavailableError(f"Storytel request to {safe_url} failed: {err}") from err
921 except (ContentTypeError, JSONDecodeError) as err:
922 self.logger.error(
923 "Storytel request %s %s returned an invalid response: %s", method, safe_url, err
924 )
925 raise ProviderUnavailableError(
926 f"Storytel returned an invalid response for {method} {safe_url}"
927 ) from err
928
929 def _redact_url_query(self, url: str) -> str:
930 """Mask sensitive login credentials in a URL's query string before it is logged."""
931 try:
932 parsed = URL(url)
933 if not parsed.query:
934 return str(parsed)
935 redacted_query = {
936 key: ("***" if key in self._SENSITIVE_QUERY_PARAMS else value)
937 for key, value in parsed.query.items()
938 }
939 return str(parsed.with_query(redacted_query))
940 except ValueError:
941 return url.split("?", 1)[0]
942
943 async def _fetch_explore_page(self, page_url: str) -> dict[str, Any]:
944 """Fetch a Storytel explore page or list-resource payload."""
945 headers = self.headers_api()
946 headers["Accept"] = API_HEADER_CONTENT_TYPE_EXPLORE
947 return await self._request_json("GET", page_url, headers=headers)
948
949 async def _recommendation_folder_from_block_url(
950 self,
951 block_url: str,
952 *,
953 block_id_prefixes: tuple[str, ...],
954 folder_item_id: str,
955 folder_name: str | None = None,
956 translation_key: str | None = None,
957 ) -> RecommendationFolder | None:
958 """Build a recommendation folder from a Storytel explore page URL."""
959 page_response = await self._fetch_explore_page(block_url)
960 content_blocks = page_response.get("contentBlocks") or []
961
962 block = next(
963 (
964 content_block
965 for content_block in content_blocks
966 if any(
967 str(content_block.get("id") or "").startswith(prefix)
968 for prefix in block_id_prefixes
969 )
970 ),
971 None,
972 )
973 if not block:
974 self.logger.debug(
975 "No Storytel recommendation block matched prefixes %s",
976 ",".join(block_id_prefixes),
977 )
978 return None
979
980 items = block.get("items") or []
981 if not items:
982 items_url = str(block.get("itemsUrl") or "").strip()
983 if items_url:
984 self.logger.debug(
985 "Storytel recommendation block %s has no inline items; fetching itemsUrl",
986 block.get("id") or "<missing>",
987 )
988 items_response = await self._fetch_explore_page(items_url)
989 items = items_response.get("items") or []
990 else:
991 self.logger.debug(
992 "Storytel recommendation block %s did not include items or itemsUrl",
993 block.get("id") or "<missing>",
994 )
995 parsed_items = await self._parse_recommendation_items(items)
996 if not parsed_items:
997 self.logger.debug(
998 "Storytel recommendation block %s produced no parsable items",
999 block.get("id") or "<missing>",
1000 )
1001 return None
1002
1003 folder = RecommendationFolder(
1004 item_id=folder_item_id,
1005 provider=self.provider_id,
1006 icon="mdi-star-circle-outline",
1007 name=(folder_name or str(block.get("title") or "Recommendations")).strip(),
1008 translation_key=translation_key,
1009 )
1010 folder.items.extend(parsed_items)
1011 return folder
1012
1013 async def _parse_recommendation_items(
1014 self, items: list[dict[str, Any]]
1015 ) -> list[Audiobook | Podcast]:
1016 """Parse Storytel recommendation payload items into media items."""
1017
1018 async def _parse_item(item: dict[str, Any]) -> Audiobook | Podcast | None:
1019 item_type = str(item.get("resultType") or "").strip().lower()
1020 if item_type not in {"book", "podcast"}:
1021 return None
1022 item_id = str(item.get("id") or "").strip()
1023 if not item_id:
1024 return None
1025 try:
1026 if item_type == "book":
1027 return await self._parse_search_audiobook_item(item)
1028 return await self._parse_search_podcast_item(item)
1029 except (InvalidDataError, MediaNotFoundError, ProviderUnavailableError) as err:
1030 self.logger.debug("Skipping Storytel recommendation item %s: %s", item_id, err)
1031 return None
1032
1033 task_results: list[Task[Audiobook | Podcast | None]] = []
1034
1035 async with TaskGroup() as tg:
1036 for item in items:
1037 task_results.append(tg.create_task(_parse_item(item)))
1038
1039 parsed_items: list[Audiobook | Podcast] = []
1040 for task in task_results:
1041 parsed_item = task.result()
1042 if parsed_item is not None:
1043 parsed_items.append(parsed_item)
1044 return parsed_items
1045
1046 def _encrypt_password_hex(self, password: str) -> str:
1047 """Encrypt password for Storytel API."""
1048 # AES-128-CBC encrypt password, hex encoded (PKCS#7 padding).
1049 # Source for key and IV: https://github.com/MauritsWilke/storytel-api/blob/v1_archive/src/utils/encryptPassword.ts
1050 cipher = AES.new(self._KEY, AES.MODE_CBC, self._IV)
1051 enc: bytes = cipher.encrypt(pad(password.encode("utf-8"), AES.block_size))
1052 return enc.hex()
1053
1054 def _abook_is_released(self, formats_data: list[dict[str, Any]]) -> bool:
1055 for format_data in formats_data:
1056 if format_data.get("type") != "abook":
1057 continue
1058 is_released: bool = format_data.get("isReleased") or False
1059 return is_released
1060 return False
1061
1062 def _is_audiobook(self, model_data: dict[str, Any]) -> bool:
1063 result_type: str = model_data.get("resultType") or ""
1064 return result_type == "book"
1065
1066 def _parse_duration(self, duration_data: dict[str, int]) -> int:
1067 """Parse duration data to seconds."""
1068 if not duration_data:
1069 return 0
1070 hours = int(duration_data.get("hours") or 0)
1071 minutes = int(duration_data.get("minutes") or 0)
1072 seconds = int(duration_data.get("seconds") or 0)
1073 return hours * 3600 + minutes * 60 + seconds
1074
1075 async def _fetch_chapters(self, consumable_id: str) -> list[dict[str, Any]]:
1076 """Fetch chapters for a given consumable_id."""
1077 chapters: list[dict[str, Any]] = []
1078
1079 url = URL_PLAYBACK_BOOK_DETAILS.replace("{CONSUMABLE_ID}", consumable_id)
1080 try:
1081 data = await self._request_json("GET", url, headers=self.headers_api())
1082 formats = data.get("formats") or []
1083 audiobook_format = next((f for f in formats if f.get("type") == "abook"), None)
1084 chapters = audiobook_format.get("chapters") or [] if audiobook_format else []
1085 except (ProviderUnavailableError, KeyError, TypeError, ValueError) as err:
1086 self.logger.debug("Failed to fetch chapters for %s: %s", consumable_id, err)
1087
1088 return list(chapters)
1089
1090 def _parse_chapters(self, chapters_data: list[dict[str, Any]]) -> list[MediaItemChapter]:
1091 """Parse raw chapter data into MediaChapter objects."""
1092 chapters: list[MediaItemChapter] = []
1093 chapters_data = self._compute_chapter_start(chapters_data)
1094 for chap in chapters_data:
1095 chapter_number = int(chap.get("number") or 0)
1096 title = chap.get("title") or f"Chapter {chapter_number}"
1097 start = chap.get("startPosition") or 0
1098 end = chap.get("endPosition") or 0
1099 chapter = MediaItemChapter(
1100 position=chapter_number,
1101 name=title,
1102 start=start,
1103 end=end,
1104 )
1105 chapters.append(chapter)
1106 return chapters
1107
1108 def _compute_chapter_start(self, chapters_data: list[dict[str, Any]]) -> list[dict[str, Any]]:
1109 """Augment the chapters data with computed start positions."""
1110 for i, chap in enumerate(chapters_data):
1111 if i == 0:
1112 chap["startPosition"] = 0
1113 chap["endPosition"] = int(chap.get("durationInSeconds", 0))
1114 else:
1115 prev_chap = chapters_data[i - 1]
1116 chap["startPosition"] = prev_chap.get("endPosition", 0)
1117 chap["endPosition"] = chap["startPosition"] + int(chap.get("durationInSeconds", 0))
1118 return chapters_data
1119
1120 async def _parse_podcast_episode_item(self, item_data: dict[str, Any]) -> PodcastEpisode:
1121 consumable_id = item_data.get("consumableId") or ""
1122 title = item_data.get("title") or "Unknown"
1123 duration_seconds = self._parse_duration(item_data.get("duration") or {})
1124 podcast_info = item_data.get("seriesInfo") or {}
1125 mass_podcast = await self.provider_instance.get_podcast(podcast_info.get("id") or "")
1126 hosts = parse_podcast_hosts(item_data)
1127 episode_number = item_data.get("seriesInfo", {}).get("orderInSeries") or 0
1128 media_item = PodcastEpisode(
1129 item_id=consumable_id,
1130 provider=self.provider_id,
1131 podcast=mass_podcast,
1132 name=title,
1133 duration=duration_seconds,
1134 position=0,
1135 provider_mappings={
1136 ProviderMapping(
1137 item_id=consumable_id,
1138 provider_domain=self.provider_domain,
1139 provider_instance=self.provider_id,
1140 )
1141 },
1142 )
1143 if hosts:
1144 media_item.metadata.performers = set(hosts)
1145 if episode_number:
1146 media_item.position = episode_number
1147 return media_item
1148
1149 async def _parse_audiobook_item(self, item_data: dict[str, Any]) -> Audiobook:
1150 consumable_id = item_data.get("consumableId") or ""
1151 title = item_data.get("title") or "Unknown"
1152 duration_seconds = self._parse_duration(item_data.get("duration") or {})
1153 authors = [a.get("name") for a in item_data.get("authors", []) if a.get("name")]
1154 narrators = [n.get("name") for n in item_data.get("narrators", []) if n.get("name")]
1155 publisher = ((item_data.get("formats") or [{}])[0].get("publisher") or {}).get("name") or ""
1156 media_item = Audiobook(
1157 item_id=consumable_id,
1158 provider=self.provider_id,
1159 name=title,
1160 duration=duration_seconds,
1161 provider_mappings={
1162 ProviderMapping(
1163 item_id=consumable_id,
1164 provider_domain=self.provider_domain,
1165 provider_instance=self.provider_id,
1166 )
1167 },
1168 publisher=publisher,
1169 favorite=False,
1170 )
1171 chapters = await self._fetch_chapters(consumable_id=consumable_id)
1172 chapters_list = self._parse_chapters(chapters)
1173 if authors:
1174 media_item.authors.set(authors)
1175 if narrators:
1176 media_item.narrators.set(narrators)
1177 if chapters_list:
1178 media_item.metadata.chapters = chapters_list
1179 return media_item
1180
1181 def _apply_media_item_metadata(
1182 self,
1183 media_item: Audiobook | PodcastEpisode,
1184 item_data: dict[str, Any],
1185 ) -> Audiobook | PodcastEpisode:
1186 release_date = (item_data.get("formats") or [{}])[0].get("releaseDate") or ""
1187 description = item_data.get("description") or ""
1188 cover_url = (item_data.get("cover") or {}).get("url")
1189 language = item_data.get("language") or ""
1190 category_name = (item_data.get("category") or {}).get("name") or ""
1191 languages = UniqueList([language]) if language else UniqueList()
1192 genres = {category_name} if category_name else set()
1193
1194 if release_date:
1195 media_item.metadata.release_date = datetime.fromisoformat(release_date).astimezone(UTC)
1196 if description:
1197 media_item.metadata.description = description
1198 if cover_url:
1199 media_item.metadata.images = UniqueList(
1200 [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=self.provider_id)]
1201 )
1202 if languages:
1203 media_item.metadata.languages = languages
1204 if genres:
1205 media_item.metadata.genres = genres
1206
1207 return media_item
1208
1209 async def _parse_search_audiobook_item(self, item_data: dict[str, Any]) -> Audiobook:
1210 """Build a lightweight audiobook item from a search result payload."""
1211 consumable_id = item_data.get("id") or ""
1212 title = item_data.get("title") or item_data.get("name") or "Unknown"
1213 authors_data = item_data.get("authors") or []
1214 authors = [
1215 str(author.get("name"))
1216 for author in authors_data
1217 if isinstance(author, dict) and author.get("name")
1218 ]
1219 media_item = Audiobook(
1220 item_id=consumable_id,
1221 provider=self.provider_id,
1222 name=title,
1223 provider_mappings={
1224 ProviderMapping(
1225 item_id=consumable_id,
1226 provider_domain=self.provider_domain,
1227 provider_instance=self.provider_id,
1228 )
1229 },
1230 favorite=False,
1231 )
1232 if authors:
1233 media_item.authors.set(authors)
1234
1235 audiobook_formats_data = [
1236 format_data
1237 for format_data in (item_data.get("formats") or [])
1238 if isinstance(format_data, dict) and format_data.get("type") == "abook"
1239 ]
1240 cover_url = None
1241 for format_data in audiobook_formats_data:
1242 if isinstance(format_data.get("cover"), dict) and (
1243 cover_url := format_data["cover"].get("url")
1244 ):
1245 break
1246 if cover_url:
1247 media_item.metadata.images = UniqueList(
1248 [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=self.provider_id)]
1249 )
1250 return media_item
1251
1252 async def _parse_search_podcast_item(self, item_data: dict[str, Any]) -> Podcast:
1253 """Build a lightweight podcast item from a search result payload."""
1254 podcast_id = item_data.get("id") or ""
1255 title = item_data.get("title") or item_data.get("name") or "Unknown"
1256 total_episodes = int(item_data.get("numberOfEpisodes") or 0)
1257 podcast = Podcast(
1258 item_id=podcast_id,
1259 provider=self.provider_id,
1260 name=title,
1261 total_episodes=total_episodes,
1262 provider_mappings={
1263 ProviderMapping(
1264 item_id=podcast_id,
1265 provider_domain=self.provider_domain,
1266 provider_instance=self.provider_id,
1267 )
1268 },
1269 )
1270 cover_url = (
1271 (item_data.get("cover") or {}).get("url")
1272 if isinstance(item_data.get("cover"), dict)
1273 else None
1274 )
1275 if cover_url:
1276 podcast.metadata.images = UniqueList(
1277 [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=self.provider_id)]
1278 )
1279 return podcast
1280
1281 async def _fetch_search_page(
1282 self,
1283 query: str,
1284 search_for: str,
1285 page_token: str,
1286 filter_func: Callable[[list[dict[str, Any]]], list[dict[str, Any]]],
1287 fetch_func: Callable[[dict[str, Any]], Coroutine[Any, Any, T]],
1288 ) -> tuple[list[T], int, int]:
1289 """
1290 Fetch a single search page and prepare items for the given search type.
1291
1292 :param query: The search query string.
1293 :param search_for: The search type (e.g., "podcast_shows" or "books").
1294 :param page_token: The page token for this request.
1295 :param filter_func: Callable to filter item IDs from results.
1296 :param fetch_func: Async callable to fetch full item details by ID.
1297 """
1298 url = URL_SEARCH
1299 url += (
1300 "?configVariant=baseline"
1301 f"&searchFor={quote(search_for, safe='')}"
1302 "&includeFormats=abook"
1303 f"&includeLanguages={quote(self.languages_query, safe='')}"
1304 f"&kidsMode={self.kids_mode_query}"
1305 f"&query={quote(query, safe='')}"
1306 "&v2=true"
1307 )
1308 if page_token != "":
1309 url += f"&page={quote(page_token, safe='')}"
1310 headers = self.headers_api()
1311 headers["Accept"] = API_HEADER_CONTENT_TYPE_SEARCH
1312 data = await self._request_json("GET", url, headers=headers)
1313
1314 if data is None:
1315 raise MediaNotFoundError(f"No search results found for query '{query}'")
1316 results = data.get("items") or []
1317 item_results = filter_func(results)
1318 self.logger.debug(
1319 "Filtered away %d results, %d candidates on page %s for query '%s'",
1320 len(results) - len(item_results),
1321 len(item_results),
1322 page_token or "0",
1323 query,
1324 )
1325
1326 page_items: list[T] = []
1327 if item_results:
1328 task_results: list[Task[T]] = []
1329 async with TaskGroup() as tg:
1330 for item_data in item_results:
1331 task_results.append(tg.create_task(fetch_func(item_data)))
1332 for task in task_results:
1333 page_items.append(task.result())
1334
1335 total_results = int(data.get("totalCount", 0) or 0)
1336 next_page_offset = int(data.get("nextPageToken", 0) or 0)
1337 return page_items, total_results, next_page_offset
1338
1339
1340def parse_raw_headers(raw_headers: tuple[tuple[bytes, bytes], ...]) -> dict[str, str]:
1341 """
1342 Parse raw bytes headers into a dictionary of string key-pairs.
1343
1344 :param raw_headers: the raw headers to parse.
1345 """
1346 headers: dict[str, str] = {}
1347 for key_bytes, value_bytes in raw_headers:
1348 key = key_bytes.decode("utf-8").lower()
1349 value = value_bytes.decode("utf-8")
1350 headers[key] = value
1351 return headers
1352
1353
1354def parse_content_type(content_type: str) -> ContentType | None:
1355 """
1356 Parse the content type string to a MA ContentType.
1357
1358 :param content_type: the content type string.
1359 """
1360 # Split on ';' to separate media type from codec params (e.g. "audio/mp4;codecs=mp4a.40.2")
1361 parts = content_type.split(";", 1) if content_type else []
1362 content_type = parts[0].strip() if parts else ""
1363 codec = parts[1].strip() if len(parts) > 1 else ""
1364 if codec == "codecs=ec-3":
1365 return ContentType.EAC3
1366 if codec == "codecs=mp4a.40.2":
1367 return ContentType.AAC
1368 if content_type == "audio/mp4":
1369 return ContentType.MP4
1370 if content_type == "audio/mpeg":
1371 return ContentType.MP3
1372 return None
1373
1374
1375def parse_partial_content_probe(
1376 status_code: int, stream_headers: dict[str, str]
1377) -> tuple[str, int]:
1378 """
1379 Validate a range probe response and extract stream metadata.
1380
1381 :param status_code: the HTTP status code from the probe response.
1382 :param stream_headers: the parsed response headers.
1383 """
1384 if status_code != 206:
1385 raise UnplayableMediaError(
1386 "Storytel stream endpoint did not honor range request (expected HTTP 206)"
1387 )
1388
1389 content_type = stream_headers.get("content-type", "")
1390 content_content_range = stream_headers.get("content-range", "")
1391 if content_content_range == "":
1392 raise UnplayableMediaError("Storytel stream response missing Content-Range header")
1393
1394 try:
1395 content_full_size = int(content_content_range.rsplit("/", maxsplit=1)[-1])
1396 except ValueError as err:
1397 raise UnplayableMediaError(
1398 "Storytel stream response has invalid Content-Range header"
1399 ) from err
1400 return content_type, content_full_size
1401
1402
1403def parse_podcast_hosts(podcast_metadata: dict[str, Any]) -> list[str]:
1404 """
1405 Parse podcast hosts from metadata.
1406
1407 :param podcast_metadata: the podcast metadata dictionary.
1408 """
1409 hosts_list = []
1410 hosts_metadata = podcast_metadata.get("hosts") or []
1411 for host in hosts_metadata:
1412 if host.get("name"):
1413 hosts_list.append(host.get("name"))
1414
1415 return hosts_list
1416