/
/
/
1"""
2Storytel provider integration.
3
4Provides the MusicProvider implementation and glue to the StorytelHelper
5lightweight client used to interact with Storytel APIs.
6"""
7
8from __future__ import annotations
9
10import functools
11from asyncio import Semaphore, Task, TaskGroup
12from collections.abc import AsyncGenerator, Callable
13from datetime import UTC, datetime
14from typing import TYPE_CHECKING, Any, TypeVar, cast
15
16from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
17from music_assistant_models.enums import (
18 ConfigEntryType,
19 MediaType,
20 ProviderFeature,
21)
22from music_assistant_models.errors import (
23 InvalidDataError,
24 LoginFailed,
25 MediaNotFoundError,
26 ProviderUnavailableError,
27 SetupFailedError,
28)
29from music_assistant_models.media_items import (
30 Audiobook,
31 BrowseFolder,
32 ItemMapping,
33 Podcast,
34 PodcastEpisode,
35 RecommendationFolder,
36 SearchResults,
37 UniqueList,
38)
39
40from music_assistant.models.music_provider import MusicProvider
41from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
42
43from .constants import (
44 ALL_LANGUAGES,
45 CACHE_CATEGORY_AUDIOBOOK,
46 CACHE_CATEGORY_PODCAST,
47 CACHE_CATEGORY_PODCAST_EPISODE,
48 CACHE_CATEGORY_PODCAST_EPISODES,
49 CONF_KIDS_MODE,
50 CONF_LANGUAGES,
51 CONF_PASSWORD,
52 CONF_USERNAME,
53 DEFAULT_LANGUAGES,
54)
55from .storytel_helper import StorytelHelper
56
57if TYPE_CHECKING:
58 from music_assistant_models.config_entries import ProviderConfig
59 from music_assistant_models.media_items import MediaItemType
60 from music_assistant_models.provider import ProviderManifest
61 from music_assistant_models.streamdetails import StreamDetails
62
63 from music_assistant.mass import MusicAssistant
64 from music_assistant.models import ProviderInstanceType
65
66F = TypeVar("F", bound=Callable[..., Any])
67
68
69def _unwrap_single_exception(err: ExceptionGroup[Exception]) -> BaseException:
70 """
71 Return the sole exception wrapped by a TaskGroup's ExceptionGroup, if there is only one.
72
73 A concurrent TaskGroup wraps any child failure in an ExceptionGroup, which would otherwise
74 hide the real (and often specific, catchable) exception type from callers.
75
76 :param err: the ExceptionGroup raised by a TaskGroup.
77 """
78 if len(err.exceptions) == 1 and not isinstance(err.exceptions[0], ExceptionGroup):
79 return err.exceptions[0]
80 return err
81
82
83# -------------------------------
84# Setup and configuration entries
85# -------------------------------
86
87
88async def setup(
89 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
90) -> ProviderInstanceType:
91 """
92 Set up the Storytel provider.
93
94 :param mass: the MusicAssistant instance.
95 :param manifest: the provider manifest.
96 :param config: the provider config.
97 """
98 return Storytel(mass=mass, manifest=manifest, config=config)
99
100
101# -------------------------------
102# Provider implementation
103# -------------------------------
104
105
106class Storytel(RecommendationPayloadMixin, MusicProvider):
107 """Storytel provider."""
108
109 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
110 """
111 Return the config entries for the Storytel provider.
112
113 Credentials are collected in the setup flow and persisted in setup_data.
114 The options page only exposes persisted discovery settings.
115 """
116 setup_languages = self.get_setup_value(CONF_LANGUAGES)
117 if isinstance(setup_languages, list) and setup_languages:
118 default_languages = [name for name in setup_languages if isinstance(name, str)]
119 else:
120 default_languages = list(DEFAULT_LANGUAGES)
121 return (
122 ConfigEntry(
123 key=CONF_LANGUAGES,
124 type=ConfigEntryType.STRING,
125 multi_value=True,
126 required=False,
127 default_value=default_languages,
128 options=[ConfigValueOption(name, name) for name in sorted(ALL_LANGUAGES.keys())],
129 ),
130 ConfigEntry(
131 key=CONF_KIDS_MODE,
132 type=ConfigEntryType.BOOLEAN,
133 required=False,
134 default_value=False,
135 ),
136 )
137
138 @staticmethod
139 def handle_login_failed(
140 method: F,
141 ) -> F:
142 """
143 Decorate a method to retry once after a login failure.
144
145 :param method: the method to decorate.
146 """
147
148 @functools.wraps(method)
149 async def wrapper(*args: Any, **kwargs: Any) -> Any:
150 self = cast("Storytel", args[0])
151 try:
152 return await method(*args, **kwargs)
153 except LoginFailed:
154 await self.api.revalidate_account()
155 return await method(*args, **kwargs)
156 except ExceptionGroup as err:
157 if err.subgroup(LoginFailed):
158 await self.api.revalidate_account()
159 return await method(*args, **kwargs)
160 raise _unwrap_single_exception(err) from err
161
162 return cast("F", wrapper)
163
164 @staticmethod
165 def handle_login_failed_generator(
166 method: F,
167 ) -> F:
168 """
169 Decorate an async generator method to retry once after login failure.
170
171 :param method: the async generator method to decorate.
172 """
173
174 @functools.wraps(method)
175 async def wrapper(*args: Any, **kwargs: Any) -> AsyncGenerator[Any]:
176 self = cast("Storytel", args[0])
177 try:
178 async for item in method(*args, **kwargs):
179 yield item
180 except LoginFailed:
181 await self.api.revalidate_account()
182 async for item in method(*args, **kwargs):
183 yield item
184 except ExceptionGroup as err:
185 if err.subgroup(LoginFailed):
186 await self.api.revalidate_account()
187 async for item in method(*args, **kwargs):
188 yield item
189 return
190 raise _unwrap_single_exception(err) from err
191
192 return cast("F", wrapper)
193
194 def __init__(
195 self,
196 mass: MusicAssistant,
197 manifest: ProviderManifest,
198 config: ProviderConfig,
199 ) -> None:
200 """
201 Initialize the Storytel provider instance.
202
203 Accepts the same args/kwargs as the base MusicProvider constructor.
204 """
205 super().__init__(mass=mass, manifest=manifest, config=config)
206 self._api: StorytelHelper | None = None
207 # Whether to use Kids mode for bookmarks
208 self._kids_mode: bool = bool(self.get_setup_value(CONF_KIDS_MODE))
209 # Selected languages for discovery features (recommendations, search)
210 self._languages: dict[str, str] = DEFAULT_LANGUAGES.copy()
211
212 @property
213 def api(self) -> StorytelHelper:
214 """Return the initialized Storytel helper API instance."""
215 if self._api is None:
216 raise SetupFailedError("Storytel provider API not initialized")
217 return self._api
218
219 @property
220 def supported_features(self) -> set[ProviderFeature]:
221 """Return the supported features by this provider."""
222 return {
223 ProviderFeature.LIBRARY_AUDIOBOOKS,
224 ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT,
225 ProviderFeature.LIBRARY_PODCASTS,
226 ProviderFeature.LIBRARY_PODCASTS_EDIT,
227 ProviderFeature.RECOMMENDATIONS,
228 ProviderFeature.SEARCH,
229 }
230
231 async def handle_async_init(self) -> None:
232 """Handle async initialization of the provider."""
233 username = str(self.get_setup_value(CONF_USERNAME) or "")
234 password = str(self.get_setup_value(CONF_PASSWORD) or "")
235 if not username or not password:
236 raise LoginFailed("Username/password required")
237
238 languages_list = self.get_setup_value(CONF_LANGUAGES)
239 if isinstance(languages_list, list) and languages_list:
240 self._languages = {
241 name: ALL_LANGUAGES[name]
242 for name in languages_list
243 if isinstance(name, str) and name in ALL_LANGUAGES
244 }
245 if not self._languages:
246 self._languages = DEFAULT_LANGUAGES.copy()
247
248 self._kids_mode = bool(self.get_setup_value(CONF_KIDS_MODE))
249
250 self._api = StorytelHelper(
251 session=self.mass.http_session,
252 provider_instance=self,
253 provider_id=self.instance_id,
254 provider_domain=self.domain,
255 kids_mode=self._kids_mode,
256 languages=self._languages,
257 )
258 try:
259 await self._api.login(username, password)
260 except LoginFailed:
261 raise SetupFailedError("Invalid Storytel username or password")
262 except ProviderUnavailableError as err:
263 raise SetupFailedError(f"Storytel login failed: {err}") from err
264 try:
265 await self._api.fetch_resource_version()
266 except ProviderUnavailableError as err:
267 self.logger.warning(
268 "Storytel resource version refresh failed during setup; using default: %s",
269 err,
270 )
271
272 @property
273 def is_streaming_provider(self) -> bool:
274 """Return True if the provider is a streaming provider."""
275 return True
276
277 # -------------------
278 # Library: Audiobooks
279 # -------------------
280
281 @handle_login_failed_generator
282 async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
283 """Yield audiobooks from the user's Storytel bookshelf."""
284 books, _ = await self.api.get_library()
285 failed_provider_error: ProviderUnavailableError | None = None
286 yielded_any = False
287 for consumable_id in books:
288 try:
289 audiobook = await self.get_audiobook(consumable_id)
290 yielded_any = True
291 yield audiobook
292 except (MediaNotFoundError, ProviderUnavailableError, InvalidDataError) as err:
293 if isinstance(err, ProviderUnavailableError):
294 failed_provider_error = failed_provider_error or err
295 self.report_skipped_sync_item(MediaType.AUDIOBOOK, consumable_id, err)
296
297 if not yielded_any and failed_provider_error is not None:
298 raise failed_provider_error
299
300 @handle_login_failed
301 async def get_audiobook(self, prov_audiobook_id: str, use_cache: bool = True) -> Audiobook:
302 """
303 Fetch a single audiobook by our provider id.
304
305 :param prov_audiobook_id: the provider specific ID of the audiobook.
306 :param use_cache: boolean to indicate if a cache lookup is allowed.
307 """
308 if use_cache:
309 cached_book = await self.mass.cache.get(
310 key=prov_audiobook_id,
311 provider=self.instance_id,
312 category=CACHE_CATEGORY_AUDIOBOOK,
313 default=None,
314 )
315 if cached_book is not None:
316 return Audiobook.from_dict(cached_book)
317 item_data = await self.api.get_consumable_details(prov_audiobook_id)
318
319 if item_data is None:
320 raise MediaNotFoundError(f"Storytel book not found: {prov_audiobook_id}")
321
322 media_item = await self.api.parse_media_item(item_data)
323
324 await self.mass.cache.set(
325 key=prov_audiobook_id,
326 provider=self.instance_id,
327 category=CACHE_CATEGORY_AUDIOBOOK,
328 data=media_item.to_dict(),
329 )
330 return cast("Audiobook", media_item)
331
332 # -------------------
333 # Streaming
334 # -------------------
335
336 @handle_login_failed
337 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
338 """
339 Fetch stream details for a given item id and media type.
340
341 :param item_id: the provider item id of the media.
342 :param media_type: the media type to stream (Audiobook or PodcastEpisode).
343 """
344 if media_type not in {MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE}:
345 self.logger.error("Unsupported media type for streaming: %s", media_type)
346 raise InvalidDataError("Only audiobooks and podcasts are supported")
347 return await self.api.get_stream_details(item_id, media_type)
348
349 @handle_login_failed
350 async def get_resume_position(
351 self, item_id: str, media_type: MediaType
352 ) -> tuple[bool, int, datetime | None]:
353 """
354 Get the resume position for a media item.
355
356 :param item_id: the provider item id.
357 :param media_type: the media type of the item.
358 """
359 if media_type not in {MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE}:
360 self.logger.error("Unsupported media type for resume position: %s", media_type)
361 raise InvalidDataError("Only audiobooks and podcasts are supported for resume position")
362 bm = await self.api.get_bookmark(item_id)
363 if not bm:
364 self.logger.debug("No bookmark found for %s", item_id)
365 return False, 0, None
366 # Storytel returns the position in milliseconds.
367 pos = int(bm.get("position") or 0)
368 updated_ts = bm.get("updatedTime")
369 bookmark_updated_dt: datetime | None = None
370 if updated_ts:
371 try:
372 bookmark_updated_dt = datetime.fromisoformat(str(updated_ts)).astimezone(UTC)
373 except ValueError:
374 bookmark_updated_dt = None
375
376 return False, pos, bookmark_updated_dt
377
378 @handle_login_failed
379 async def on_played(
380 self,
381 media_type: MediaType,
382 prov_item_id: str,
383 fully_played: bool,
384 position: int,
385 media_item: MediaItemType,
386 is_playing: bool = False,
387 ) -> None:
388 """
389 Handle played event of a media item.
390
391 :param media_type: the media type of the item.
392 :param prov_item_id: the provider item id.
393 :param fully_played: True if the item was fully played.
394 :param position: the resume position in the item.
395 :param media_item: the media item object.
396 :param is_playing: True if the item is currently playing.
397 """
398 consumable_id = prov_item_id
399 if media_type not in {MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE}:
400 self.logger.error("Unsupported media type for bookmark update: %s", media_type)
401 return
402 # Set bookmark position to the media duration if the item was fully played and the duration is known. Otherwise, use the provided position.
403 bookmark_position = position
404 if fully_played and isinstance(media_item, (Audiobook, PodcastEpisode)):
405 bookmark_position = media_item.duration
406 try:
407 await self.api.set_bookmark(
408 consumable_id,
409 bookmark_position,
410 kids_mode=self._kids_mode,
411 )
412 except LoginFailed:
413 raise
414 except ProviderUnavailableError as err:
415 self.logger.warning("Failed to update Storytel bookmark: %s", err)
416
417 @handle_login_failed
418 async def get_podcast(self, prov_podcast_id: str, use_cache: bool = True) -> Podcast:
419 """
420 Fetch a podcast by our provider id.
421
422 :param prov_podcast_id: the provider specific ID of the podcast.
423 :param use_cache: boolean to indicate if a cache lookup is allowed.
424 """
425 if use_cache:
426 cached_podcast = await self.mass.cache.get(
427 key=prov_podcast_id,
428 provider=self.instance_id,
429 category=CACHE_CATEGORY_PODCAST,
430 default=None,
431 )
432 if cached_podcast is not None:
433 return Podcast.from_dict(cached_podcast)
434 item_data = await self.api.get_podcast_details(prov_podcast_id)
435 if item_data is None:
436 raise MediaNotFoundError(f"Storytel podcast not found: {prov_podcast_id}")
437
438 podcast = self.api.parse_podcast(item_data)
439
440 await self.mass.cache.set(
441 key=prov_podcast_id,
442 provider=self.instance_id,
443 category=CACHE_CATEGORY_PODCAST,
444 data=podcast.to_dict(),
445 )
446 return podcast
447
448 @handle_login_failed
449 async def get_podcast_episode(
450 self, prov_episode_id: str, use_cache: bool = True
451 ) -> PodcastEpisode:
452 """
453 Fetch a podcast episode by our provider id.
454
455 :param prov_episode_id: the provider specific ID of the podcast episode.
456 :param use_cache: boolean to indicate if a cache lookup is allowed.
457 """
458 if use_cache:
459 cached_episode = await self.mass.cache.get(
460 key=prov_episode_id,
461 provider=self.instance_id,
462 category=CACHE_CATEGORY_PODCAST_EPISODE,
463 default=None,
464 )
465 if cached_episode is not None:
466 return PodcastEpisode.from_dict(cached_episode)
467 item_data = await self.api.get_consumable_details(prov_episode_id)
468
469 if item_data is None:
470 raise MediaNotFoundError(f"Storytel podcast episode not found: {prov_episode_id}")
471
472 podcast_episode = await self.api.parse_media_item(item_data)
473
474 await self.mass.cache.set(
475 key=prov_episode_id,
476 provider=self.instance_id,
477 category=CACHE_CATEGORY_PODCAST_EPISODE,
478 data=podcast_episode.to_dict(),
479 )
480 return cast("PodcastEpisode", podcast_episode)
481
482 @handle_login_failed_generator
483 async def get_podcast_episodes(
484 self, prov_podcast_id: str, use_cache: bool = True
485 ) -> AsyncGenerator[PodcastEpisode]:
486 """
487 Fetch all episodes for a given podcast by our provider id.
488
489 :param prov_podcast_id: the provider specific ID of the podcast.
490 :param use_cache: boolean to indicate if a cache lookup is allowed.
491 """
492 failed_provider_error: ProviderUnavailableError | None = None
493 episode_semaphore = Semaphore(10)
494
495 async def _fetch_episode(episode: dict[str, Any]) -> PodcastEpisode | None:
496 nonlocal failed_provider_error
497 consumable_id = str(episode.get("id") or "").strip()
498 if not consumable_id:
499 return None
500 async with episode_semaphore:
501 try:
502 return await self.get_podcast_episode(consumable_id, use_cache=True)
503 except (MediaNotFoundError, ProviderUnavailableError, InvalidDataError) as err:
504 if isinstance(err, ProviderUnavailableError):
505 failed_provider_error = failed_provider_error or err
506 self.logger.debug(
507 "Skipping Storytel podcast episode %s: %s", consumable_id, err
508 )
509 return None
510
511 if use_cache:
512 cached_episodes = await self.mass.cache.get(
513 key=prov_podcast_id,
514 provider=self.instance_id,
515 category=CACHE_CATEGORY_PODCAST_EPISODES,
516 default=None,
517 )
518 if cached_episodes is not None:
519 async with TaskGroup() as tg:
520 cached_tasks: list[Task[PodcastEpisode | None]] = []
521 for episode in cached_episodes:
522 cached_tasks.append(tg.create_task(_fetch_episode(episode)))
523 yielded_any = False
524 for task in cached_tasks:
525 if episode := task.result():
526 yielded_any = True
527 yield episode
528 if not yielded_any and failed_provider_error is not None:
529 raise failed_provider_error
530 return
531 podcast = await self.get_podcast(prov_podcast_id, use_cache=use_cache)
532 episode_languages = {lang for lang in self.api.languages_query.split(",") if lang}
533 podcast_languages: UniqueList[str] = podcast.metadata.languages or UniqueList()
534 episode_languages.update(
535 str(language).strip() for language in podcast_languages if str(language).strip()
536 )
537 podcast_episodes = await self.api.get_podcast_episodes(
538 prov_podcast_id,
539 total_episodes=podcast.total_episodes or 0,
540 include_languages=",".join(sorted(episode_languages)) or None,
541 )
542
543 if podcast_episodes is not None:
544 if use_cache:
545 await self.mass.cache.set(
546 key=prov_podcast_id,
547 provider=self.instance_id,
548 category=CACHE_CATEGORY_PODCAST_EPISODES,
549 data=podcast_episodes,
550 )
551 async with TaskGroup() as tg:
552 live_tasks: list[Task[PodcastEpisode | None]] = []
553 for episode in podcast_episodes:
554 live_tasks.append(tg.create_task(_fetch_episode(episode)))
555 yielded_any = False
556 for task in live_tasks:
557 if episode := task.result():
558 yielded_any = True
559 yield episode
560 if not yielded_any and failed_provider_error is not None:
561 raise failed_provider_error
562
563 @handle_login_failed_generator
564 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
565 """Yield podcasts from the user's library."""
566 _, podcasts = await self.api.get_library()
567 failed_provider_error: ProviderUnavailableError | None = None
568 yielded_any = False
569 for podcast_data in podcasts.values():
570 model = podcast_data.get("model") or {}
571 podcast_id = model.get("id") or ""
572 result_type = str(model.get("resultType") or podcast_data.get("resultType") or "")
573 if result_type and result_type.lower() != "podcast":
574 self.logger.debug(
575 "Skipping Storytel followed item %s with result type %s",
576 podcast_id,
577 result_type,
578 )
579 continue
580 try:
581 podcast = await self.get_podcast(podcast_id)
582 yielded_any = True
583 yield podcast
584 except (MediaNotFoundError, ProviderUnavailableError, InvalidDataError) as err:
585 if isinstance(err, ProviderUnavailableError):
586 failed_provider_error = failed_provider_error or err
587 self.report_skipped_sync_item(MediaType.PODCAST, podcast_id or None, err)
588
589 if not yielded_any and failed_provider_error is not None:
590 raise failed_provider_error
591
592 @handle_login_failed
593 async def search(
594 self, search_query: str, media_types: list[MediaType], limit: int = 10
595 ) -> SearchResults:
596 """
597 Perform a search on Storytel.
598
599 :param search_query: the search query.
600 :param media_types: the media types to search for.
601 :param limit: the maximum number of results per media type.
602 """
603 result = SearchResults()
604 task_audiobooks: Task[list[Audiobook]] | None = None
605 task_podcasts: Task[list[Podcast]] | None = None
606
607 async with TaskGroup() as tg:
608 if MediaType.AUDIOBOOK in media_types:
609 task_audiobooks = tg.create_task(
610 self.api.search_audiobooks(search_query, limit=limit)
611 )
612 if MediaType.PODCAST in media_types:
613 task_podcasts = tg.create_task(self.api.search_podcasts(search_query, limit=limit))
614
615 if MediaType.AUDIOBOOK in media_types and task_audiobooks:
616 result.audiobooks = task_audiobooks.result()
617
618 if MediaType.PODCAST in media_types and task_podcasts:
619 result.podcasts = task_podcasts.result()
620
621 return result
622
623 @handle_login_failed
624 async def get_recommendations(self) -> list[RecommendationFolder]:
625 """Return personalized recommendation folders without items."""
626 return await self._recommendation_rows_from_payload()
627
628 @handle_login_failed
629 async def get_recommendation_items(
630 self, item_id: str
631 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
632 """
633 Return the items for a single recommendation folder.
634
635 :param item_id: the item id of the recommendation row.
636 """
637 return await self._recommendation_items_from_payload(item_id)
638
639 @handle_login_failed
640 async def library_add(self, item: MediaItemType) -> bool:
641 """
642 Add an item to the provider library.
643
644 :param item: the media item to add.
645 :return: True if successful, False otherwise.
646 """
647 if item.media_type not in (MediaType.AUDIOBOOK, MediaType.PODCAST):
648 self.logger.error("Unsupported media type for library add: %s", item.media_type)
649 raise InvalidDataError(
650 "Only audiobooks and podcasts are supported for library management"
651 )
652 return await self.api.add_to_bookshelf(item.item_id, item)
653
654 @handle_login_failed
655 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
656 """
657 Remove an item from the provider library.
658
659 :param prov_item_id: the provider item id to remove.
660 :param media_type: the media type of the item.
661 :return: True if successful, False otherwise.
662 """
663 if media_type not in (MediaType.AUDIOBOOK, MediaType.PODCAST):
664 self.logger.error("Unsupported media type for library remove: %s", media_type)
665 raise InvalidDataError(
666 "Only audiobooks and podcasts are supported for library management"
667 )
668 return await self.api.remove_from_bookshelf(prov_item_id, media_type)
669
670 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
671 """Fetch the full Storytel recommendation payload with items."""
672 recommendations = await self.api.get_recommendations()
673 return recommendations or []
674