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                logger=self.logger,
143            )
144
145            self.logger.info("Successfully authenticated with Audible.")
146
147        except LoginFailed:
148            raise
149        except Exception as e:
150            self.logger.error(f"Failed to authenticate with Audible: {e}")
151            raise LoginFailed(f"Failed to authenticate with Audible: {e}") from e
152
153    @property
154    def is_streaming_provider(self) -> bool:
155        """Return True if the provider is a streaming provider."""
156        return True
157
158    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
159        """Get all audiobooks from the library."""
160        async for audiobook in self.helper.get_library():
161            yield audiobook
162
163    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
164        """Get full audiobook details by id."""
165        return await self.helper.get_audiobook(asin=prov_audiobook_id, use_cache=False)
166
167    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
168        """
169        Browse this provider's items.
170
171        :param path: The path to browse, (e.g. provider_id://authors).
172        """
173        item_path = path.split("://", 1)[1] if "://" in path else ""
174        parts = item_path.split("/") if item_path else []
175
176        # Root - return main folders
177        if not item_path:
178            return self._browse_root(path)
179
180        # Authors listing
181        if parts[0] == "authors":
182            if len(parts) == 1:
183                return await self._browse_authors(path)
184            # Specific author's books
185            return await self._browse_author_books(unquote(parts[1]))
186
187        # Series listing
188        if parts[0] == "series":
189            if len(parts) == 1:
190                return await self._browse_series(path)
191            # Specific series' books
192            return await self._browse_series_books(unquote(parts[1]))
193
194        # Narrators listing
195        if parts[0] == "narrators":
196            if len(parts) == 1:
197                return await self._browse_narrators(path)
198            return await self._browse_narrator_books(unquote(parts[1]))
199
200        # Genres listing
201        if parts[0] == "genres":
202            if len(parts) == 1:
203                return await self._browse_genres(path)
204            return await self._browse_genre_books(unquote(parts[1]))
205
206        # Publishers listing
207        if parts[0] == "publishers":
208            if len(parts) == 1:
209                return await self._browse_publishers(path)
210            return await self._browse_publisher_books(unquote(parts[1]))
211
212        # Fall back to base implementation for audiobooks/podcasts
213        return await super().browse(path)
214
215    def _browse_root(self, base_path: str) -> list[BrowseFolder]:
216        """Return root browse folders."""
217        return [
218            BrowseFolder(
219                item_id="audiobooks",
220                provider=self.instance_id,
221                path=f"{base_path}audiobooks",
222                name="Audiobooks",
223                translation_key="audiobooks",
224            ),
225            BrowseFolder(
226                item_id="podcasts",
227                provider=self.instance_id,
228                path=f"{base_path}podcasts",
229                name="Podcasts",
230                translation_key="podcasts",
231            ),
232            BrowseFolder(
233                item_id="authors",
234                provider=self.instance_id,
235                path=f"{base_path}authors",
236                name="Authors",
237                translation_key="authors",
238            ),
239            BrowseFolder(
240                item_id="series",
241                provider=self.instance_id,
242                path=f"{base_path}series",
243                name="Series",
244                translation_key="series",
245            ),
246            BrowseFolder(
247                item_id="narrators",
248                provider=self.instance_id,
249                path=f"{base_path}narrators",
250                name="Narrators",
251                translation_key="narrators",
252            ),
253            BrowseFolder(
254                item_id="genres",
255                provider=self.instance_id,
256                path=f"{base_path}genres",
257                name="Genres",
258                translation_key="genres",
259            ),
260            BrowseFolder(
261                item_id="publishers",
262                provider=self.instance_id,
263                path=f"{base_path}publishers",
264                name="Publishers",
265                translation_key="publishers",
266            ),
267        ]
268
269    async def _browse_authors(self, base_path: str) -> list[BrowseFolder]:
270        """Return list of all authors."""
271        authors = await self.helper.get_authors()
272        return [
273            BrowseFolder(
274                item_id=asin,
275                provider=self.instance_id,
276                path=f"{base_path}/{quote(asin)}",
277                name=name,
278            )
279            for asin, name in sorted(authors.items(), key=lambda x: x[1])
280        ]
281
282    async def _browse_author_books(self, author_asin: str) -> list[Audiobook]:
283        """Return audiobooks by a specific author."""
284        return await self.helper.get_audiobooks_by_author(author_asin)
285
286    async def _browse_series(self, base_path: str) -> list[BrowseFolder]:
287        """Return list of all series."""
288        series = await self.helper.get_series()
289        return [
290            BrowseFolder(
291                item_id=asin,
292                provider=self.instance_id,
293                path=f"{base_path}/{quote(asin)}",
294                name=title,
295            )
296            for asin, title in sorted(series.items(), key=lambda x: x[1])
297        ]
298
299    async def _browse_series_books(self, series_asin: str) -> list[Audiobook]:
300        """Return audiobooks in a specific series."""
301        return await self.helper.get_audiobooks_by_series(series_asin)
302
303    async def _browse_narrators(self, base_path: str) -> list[BrowseFolder]:
304        """Return list of all narrators."""
305        narrators = await self.helper.get_narrators()
306        return [
307            BrowseFolder(
308                item_id=asin,
309                provider=self.instance_id,
310                path=f"{base_path}/{quote(asin)}",
311                name=name,
312            )
313            for asin, name in sorted(narrators.items(), key=lambda x: x[1])
314        ]
315
316    async def _browse_narrator_books(self, narrator_asin: str) -> list[Audiobook]:
317        """Return audiobooks by a specific narrator."""
318        return await self.helper.get_audiobooks_by_narrator(narrator_asin)
319
320    async def _browse_genres(self, base_path: str) -> list[BrowseFolder]:
321        """Return list of all genres."""
322        genres = await self.helper.get_genres()
323        return [
324            BrowseFolder(
325                item_id=genre,
326                provider=self.instance_id,
327                path=f"{base_path}/{quote(genre)}",
328                name=genre,
329            )
330            for genre in sorted(genres)
331        ]
332
333    async def _browse_genre_books(self, genre: str) -> list[Audiobook]:
334        """Return audiobooks matching a genre."""
335        return await self.helper.get_audiobooks_by_genre(genre)
336
337    async def _browse_publishers(self, base_path: str) -> list[BrowseFolder]:
338        """Return list of all publishers."""
339        publishers = await self.helper.get_publishers()
340        return [
341            BrowseFolder(
342                item_id=publisher,
343                provider=self.instance_id,
344                path=f"{base_path}/{quote(publisher)}",
345                name=publisher,
346            )
347            for publisher in sorted(publishers)
348        ]
349
350    async def _browse_publisher_books(self, publisher: str) -> list[Audiobook]:
351        """Return audiobooks from a specific publisher."""
352        return await self.helper.get_audiobooks_by_publisher(publisher)
353
354    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
355        """Get all podcasts from the library."""
356        async for podcast in self.helper.get_library_podcasts():
357            yield podcast
358
359    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
360        """Get full podcast details by id."""
361        return await self.helper.get_podcast(asin=prov_podcast_id)
362
363    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
364        """Get all episodes for a podcast."""
365        async for episode in self.helper.get_podcast_episodes(prov_podcast_id):
366            yield episode
367
368    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
369        """Get full podcast episode details by id."""
370        return await self.helper.get_podcast_episode(prov_episode_id)
371
372    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
373        """
374        Get stream details for an audiobook or podcast episode.
375
376        :param item_id: The ASIN of the audiobook or podcast episode.
377        :param media_type: The type of media (audiobook or podcast episode).
378        """
379        try:
380            return await self.helper.get_stream(asin=item_id, media_type=media_type)
381        except ValueError as exc:
382            raise MediaNotFoundError(f"Failed to get stream details for {item_id}") from exc
383
384    async def on_played(
385        self,
386        media_type: MediaType,
387        prov_item_id: str,
388        fully_played: bool,
389        position: int,
390        media_item: MediaItemType,
391        is_playing: bool = False,
392    ) -> None:
393        """
394        Handle callback when a (playable) media item has been played.
395
396        This is called by the Queue controller when;
397            - a track has been fully played
398            - a track has been stopped (or skipped) after being played
399            - every 30s when a track is playing
400
401        Fully played is True when the track has been played to the end.
402
403        Position is the last known position of the track in seconds, to sync resume state.
404        When fully_played is set to false and position is 0,
405        the user marked the item as unplayed in the UI.
406
407        is_playing is True when the track is currently playing.
408
409        media_item is the full media item details of the played/playing track.
410        """
411        await self.helper.set_last_position(prov_item_id, position, media_type)
412
413    async def get_resume_position(
414        self, item_id: str, media_type: MediaType
415    ) -> tuple[bool, int, datetime | None]:
416        """
417        Return the resume position from Audible for the given item.
418
419        :param item_id: The provider item ID (ASIN) of the audiobook.
420        :param media_type: The media type of the item.
421        """
422        if media_type != MediaType.AUDIOBOOK:
423            raise NotImplementedError
424        return await self.helper.get_audible_resume_position(item_id)
425
426    async def unload(self, is_removed: bool = False) -> None:
427        """
428        Handle unload/close of the provider.
429
430        Called when provider is deregistered (e.g. MA exiting or config reloading).
431        is_removed will be set to True when the provider is removed from the configuration.
432        """
433        if is_removed:
434            await self.helper.deregister()
435