music-assistant-server

1"""Audible provider for Music Assistant, utilizing the audible library."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator, Sequence
7from datetime import datetime
8from logging import getLevelName
9from typing import TYPE_CHECKING, ClassVar, cast
10from urllib.parse import quote, unquote
11
12import audible
13from music_assistant_models.enums import MediaType, ProviderFeature
14from music_assistant_models.errors import LoginFailed, MediaNotFoundError
15from music_assistant_models.media_items import BrowseFolder, ItemMapping
16
17from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
18from music_assistant.models.music_provider import MusicProvider
19from music_assistant.providers.audible.audible_helper import (
20    AudibleHelper,
21    cached_authenticator_from_file,
22    refresh_access_token_compat,
23)
24
25if TYPE_CHECKING:
26    from music_assistant_models.config_entries import (
27        ConfigEntry,
28        ProviderConfig,
29    )
30    from music_assistant_models.media_items import (
31        Audiobook,
32        MediaItemType,
33        Podcast,
34        PodcastEpisode,
35    )
36    from music_assistant_models.provider import ProviderManifest
37    from music_assistant_models.streamdetails import StreamDetails
38
39    from music_assistant.mass import MusicAssistant
40    from music_assistant.models import ProviderInstanceType
41
42
43# Config keys collected by the setup flow and read back at runtime
44CONF_AUTH_FILE = "auth_file"
45CONF_LOCALE = "locale"
46
47SUPPORTED_FEATURES = {
48    ProviderFeature.BROWSE,
49    ProviderFeature.LIBRARY_AUDIOBOOKS,
50    ProviderFeature.LIBRARY_PODCASTS,
51}
52
53
54async def setup(
55    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
56) -> ProviderInstanceType:
57    """Initialize provider(instance) with given configuration."""
58    return Audibleprovider(mass, manifest, config, SUPPORTED_FEATURES)
59
60
61class Audibleprovider(MusicProvider):
62    """Implementation of a Audible Audiobook Provider."""
63
64    locale: str
65    auth_file: str
66    _client: audible.AsyncClient | None = None
67
68    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
69        """
70        Return the config entries for the Audible provider.
71
72        Authentication (Amazon sign-in on their own page + device registration) runs in the
73        interactive setup flow (see ``setup_flow.py``); this provider has no further options.
74        """
75        return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
76
77    async def handle_async_init(self) -> None:
78        """Handle asynchronous initialization of the provider."""
79        self.locale = cast("str", self.get_setup_value(CONF_LOCALE) or "us")
80        self.auth_file = cast("str", self.get_setup_value(CONF_AUTH_FILE))
81        self._client: audible.AsyncClient | None = None
82        audible.log_helper.set_level(getLevelName(self.logger.level))
83        await self._login()
84
85    # Cache for authenticators to avoid repeated file I/O
86    _AUTH_CACHE: ClassVar[dict[str, audible.Authenticator]] = {}
87
88    async def _login(self) -> None:
89        """Authenticate with Audible using the saved authentication file."""
90        try:
91            auth = self._AUTH_CACHE.get(self.instance_id)
92
93            if auth is None:
94                self.logger.debug("Loading authenticator from file")
95                auth = await cached_authenticator_from_file(self.auth_file)
96                self._AUTH_CACHE[self.instance_id] = auth
97            else:
98                self.logger.debug("Using cached authenticator")
99
100            # Check if we have signing auth (preferred, stable - not affected by API changes)
101            has_signing_auth = auth.adp_token and auth.device_private_key
102            if has_signing_auth:
103                self.logger.debug("Using signing auth (stable RSA-signed requests)")
104            else:
105                self.logger.debug("Signing auth not available, using bearer auth")
106
107            # Handle token refresh if needed
108            if auth.access_token_expired:
109                self.logger.debug("Access token expired, refreshing")
110                try:
111                    # Use compatible refresh that handles new API token format
112                    if auth.refresh_token and auth.locale:
113                        refresh_data = await refresh_access_token_compat(
114                            refresh_token=auth.refresh_token,
115                            domain=auth.locale.domain,
116                            http_session=self.mass.http_session,
117                            with_username=auth.with_username or False,
118                        )
119                        auth._update_attrs(**refresh_data)
120                        await asyncio.to_thread(auth.to_file, self.auth_file)
121                        self._AUTH_CACHE[self.instance_id] = auth
122                        self.logger.debug("Token refreshed successfully")
123                    else:
124                        self.logger.warning("Cannot refresh: missing refresh_token or locale")
125                except Exception as refresh_error:
126                    self.logger.warning(f"Token refresh failed: {refresh_error}")
127                    if not has_signing_auth:
128                        # Only fail if we don't have signing auth as fallback
129                        raise LoginFailed(
130                            "Token refresh failed and signing auth not available. "
131                            "Please re-authenticate with Audible."
132                        ) from refresh_error
133                    # Continue with signing auth
134
135            self._client = audible.AsyncClient(auth)
136
137            self.helper = AudibleHelper(
138                mass=self.mass,
139                client=self._client,
140                provider_instance=self.instance_id,
141                provider_domain=self.domain,
142                provider=self,
143                logger=self.logger,
144            )
145
146            self.logger.info("Successfully authenticated with Audible.")
147
148        except LoginFailed:
149            raise
150        except Exception as e:
151            self.logger.error(f"Failed to authenticate with Audible: {e}")
152            raise LoginFailed(f"Failed to authenticate with Audible: {e}") from e
153
154    @property
155    def is_streaming_provider(self) -> bool:
156        """Return True if the provider is a streaming provider."""
157        return True
158
159    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
160        """Get all audiobooks from the library."""
161        async for audiobook in self.helper.get_library():
162            yield audiobook
163
164    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
165        """Get full audiobook details by id."""
166        return await self.helper.get_audiobook(asin=prov_audiobook_id, use_cache=False)
167
168    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
169        """
170        Browse this provider's items.
171
172        :param path: The path to browse, (e.g. provider_id://authors).
173        """
174        item_path = path.split("://", 1)[1] if "://" in path else ""
175        parts = item_path.split("/") if item_path else []
176
177        # Root - return main folders
178        if not item_path:
179            return self._browse_root(path)
180
181        # Authors listing
182        if parts[0] == "authors":
183            if len(parts) == 1:
184                return await self._browse_authors(path)
185            # Specific author's books
186            return await self._browse_author_books(unquote(parts[1]))
187
188        # Series listing
189        if parts[0] == "series":
190            if len(parts) == 1:
191                return await self._browse_series(path)
192            # Specific series' books
193            return await self._browse_series_books(unquote(parts[1]))
194
195        # Narrators listing
196        if parts[0] == "narrators":
197            if len(parts) == 1:
198                return await self._browse_narrators(path)
199            return await self._browse_narrator_books(unquote(parts[1]))
200
201        # Genres listing
202        if parts[0] == "genres":
203            if len(parts) == 1:
204                return await self._browse_genres(path)
205            return await self._browse_genre_books(unquote(parts[1]))
206
207        # Publishers listing
208        if parts[0] == "publishers":
209            if len(parts) == 1:
210                return await self._browse_publishers(path)
211            return await self._browse_publisher_books(unquote(parts[1]))
212
213        # Fall back to base implementation for audiobooks/podcasts
214        return await super().browse(path)
215
216    def _browse_root(self, base_path: str) -> list[BrowseFolder]:
217        """Return root browse folders."""
218        return [
219            BrowseFolder(
220                item_id="audiobooks",
221                provider=self.instance_id,
222                path=f"{base_path}audiobooks",
223                name="Audiobooks",
224                translation_key="audiobooks",
225            ),
226            BrowseFolder(
227                item_id="podcasts",
228                provider=self.instance_id,
229                path=f"{base_path}podcasts",
230                name="Podcasts",
231                translation_key="podcasts",
232            ),
233            BrowseFolder(
234                item_id="authors",
235                provider=self.instance_id,
236                path=f"{base_path}authors",
237                name="Authors",
238                translation_key="authors",
239            ),
240            BrowseFolder(
241                item_id="series",
242                provider=self.instance_id,
243                path=f"{base_path}series",
244                name="Series",
245                translation_key="series",
246            ),
247            BrowseFolder(
248                item_id="narrators",
249                provider=self.instance_id,
250                path=f"{base_path}narrators",
251                name="Narrators",
252                translation_key="narrators",
253            ),
254            BrowseFolder(
255                item_id="genres",
256                provider=self.instance_id,
257                path=f"{base_path}genres",
258                name="Genres",
259                translation_key="genres",
260            ),
261            BrowseFolder(
262                item_id="publishers",
263                provider=self.instance_id,
264                path=f"{base_path}publishers",
265                name="Publishers",
266                translation_key="publishers",
267            ),
268        ]
269
270    async def _browse_authors(self, base_path: str) -> list[BrowseFolder]:
271        """Return list of all authors."""
272        authors = await self.helper.get_authors()
273        return [
274            BrowseFolder(
275                item_id=asin,
276                provider=self.instance_id,
277                path=f"{base_path}/{quote(asin)}",
278                name=name,
279            )
280            for asin, name in sorted(authors.items(), key=lambda x: x[1])
281        ]
282
283    async def _browse_author_books(self, author_asin: str) -> list[Audiobook]:
284        """Return audiobooks by a specific author."""
285        return await self.helper.get_audiobooks_by_author(author_asin)
286
287    async def _browse_series(self, base_path: str) -> list[BrowseFolder]:
288        """Return list of all series."""
289        series = await self.helper.get_series()
290        return [
291            BrowseFolder(
292                item_id=asin,
293                provider=self.instance_id,
294                path=f"{base_path}/{quote(asin)}",
295                name=title,
296            )
297            for asin, title in sorted(series.items(), key=lambda x: x[1])
298        ]
299
300    async def _browse_series_books(self, series_asin: str) -> list[Audiobook]:
301        """Return audiobooks in a specific series."""
302        return await self.helper.get_audiobooks_by_series(series_asin)
303
304    async def _browse_narrators(self, base_path: str) -> list[BrowseFolder]:
305        """Return list of all narrators."""
306        narrators = await self.helper.get_narrators()
307        return [
308            BrowseFolder(
309                item_id=asin,
310                provider=self.instance_id,
311                path=f"{base_path}/{quote(asin)}",
312                name=name,
313            )
314            for asin, name in sorted(narrators.items(), key=lambda x: x[1])
315        ]
316
317    async def _browse_narrator_books(self, narrator_asin: str) -> list[Audiobook]:
318        """Return audiobooks by a specific narrator."""
319        return await self.helper.get_audiobooks_by_narrator(narrator_asin)
320
321    async def _browse_genres(self, base_path: str) -> list[BrowseFolder]:
322        """Return list of all genres."""
323        genres = await self.helper.get_genres()
324        return [
325            BrowseFolder(
326                item_id=genre,
327                provider=self.instance_id,
328                path=f"{base_path}/{quote(genre)}",
329                name=genre,
330            )
331            for genre in sorted(genres)
332        ]
333
334    async def _browse_genre_books(self, genre: str) -> list[Audiobook]:
335        """Return audiobooks matching a genre."""
336        return await self.helper.get_audiobooks_by_genre(genre)
337
338    async def _browse_publishers(self, base_path: str) -> list[BrowseFolder]:
339        """Return list of all publishers."""
340        publishers = await self.helper.get_publishers()
341        return [
342            BrowseFolder(
343                item_id=publisher,
344                provider=self.instance_id,
345                path=f"{base_path}/{quote(publisher)}",
346                name=publisher,
347            )
348            for publisher in sorted(publishers)
349        ]
350
351    async def _browse_publisher_books(self, publisher: str) -> list[Audiobook]:
352        """Return audiobooks from a specific publisher."""
353        return await self.helper.get_audiobooks_by_publisher(publisher)
354
355    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
356        """Get all podcasts from the library."""
357        async for podcast in self.helper.get_library_podcasts():
358            yield podcast
359
360    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
361        """Get full podcast details by id."""
362        return await self.helper.get_podcast(asin=prov_podcast_id)
363
364    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
365        """Get all episodes for a podcast."""
366        async for episode in self.helper.get_podcast_episodes(prov_podcast_id):
367            yield episode
368
369    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
370        """Get full podcast episode details by id."""
371        return await self.helper.get_podcast_episode(prov_episode_id)
372
373    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
374        """
375        Get stream details for an audiobook or podcast episode.
376
377        :param item_id: The ASIN of the audiobook or podcast episode.
378        :param media_type: The type of media (audiobook or podcast episode).
379        """
380        try:
381            return await self.helper.get_stream(asin=item_id, media_type=media_type)
382        except ValueError as exc:
383            raise MediaNotFoundError(f"Failed to get stream details for {item_id}") from exc
384
385    async def on_played(
386        self,
387        media_type: MediaType,
388        prov_item_id: str,
389        fully_played: bool,
390        position: int,
391        media_item: MediaItemType,
392        is_playing: bool = False,
393    ) -> None:
394        """
395        Handle callback when a (playable) media item has been played.
396
397        This is called by the Queue controller when;
398            - a track has been fully played
399            - a track has been stopped (or skipped) after being played
400            - every 30s when a track is playing
401
402        Fully played is True when the track has been played to the end.
403
404        Position is the last known position of the track in seconds, to sync resume state.
405        When fully_played is set to false and position is 0,
406        the user marked the item as unplayed in the UI.
407
408        is_playing is True when the track is currently playing.
409
410        media_item is the full media item details of the played/playing track.
411        """
412        await self.helper.set_last_position(prov_item_id, position, media_type)
413
414    async def get_resume_position(
415        self, item_id: str, media_type: MediaType
416    ) -> tuple[bool, int, datetime | None]:
417        """
418        Return the resume position from Audible for the given item.
419
420        :param item_id: The provider item ID (ASIN) of the audiobook.
421        :param media_type: The media type of the item.
422        """
423        if media_type != MediaType.AUDIOBOOK:
424            raise NotImplementedError
425        return await self.helper.get_audible_resume_position(item_id)
426
427    async def unload(self, is_removed: bool = False) -> None:
428        """
429        Handle unload/close of the provider.
430
431        Called when provider is deregistered (e.g. MA exiting or config reloading).
432        is_removed will be set to True when the provider is removed from the configuration.
433        """
434        if is_removed:
435            await self.helper.deregister()
436