music-assistant-server

1"""Audiobookshelf (abs) provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import functools
7import itertools
8import time
9from collections.abc import AsyncGenerator, Callable, Coroutine, Sequence
10from contextlib import suppress
11from datetime import datetime
12from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
13
14import aioaudiobookshelf as aioabs
15from aioaudiobookshelf.client.session_configuration import (
16    SessionConfiguration as AbsSessionConfiguration,
17)
18from aioaudiobookshelf.exceptions import AbsError, RefreshTokenExpiredError
19from aioaudiobookshelf.exceptions import (
20    LoginError as AbsLoginError,
21)
22from aioaudiobookshelf.exceptions import (
23    NotFoundError as AbsNotFoundError,
24)
25from aioaudiobookshelf.exceptions import SessionNotFoundError as AbsSessionNotFoundError
26from aioaudiobookshelf.exceptions import (
27    SessionSyncError as AbsSessionSyncError,
28)
29from aioaudiobookshelf.schema.author import AuthorExpanded
30from aioaudiobookshelf.schema.calls_authors import (
31    AuthorWithItemsAndSeries as AbsAuthorWithItemsAndSeries,
32)
33from aioaudiobookshelf.schema.calls_items import (
34    PlaybackSessionParameters as AbsPlaybackSessionParameters,
35)
36from aioaudiobookshelf.schema.calls_playlists import (
37    CreatePlaylistParameters as AbsCreatePlaylistParameters,
38)
39from aioaudiobookshelf.schema.calls_series import SeriesWithProgress as AbsSeriesWithProgress
40from aioaudiobookshelf.schema.calls_session import SyncOpenSessionParameters
41from aioaudiobookshelf.schema.library import (
42    LibraryItemExpanded,
43    LibraryItemExpandedBook,
44    LibraryItemExpandedPodcast,
45    LibraryItemMinifiedPodcast,
46)
47from aioaudiobookshelf.schema.library import LibraryItemExpandedBook as AbsLibraryItemExpandedBook
48from aioaudiobookshelf.schema.library import (
49    LibraryItemExpandedPodcast as AbsLibraryItemExpandedPodcast,
50)
51from aioaudiobookshelf.schema.library import LibraryMediaType as AbsLibraryMediaType
52from aioaudiobookshelf.schema.playlist import PlaylistExpanded as AbsPlaylistExpanded
53from aioaudiobookshelf.schema.playlist import PlaylistItem as AbsPlaylistItem
54from aioaudiobookshelf.schema.playlist import (
55    PlaylistItemExpandedBook as AbsPlaylistItemExpandedBook,
56)
57from aioaudiobookshelf.schema.playlist import (
58    PlaylistItemExpandedPodcast as AbsPlaylistItemExpandedPodcast,
59)
60from aioaudiobookshelf.schema.session import DeviceInfo as AbsDeviceInfo
61from aioaudiobookshelf.schema.session import PlaybackSessionExpanded as AbsPlaybackSessionExpanded
62from aioaudiobookshelf.schema.shelf import (
63    LibraryItemMinifiedPodcast as ShelfLibraryItemMinifiedPodcast,
64)
65from aioaudiobookshelf.schema.shelf import (
66    SeriesShelf,
67    ShelfAuthors,
68    ShelfBook,
69    ShelfEpisode,
70    ShelfLibraryItemMinified,
71    ShelfPodcast,
72    ShelfSeries,
73)
74from aioaudiobookshelf.schema.shelf import ShelfId as AbsShelfId
75from aioaudiobookshelf.schema.shelf import ShelfType as AbsShelfType
76from aiohttp import web
77from music_assistant_models.config_entries import (
78    ConfigEntry,
79    ProviderConfig,
80)
81from music_assistant_models.enums import (
82    ArtistType,
83    ConfigEntryType,
84    ContentType,
85    MediaType,
86    ProviderFeature,
87    StreamType,
88)
89from music_assistant_models.errors import InvalidDataError, LoginFailed, MediaNotFoundError
90from music_assistant_models.media_items import (
91    Artist,
92    Audiobook,
93    AudioFormat,
94    BrowseFolder,
95    ItemMapping,
96    MediaItemType,
97    Playlist,
98    PodcastEpisode,
99    UniqueList,
100)
101from music_assistant_models.media_items.media_item import RecommendationFolder
102from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
103
104from music_assistant.constants import PLAYBACK_REPORT_INTERVAL_SECONDS, PlaylistPlayableItem
105from music_assistant.helpers.datetime import from_utc_timestamp
106from music_assistant.models.music_provider import MusicProvider
107from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
108from music_assistant.providers.audiobookshelf.parsers import (
109    parse_audiobook,
110    parse_author,
111    parse_narrator,
112    parse_playlist,
113    parse_podcast,
114    parse_podcast_episode,
115)
116
117from .constants import (
118    ABS_BROWSE_ITEMS_BOOK_TO_PATH,
119    ABS_BROWSE_ITEMS_PODCAST_TO_PATH,
120    ABS_SHELF_ID_ICONS,
121    ABS_SHELF_ID_TRANSLATION_KEY,
122    AIOHTTP_TIMEOUT,
123    CACHE_CATEGORY_LIBRARIES,
124    CACHE_KEY_LIBRARIES,
125    CONF_API_TOKEN,
126    CONF_HIDE_EMPTY_PODCASTS,
127    CONF_OLD_TOKEN,
128    CONF_PASSWORD,
129    CONF_URL,
130    CONF_USERNAME,
131    CONF_VERIFY_SSL,
132    STREAMDETAILS_EXPIRATION_S,
133    AbsBrowseItemsBookTranslationKey,
134    AbsBrowseItemsPodcastTranslationKey,
135    AbsBrowsePaths,
136)
137from .helpers import LibrariesHelper, LibraryHelper, NarratorHelper, ProgressGuard, SessionHelper
138
139if TYPE_CHECKING:
140    from aioaudiobookshelf.schema.events_socket import LibraryItemRemoved
141    from aioaudiobookshelf.schema.media_progress import MediaProgress
142    from aioaudiobookshelf.schema.user import User
143    from music_assistant_models.media_items import Podcast
144    from music_assistant_models.provider import ProviderManifest
145
146    from music_assistant.mass import MusicAssistant
147    from music_assistant.models import ProviderInstanceType
148
149SUPPORTED_FEATURES = {
150    ProviderFeature.LIBRARY_PODCASTS,
151    ProviderFeature.LIBRARY_AUDIOBOOKS,
152    ProviderFeature.LIBRARY_PLAYLISTS,
153    ProviderFeature.LIBRARY_ARTISTS,  # authors/ narrators
154    ProviderFeature.BROWSE,
155    ProviderFeature.RECOMMENDATIONS,
156}
157
158
159async def setup(
160    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
161) -> ProviderInstanceType:
162    """Initialize provider(instance) with given configuration."""
163    return Audiobookshelf(mass, manifest, config, SUPPORTED_FEATURES)
164
165
166R = TypeVar("R")
167P = ParamSpec("P")
168
169
170class Audiobookshelf(RecommendationPayloadMixin, MusicProvider):
171    """Audiobookshelf MusicProvider."""
172
173    _on_unload_callbacks: list[Callable[[], None]]
174
175    def __init__(
176        self,
177        mass: MusicAssistant,
178        manifest: ProviderManifest,
179        config: ProviderConfig,
180        supported_features: set[ProviderFeature] | None = None,
181    ) -> None:
182        """Initialize the Audiobookshelf provider."""
183        super().__init__(mass, manifest, config, supported_features)
184        self.libraries = LibrariesHelper()
185
186    @staticmethod
187    def handle_refresh_token(
188        method: Callable[P, Coroutine[Any, Any, R]],
189    ) -> Callable[P, Coroutine[Any, Any, R]]:
190        """Decorate a method to handle an expired refresh token by relogin."""
191
192        @functools.wraps(method)
193        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
194            self = cast("Audiobookshelf", args[0])
195            try:
196                return await method(*args, **kwargs)
197            except RefreshTokenExpiredError:
198                self.logger.debug("Refresh token expired. Trying to renew.")
199                await self.reauthenticate()
200                return await method(*args, **kwargs)
201
202        return wrapper
203
204    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
205        """Return Config entries to setup this provider."""
206        return (
207            ConfigEntry(
208                key=CONF_HIDE_EMPTY_PODCASTS,
209                type=ConfigEntryType.BOOLEAN,
210                required=False,
211                advanced=True,
212                default_value=False,
213            ),
214        )
215
216    async def handle_async_init(self) -> None:
217        """Pass config values to client and initialize."""
218        self._on_unload_callbacks: list[Callable[[], None]] = []
219        self.sessions: dict[str, SessionHelper] = {}  # key is the mass_item_id
220        self.create_session_lock = asyncio.Lock()
221        base_url = str(self.get_setup_value(CONF_URL))
222        username = str(self.get_setup_value(CONF_USERNAME))
223        password = str(self.get_setup_value(CONF_PASSWORD))
224        token_old = self.get_setup_value(CONF_OLD_TOKEN)
225        token_api = self.get_setup_value(CONF_API_TOKEN)
226        verify_ssl = bool(self.get_setup_value(CONF_VERIFY_SSL))
227        session_config = AbsSessionConfiguration(
228            session=self.mass.http_session,
229            url=base_url,
230            verify_ssl=verify_ssl,
231            logger=self.logger,
232            pagination_items_per_page=30,  # audible provider goes with 50 for pagination
233            timeout=AIOHTTP_TIMEOUT,
234        )
235        # If we are configured with a non-expiring API key or not.
236        self.is_token_user = False
237        try:
238            if token_api is not None or token_old is not None:
239                _token = token_api if token_api is not None else token_old
240                session_config.token = str(_token)
241                (
242                    self._client,
243                    self._client_socket,
244                ) = await aioabs.get_user_and_socket_client_by_token(session_config=session_config)
245                self.is_token_user = True
246            else:
247                self._client, self._client_socket = await aioabs.get_user_and_socket_client(
248                    session_config=session_config, username=username, password=password
249                )
250            await self._client_socket.init_client()
251        except AbsLoginError as exc:
252            raise LoginFailed(
253                f"Login to abs instance at {base_url} failed.",
254                translation_key="login_failed",
255                translation_owner=self.translation_owner,
256                translation_args=[base_url],
257            ) from exc
258
259        if token_old is not None and token_api is None:
260            # Log Message that the old token won't work
261            _version = self._client.server_settings.version.split(".")
262            if len(_version) >= 2:
263                try:
264                    major, minor = int(_version[0]), int(_version[1])
265                except ValueError:
266                    major = minor = 0
267                if major >= 2 and minor >= 26:
268                    self.logger.warning(
269                        """
270
271######## Audiobookshelf API key change #############################################################
272
273Audiobookshelf introduced a new API key system in version 2.26 (JWT).
274You are still using a token configured with a previous version of Audiobookshelf,
275but you are running version %s. This will stop working in a future Audiobookshelf release.
276Please create a non-expiring API Key instead, and update your configuration accordingly.
277Refer to the documentation of Audiobookshelf, https://www.audiobookshelf.org/guides/api-keys/
278and of Music Assistant https://www.music-assistant.io/music-providers/audiobookshelf/
279for more details.
280
281""",
282                        self._client.server_settings.version,
283                    )
284
285        cached_libraries = await self.mass.cache.get(
286            key=CACHE_KEY_LIBRARIES,
287            provider=self.instance_id,
288            category=CACHE_CATEGORY_LIBRARIES,
289            default=None,
290        )
291        if cached_libraries is None:
292            self.libraries = LibrariesHelper()
293        else:
294            self.libraries = LibrariesHelper.from_dict(cached_libraries)
295
296        libraries = await self._client.get_all_libraries()
297        if libraries:
298            self._sync_library_keys(libraries)
299
300        # cache username
301        self.abs_username = (await self._client.get_my_user()).username
302
303        # set socket callbacks
304        self._client_socket.set_item_callbacks(
305            on_item_added=self._socket_abs_item_changed,
306            on_item_updated=self._socket_abs_item_changed,
307            on_item_removed=self._socket_abs_item_removed,
308            on_items_added=self._socket_abs_item_changed,
309            on_items_updated=self._socket_abs_item_changed,
310        )
311
312        self._client_socket.set_user_callbacks(
313            on_user_item_progress_updated=self._socket_abs_user_item_progress_updated,
314        )
315
316        self._client_socket.set_refresh_token_expired_callback(
317            on_refresh_token_expired=self._socket_abs_refresh_token_expired
318        )
319
320        self._client_socket.set_playlist_callbacks(
321            on_playlist_added=self._socket_abs_playlist_changed,
322            on_playlist_updated=self._socket_abs_playlist_changed,
323            on_playlist_removed=self._socket_abs_playlist_removed,
324        )
325
326        # progress guard
327        self.progress_guard = ProgressGuard()
328
329        # safe guard reauthentication
330        self.reauthenticate_lock = asyncio.Lock()
331        self.reauthenticate_last = 0.0
332
333        # safe guard playlist updates
334        self.playlist_lock = asyncio.Lock()
335        self.playlist_last = 0.0
336
337        # create close sessions task
338        self._close_sessions_task = self.mass.create_task(self._cleanup_open_sessions_loop())
339
340        # register dynamic stream route for audiobook parts
341        self._on_unload_callbacks.append(
342            self.mass.streams.register_dynamic_route(
343                f"/{self.instance_id}_part_stream", self._handle_session_part_request
344            )
345        )
346
347    async def unload(self, is_removed: bool = False) -> None:
348        """
349        Handle unload/close of the provider.
350
351        Called when provider is deregistered (e.g. MA exiting or config reloading).
352        is_removed will be set to True when the provider is removed from the configuration.
353        """
354        # run the unload chain first: RecommendationPayloadMixin cancels and awaits its
355        # payload tasks, so no fetch is still running against the clients logging out below
356        await super().unload(is_removed)
357
358        # cancel close sessions task, and close remaining
359        if self._close_sessions_task:
360            self._close_sessions_task.cancel()
361            with suppress(asyncio.CancelledError):
362                await self._close_sessions_task
363
364        # close the tracked sessions concurrently, so an unreachable server can neither
365        # stall nor abort the unload below
366        await asyncio.gather(
367            *(
368                self._client.close_open_session(session_id=x.abs_session_id)
369                for x in self.sessions.values()
370            ),
371            return_exceptions=True,
372        )
373        self.sessions.clear()
374        try:
375            await self._client.logout()
376            await self._client_socket.logout()
377        except AbsError as err:
378            self.logger.debug("Ignoring error during logout: %s", err)
379        for callback in self._on_unload_callbacks:
380            callback()
381
382    @property
383    def is_streaming_provider(self) -> bool:
384        """Return True if the provider is a streaming provider."""
385        # For streaming providers return True here but for local file based providers return False.
386        return False
387
388    @property
389    def supported_features(self) -> set[ProviderFeature]:
390        """
391        Get supported features.
392
393        ABS supports multiple libraries, but they must be of the same media type. If we only
394        have a single library of a media type, mapping the playlist creation is unambiguous.
395        """
396        features = SUPPORTED_FEATURES.copy()
397        if len(self.libraries.audiobooks) > 1 or len(self.libraries.podcasts) > 1:
398            return features
399        features.add(ProviderFeature.PLAYLIST_TRACKS_EDIT)
400        features.add(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
401        if len(self.libraries.audiobooks) == 1:
402            features.add(ProviderFeature.PLAYLIST_CREATE_AUDIOBOOKS)
403        if len(self.libraries.podcasts) == 1:
404            features.add(ProviderFeature.PLAYLIST_CREATE_PODCAST_EPISODES)
405        return features
406
407    @property
408    def supported_artist_types(self) -> set[ArtistType]:
409        """Supported artist types."""
410        return {ArtistType.AUTHOR, ArtistType.NARRATOR}
411
412    @handle_refresh_token
413    async def sync_library(self, media_type: MediaType) -> None:
414        """Obtain audiobook library ids and podcast library ids."""
415        if media_type == MediaType.AUDIOBOOK:
416            self.libraries.audiobooks.clear()
417            self.libraries.audiobook_narrators.clear()
418        elif media_type == MediaType.PODCAST:
419            self.libraries.podcasts.clear()
420        elif media_type == MediaType.PLAYLIST:
421            self.libraries.playlists_audiobooks.clear()
422            self.libraries.playlists_podcasts.clear()
423        elif media_type == MediaType.ARTIST:
424            self.libraries.authors.clear()
425            self.libraries.narrators.clear()
426
427        libraries = await self._client.get_all_libraries()
428        if len(libraries) == 0:
429            self._log_no_libraries()
430        for library in libraries:
431            if library.media_type == AbsLibraryMediaType.BOOK and media_type == MediaType.AUDIOBOOK:
432                self.libraries.audiobooks[library.id_] = LibraryHelper(name=library.name)
433                await self._update_book_narrators(library.id_)
434            elif (
435                library.media_type == AbsLibraryMediaType.PODCAST
436                and media_type == MediaType.PODCAST
437            ):
438                self.libraries.podcasts[library.id_] = LibraryHelper(name=library.name)
439            elif media_type == MediaType.PLAYLIST:
440                if library.media_type == AbsLibraryMediaType.PODCAST:
441                    self.libraries.playlists_podcasts[library.id_] = set()
442                if library.media_type == AbsLibraryMediaType.BOOK:
443                    self.libraries.playlists_audiobooks[library.id_] = set()
444            elif library.media_type == AbsLibraryMediaType.BOOK and media_type == MediaType.ARTIST:
445                self.libraries.narrators[library.id_] = set()
446                self.libraries.authors[library.id_] = set()
447
448        await super().sync_library(media_type)
449        await self._cache_set_helper_libraries()
450
451        # update playlog
452        user = await self._client.get_my_user()
453        await self._set_playlog_from_user(user)
454
455    async def get_library_artists(self) -> AsyncGenerator[Artist]:
456        """Get authors and narrators."""
457        libraries = await self._client.get_all_libraries()
458        library_ids_audiobook: set[str] = set()
459        for library in libraries:
460            if library.media_type == AbsLibraryMediaType.BOOK:
461                library_ids_audiobook.add(library.id_)
462        for book_lib_id in library_ids_audiobook:
463            for abs_author in await self._client.get_library_authors(library_id=book_lib_id):
464                self.libraries.authors[book_lib_id].add(abs_author.id_)
465                yield parse_author(
466                    abs_author=abs_author,
467                    instance_id=self.instance_id,
468                    domain=self.domain,
469                    token=self._client.token,
470                    base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
471                )
472            for abs_narrator in await self._client.get_library_narrators(library_id=book_lib_id):
473                self.libraries.narrators[book_lib_id].add(abs_narrator.id_)
474                yield parse_narrator(
475                    abs_narrator=abs_narrator, instance_id=self.instance_id, domain=self.domain
476                )
477
478    async def get_artist(self, prov_artist_id: str) -> Artist:
479        """Get an author or narrator."""
480        for library_id, narrator_ids in self.libraries.narrators.items():
481            if prov_artist_id in narrator_ids:
482                for abs_narrator in await self._client.get_library_narrators(library_id=library_id):
483                    if abs_narrator.id_ == prov_artist_id:
484                        return parse_narrator(
485                            abs_narrator=abs_narrator,
486                            instance_id=self.instance_id,
487                            domain=self.domain,
488                        )
489
490        return parse_author(
491            abs_author=await self._client.get_author(author_id=prov_artist_id),
492            instance_id=self.instance_id,
493            domain=self.domain,
494            token=self._client.token,
495            base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
496        )
497
498    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
499        """Retrieve playlists from abs."""
500        for playlist_dict, media_type in zip(
501            [
502                self.libraries.playlists_audiobooks,
503                self.libraries.playlists_podcasts,
504            ],
505            [MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE],
506            strict=True,
507        ):
508            for library_id in playlist_dict:
509                async for response in self._client.get_library_playlists(library_id=library_id):
510                    if not response.results:
511                        break
512                    for abs_playlist in response.results:
513                        playlist_dict[library_id].add(abs_playlist.id_)
514                        yield parse_playlist(
515                            abs_playlist=abs_playlist,
516                            instance_id=self.instance_id,
517                            domain=self.domain,
518                            token=self._client.token,
519                            base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
520                            owner=self.abs_username,
521                            media_type=media_type,
522                        )
523
524    @handle_refresh_token
525    async def get_playlist_tracks(
526        self, prov_playlist_id: str, page: int = 0
527    ) -> list[PlaylistPlayableItem]:
528        """Get playlist items."""
529        if page > 0:
530            # no pages in abs' playlist items api
531            return []
532        playlist_items: list[PlaylistPlayableItem] = []
533        try:
534            playlist = await self._client.get_playlist(playlist_id=prov_playlist_id)
535        except AbsNotFoundError:
536            # this is an edge case - abs deletes the playlist automatically, when
537            # the last item is removed, but the frontend then still asks for tracks.
538            # Due to our guard, we also block playlist removal via a socket update, so we can
539            # do that here
540            if ma_playlist := await self.mass.music.get_library_item_by_prov_id(
541                media_type=MediaType.PLAYLIST,
542                item_id=prov_playlist_id,
543                provider_instance_id_or_domain=self.instance_id,
544            ):
545                self.logger.debug(
546                    "Removing a playlist with no tracks from MA library, %s", ma_playlist.name
547                )
548                await self.mass.music.remove_item_from_library(
549                    media_type=MediaType.PLAYLIST, library_item_id=ma_playlist.item_id
550                )
551            return []
552        for item in playlist.items:
553            if isinstance(item, AbsPlaylistItemExpandedBook):
554                progress = await self._client.get_my_media_progress(item_id=item.library_item.id_)
555                playlist_items.append(
556                    parse_audiobook(
557                        abs_audiobook=item.library_item,
558                        instance_id=self.instance_id,
559                        audiobook_narrators=await self._get_audiobook_narrators(item.library_item),
560                        domain=self.domain,
561                        token=self._client.token,
562                        media_progress=progress,
563                        base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
564                    )
565                )
566            elif isinstance(item, AbsPlaylistItemExpandedPodcast):
567                progress = await self._client.get_my_media_progress(
568                    item_id=item.library_item.id_, episode_id=item.episode_id
569                )
570                playlist_items.append(
571                    parse_podcast_episode(
572                        episode=item.episode,
573                        prov_podcast_id=item.library_item.id_,
574                        prov_podcast_name=item.library_item.media.metadata.title,
575                        fallback_episode_cnt=None,
576                        instance_id=self.instance_id,
577                        domain=self.domain,
578                        token=self._client.token,
579                        base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
580                        media_progress=progress,
581                        cover_path=item.library_item.media.cover_path,
582                        cover_version=item.library_item.updated_at,
583                    )
584                )
585        for cnt, playlist_item in enumerate(playlist_items):
586            playlist_item.position = cnt
587
588        return playlist_items
589
590    @handle_refresh_token
591    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
592        """
593        Create a playlist in ABS.
594
595        This method may only be called, if we have not more than one library per media item in ABS.
596        """
597        error_msg = (
598            "The ABS provider only supports playlists of _either_ audiobooks, or podcast episodes."
599        )
600        if len(media_types) != 1:
601            raise InvalidDataError(error_msg)
602        media_type = next(iter(media_types))
603        if media_type == MediaType.AUDIOBOOK:
604            library_id = next(iter(self.libraries.audiobooks.keys()))
605        elif media_type == MediaType.PODCAST_EPISODE:
606            library_id = next(iter(self.libraries.podcasts.keys()))
607        else:
608            raise InvalidDataError(error_msg)
609        async with self.playlist_lock:
610            self.playlist_last = time.time()
611            abs_playlist = await self._client.create_playlist(
612                parameters=AbsCreatePlaylistParameters(name=name, library_id=library_id)
613            )
614            return parse_playlist(
615                abs_playlist=abs_playlist,
616                instance_id=self.instance_id,
617                domain=self.domain,
618                token=self._client.token,
619                base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
620                owner=self.abs_username,
621                media_type=media_type,
622            )
623
624    @handle_refresh_token
625    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
626        """Add items to playlist."""
627
628        def get_playlist_item(ma_id: str) -> AbsPlaylistItem:
629            item_ids = ma_id.split(" ")
630            abs_item_id = item_ids[0]
631            episode_id = item_ids[1] if len(item_ids) == 2 else None
632            return AbsPlaylistItem(library_item_id=abs_item_id, episode_id=episode_id)
633
634        abs_items = [get_playlist_item(ma_id) for ma_id in prov_track_ids]
635        async with self.playlist_lock:
636            self.playlist_last = time.time()
637            await self._client.add_item_to_playlist_batch(
638                playlist_id=prov_playlist_id, items=abs_items
639            )
640
641    @handle_refresh_token
642    async def remove_playlist_tracks(
643        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
644    ) -> None:
645        """Remove items from playlist."""
646        try:
647            abs_playlist = await self._client.get_playlist(playlist_id=prov_playlist_id)
648        except AbsNotFoundError:
649            return
650        items_to_remove: list[AbsPlaylistItem] = []
651        for item_cnt, item in enumerate(abs_playlist.items):
652            if item_cnt in positions_to_remove:
653                items_to_remove.append(
654                    AbsPlaylistItem(
655                        library_item_id=item.library_item_id, episode_id=item.episode_id
656                    )
657                )
658        if items_to_remove:
659            async with self.playlist_lock:
660                self.playlist_last = time.time()
661                await self._client.remove_item_from_playlist_batch(
662                    playlist_id=prov_playlist_id, items=items_to_remove
663                )
664
665    @handle_refresh_token
666    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
667        """Remove item from ABS."""
668        if media_type != MediaType.PLAYLIST:
669            raise InvalidDataError(
670                "Library remove is only implemented for playlists in the Audiobookshelf provider."
671            )
672        async with self.playlist_lock:
673            self.playlist_last = time.time()
674            with suppress(AbsNotFoundError):
675                # suppress due to edge case in add_library_tracks
676                await self._client.delete_playlist(playlist_id=prov_item_id)
677            return True
678
679    @handle_refresh_token
680    async def library_add(self, item: MediaItemType) -> bool:
681        """
682        Add library item.
683
684        This method is only called, if this item in question is not part of your library
685        yet, e.g. a "top 500 mix playlist". This doesn't exist in ABS.
686        """
687        self.logger.error(
688            "The library_add is not implemented on the ABS provider. Please reach out to us, "
689            "should you see this message in your log."
690        )
691        return False
692
693    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
694        """
695        Retrieve library/subscribed podcasts from the provider.
696
697        Minified podcast information is enough.
698        """
699        for pod_lib_id in self.libraries.podcasts:
700            async for response in self._client.get_library_items(library_id=pod_lib_id):
701                if not response.results:
702                    break
703                podcast_ids = [x.id_ for x in response.results]
704                # store uuids
705                self.libraries.podcasts[pod_lib_id].item_ids.update(podcast_ids)
706                for podcast_minified in response.results:
707                    assert isinstance(podcast_minified, LibraryItemMinifiedPodcast)
708                    mass_podcast = parse_podcast(
709                        abs_podcast=podcast_minified,
710                        instance_id=self.instance_id,
711                        domain=self.domain,
712                        token=self._client.token,
713                        base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
714                    )
715                    if (
716                        bool(self.config.get_value(CONF_HIDE_EMPTY_PODCASTS))
717                        and mass_podcast.total_episodes == 0
718                    ):
719                        continue
720                    yield mass_podcast
721
722    async def get_recommendations(self) -> list[RecommendationFolder]:
723        """Get the available recommendation rows, without items."""
724        if len(self.libraries.audiobooks) + len(self.libraries.podcasts) == 0:
725            self._log_no_libraries()
726            return []
727        rows = await self._recommendation_rows_from_payload()
728        rows.append(self._browse_recommendation_row())
729        return rows
730
731    async def get_recommendation_items(
732        self, item_id: str
733    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
734        """
735        Get the items for a single recommendation row.
736
737        :param item_id: The item_id of the row, as returned by get_recommendations.
738        """
739        if item_id == "browse":
740            return self._browse_recommendation_items()
741        return await self._recommendation_items_from_payload(item_id)
742
743    @handle_refresh_token
744    async def _get_abs_expanded_podcast(
745        self, prov_podcast_id: str
746    ) -> AbsLibraryItemExpandedPodcast:
747        abs_podcast = await self._client.get_library_item_podcast(
748            podcast_id=prov_podcast_id, expanded=True
749        )
750        assert isinstance(abs_podcast, AbsLibraryItemExpandedPodcast)
751
752        return abs_podcast
753
754    @handle_refresh_token
755    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
756        """Get single podcast."""
757        abs_podcast = await self._get_abs_expanded_podcast(prov_podcast_id=prov_podcast_id)
758        return parse_podcast(
759            abs_podcast=abs_podcast,
760            instance_id=self.instance_id,
761            domain=self.domain,
762            token=self._client.token,
763            base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
764        )
765
766    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
767        """
768        Get all podcast episodes of podcast.
769
770        Adds progress information.
771        """
772        abs_podcast = await self._get_abs_expanded_podcast(prov_podcast_id=prov_podcast_id)
773        episode_cnt = 1
774        # the user has the progress of all media items
775        # so we use a single api call here to obtain possibly many
776        # progresses for episodes
777        user = await self._client.get_my_user()
778        abs_progresses = {
779            x.episode_id: x
780            for x in user.media_progress
781            if x.episode_id is not None and x.library_item_id == prov_podcast_id
782        }
783        for abs_episode in abs_podcast.media.episodes:
784            progress = abs_progresses.get(abs_episode.id_)
785            mass_episode = parse_podcast_episode(
786                episode=abs_episode,
787                prov_podcast_id=prov_podcast_id,
788                prov_podcast_name=abs_podcast.media.metadata.title,
789                fallback_episode_cnt=episode_cnt,
790                instance_id=self.instance_id,
791                domain=self.domain,
792                token=self._client.token,
793                base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
794                media_progress=progress,
795                cover_path=abs_podcast.media.cover_path,
796                cover_version=abs_podcast.updated_at,
797            )
798            yield mass_episode
799            episode_cnt += 1
800
801    @handle_refresh_token
802    async def get_podcast_episode(
803        self, prov_episode_id: str, add_progress: bool = True
804    ) -> PodcastEpisode:
805        """Get single podcast episode."""
806        prov_podcast_id, e_id = prov_episode_id.split(" ")
807        abs_podcast = await self._get_abs_expanded_podcast(prov_podcast_id=prov_podcast_id)
808        episode_cnt = 1
809        for abs_episode in abs_podcast.media.episodes:
810            if abs_episode.id_ == e_id:
811                progress = None
812                if add_progress:
813                    progress = await self._client.get_my_media_progress(
814                        item_id=prov_podcast_id, episode_id=abs_episode.id_
815                    )
816                return parse_podcast_episode(
817                    episode=abs_episode,
818                    prov_podcast_id=prov_podcast_id,
819                    prov_podcast_name=abs_podcast.media.metadata.title,
820                    fallback_episode_cnt=episode_cnt,
821                    instance_id=self.instance_id,
822                    domain=self.domain,
823                    token=self._client.token,
824                    base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
825                    media_progress=progress,
826                    cover_path=abs_podcast.media.cover_path,
827                    cover_version=abs_podcast.updated_at,
828                )
829
830            episode_cnt += 1
831        raise MediaNotFoundError("Episode not found")
832
833    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
834        """
835        Get Audiobook libraries.
836
837        Need expanded version for chapters.
838        """
839        for book_lib_id in self.libraries.audiobooks:
840            async for response in self._client.get_library_items(library_id=book_lib_id):
841                if not response.results:
842                    break
843                book_ids = [x.id_ for x in response.results]
844                # store uuids
845                self.libraries.audiobooks[book_lib_id].item_ids.update(book_ids)
846                # use expanded version for chapters/ caching.
847                books_expanded = await self._client.get_library_item_batch_book(item_ids=book_ids)
848                for book_expanded in books_expanded:
849                    # If the book has no audiofiles, we skip -> ebook only.
850                    if len(book_expanded.media.tracks) == 0:
851                        continue
852                    mass_audiobook = parse_audiobook(
853                        abs_audiobook=book_expanded,
854                        audiobook_narrators=await self._get_audiobook_narrators(book_expanded),
855                        instance_id=self.instance_id,
856                        domain=self.domain,
857                        token=self._client.token,
858                        base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
859                    )
860                    yield mass_audiobook
861
862    @handle_refresh_token
863    async def _get_abs_expanded_audiobook(
864        self, prov_audiobook_id: str
865    ) -> AbsLibraryItemExpandedBook:
866        abs_audiobook = await self._client.get_library_item_book(
867            book_id=prov_audiobook_id, expanded=True
868        )
869        assert isinstance(abs_audiobook, AbsLibraryItemExpandedBook)
870
871        return abs_audiobook
872
873    @handle_refresh_token
874    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
875        """
876        Get a single audiobook.
877
878        Progress is added here.
879        """
880        progress = await self._client.get_my_media_progress(item_id=prov_audiobook_id)
881        abs_audiobook = await self._get_abs_expanded_audiobook(prov_audiobook_id=prov_audiobook_id)
882        return parse_audiobook(
883            abs_audiobook=abs_audiobook,
884            audiobook_narrators=await self._get_audiobook_narrators(abs_audiobook),
885            instance_id=self.instance_id,
886            domain=self.domain,
887            token=self._client.token,
888            base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
889            media_progress=progress,
890        )
891
892    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
893        """Get stream of item."""
894        # We always create a playback session. The default is direct playback.
895        # In that case, session.tracks holds the exact same as the audiobook/ podcast.track,
896        # so we only use the session to update our progress.
897        if media_type in (MediaType.PODCAST_EPISODE, MediaType.AUDIOBOOK):
898            session = await self._get_playback_session(mass_item_id=item_id)
899            return await self._get_stream_details_session(
900                session, session_helper=self.sessions[item_id], media_type=media_type
901            )
902        raise MediaNotFoundError("Stream unknown")
903
904    async def _get_stream_details_session(
905        self,
906        abs_session: AbsPlaybackSessionExpanded,
907        session_helper: SessionHelper,
908        media_type: MediaType,
909    ) -> StreamDetails:
910        """
911        Streamdetails audiobook.
912
913        We always use a custom stream type, also for single file, such
914        that we can handle an ffmpeg error and refresh our tokens.
915        """
916        abs_base_url = str(self.get_setup_value(CONF_URL))
917        tracks = abs_session.audio_tracks
918
919        if len(tracks) == 0:
920            raise MediaNotFoundError("Session has no tracks.")
921
922        content_type = ContentType.UNKNOWN
923        if abs_session.audio_tracks[0].metadata is not None:
924            content_type = ContentType.try_parse(abs_session.audio_tracks[0].metadata.ext)
925
926        file_parts: list[MultiPartPath] = []
927        if self.is_token_user:
928            self.logger.debug("Token User - Streams are direct.")
929        for idx, track in enumerate(tracks):
930            if self.is_token_user:
931                # an api key is long-lived
932                stream_url = f"{abs_base_url}{track.content_url}?token={self._client.token}"
933            else:
934                # to ensure token is always valid, we create a dynamic url
935                # this ensures that we always get a fresh token on each part
936                # without having to deal with a custom stream etc.
937                # we also use this for a single track, otherwise we can't seek
938                stream_url = (
939                    f"{self.mass.streams.base_url}/{self.instance_id}_part_stream?"
940                    f"session_id={abs_session.id_}&part_id={idx}"
941                )
942            file_parts.append(MultiPartPath(path=stream_url, duration=track.duration))
943
944        return StreamDetails(
945            provider=self.instance_id,
946            item_id=abs_session.id_,
947            audio_format=AudioFormat(content_type=content_type),
948            media_type=media_type,
949            stream_type=StreamType.HTTP,
950            duration=int(abs_session.duration),
951            path=file_parts[0].path if len(file_parts) == 1 else file_parts,
952            can_seek=True,
953            allow_seek=True,
954            expiration=STREAMDETAILS_EXPIRATION_S,
955        )
956
957    async def _get_playback_session(self, mass_item_id: str) -> AbsPlaybackSessionExpanded:
958        """Either creates or returns an open abs session."""
959        async with self.create_session_lock:
960            # check for an available open session
961            if session_helper := self.sessions.get(mass_item_id):
962                # reset here, as this is our "time listened".
963                session_helper.last_sync_time = time.time()
964                with suppress(AbsSessionNotFoundError):
965                    return await self._client.get_open_session(
966                        session_id=session_helper.abs_session_id
967                    )
968
969            item_ids = mass_item_id.split(" ")
970            abs_item_id = item_ids[0]
971            episode_id = item_ids[1] if len(item_ids) == 2 else None
972
973            # Abs allows a single session per device id.
974
975            client_name = f"Music Assistant {self.instance_id}"
976            device_id = f"{self.instance_id}_{mass_item_id}"
977            device_info = AbsDeviceInfo(
978                device_id=device_id,
979                client_name=client_name,
980                client_version=self.mass.version,
981                manufacturer="",
982                model=self.mass.server_id,
983            )
984
985            session = await self._client.get_playback_session(
986                # Direct play gives us the individual files. Transcode give an HLS session.
987                # Sessions without HLS proved to be stable. See:
988                # https://github.com/music-assistant/support/issues/4754
989                # https://github.com/music-assistant/support/issues/4586
990                session_parameters=AbsPlaybackSessionParameters(
991                    device_info=device_info,
992                    force_direct_play=True,
993                    force_transcode=False,
994                    # mimetypes are only checked for abs' internal "should transcode
995                    # see https://github.com/advplyr/audiobookshelf/blob/master/server/managers/PlaybackSessionManager.js
996                    supported_mime_types=[],
997                    media_player=client_name,
998                ),
999                item_id=abs_item_id,
1000                episode_id=episode_id,
1001            )
1002
1003            self.sessions[mass_item_id] = SessionHelper(
1004                abs_session_id=session.id_,
1005                last_sync_time=time.time(),
1006            )
1007            return session
1008
1009    @handle_refresh_token
1010    async def _handle_session_part_request(self, request: web.Request) -> web.Response:
1011        """
1012        Handle dynamic audiobook part stream request.
1013
1014        We redirect to the actual stream url with token.
1015        This is done because the token might expire, so we need to
1016        generate a fresh url on each part.
1017        """
1018        if not (session_id := request.query.get("session_id")):
1019            return web.Response(status=400, text="Missing session_id")
1020        if not (part_id := request.query.get("part_id")):
1021            return web.Response(status=400, text="Missing part_id")
1022        self.logger.debug(
1023            "Handling session part request for session %s and part %s", session_id, part_id
1024        )
1025        try:
1026            abs_session = await self._client.get_open_session(session_id=session_id)
1027        except AbsSessionNotFoundError as err:
1028            raise web.HTTPNotFound from err
1029        try:
1030            part_track = abs_session.audio_tracks[int(part_id)]
1031        except IndexError:
1032            return web.Response(status=404, text="Part not found")
1033
1034        base_url = str(self.get_setup_value(CONF_URL))
1035        stream_url = f"{base_url}{part_track.content_url}?token={self._client.token}"
1036        # redirect to the actual stream url
1037        raise web.HTTPFound(location=stream_url)
1038
1039    @handle_refresh_token
1040    async def get_resume_position(
1041        self, item_id: str, media_type: MediaType
1042    ) -> tuple[bool, int, datetime | None]:
1043        """Return finished:bool, position_ms: int."""
1044        # do not create a session here, as this method is called outside of stream acquisition
1045        item_ids = item_id.split(" ")
1046        abs_item_id = item_ids[0]
1047        episode_id = item_ids[1] if len(item_ids) == 2 else None
1048        progress = await self._client.get_my_media_progress(
1049            item_id=abs_item_id, episode_id=episode_id
1050        )
1051        if progress is None:
1052            # fallback to internal position
1053            raise NotImplementedError
1054        # The progress' last_update is in ms epoch
1055        timestamp = from_utc_timestamp(progress.last_update / 1000) if progress else None
1056        current_time = progress.current_time if progress.current_time is not None else 0.0
1057        self.logger.debug(
1058            "Acquired resume position %s for %s with item_id %s.",
1059            current_time,
1060            media_type.value,
1061            item_id,
1062        )
1063        return (
1064            progress.is_finished,
1065            int(current_time * 1000),
1066            timestamp,
1067        )
1068
1069    async def _recommendations_iter_shelves(
1070        self,
1071        shelves: list[ShelfBook | ShelfPodcast | ShelfAuthors | ShelfEpisode | ShelfSeries],
1072        library_id: str,
1073        items_by_shelf_id: dict[AbsShelfId, list[list[MediaItemType | BrowseFolder]]],
1074    ) -> None:
1075        # ruff: noqa: PLR0915
1076        for shelf in shelves:
1077            media_type: MediaType
1078            match shelf.type_:
1079                case AbsShelfType.PODCAST:
1080                    media_type = MediaType.PODCAST
1081                case AbsShelfType.EPISODE:
1082                    media_type = MediaType.PODCAST_EPISODE
1083                case AbsShelfType.BOOK:
1084                    media_type = MediaType.AUDIOBOOK
1085                case AbsShelfType.SERIES | AbsShelfType.AUTHORS:
1086                    media_type = MediaType.FOLDER
1087
1088            items: list[MediaItemType | BrowseFolder] = []
1089            # Recently added is the _only_ case, where we get a full podcast
1090            # We have a podcast object with only the episodes matching the
1091            # shelf.id_ otherwise.
1092            match shelf.id_:
1093                case (
1094                    AbsShelfId.RECENTLY_ADDED
1095                    | AbsShelfId.LISTEN_AGAIN
1096                    | AbsShelfId.DISCOVER
1097                    | AbsShelfId.NEWEST_EPISODES
1098                    | AbsShelfId.CONTINUE_LISTENING
1099                ):
1100                    for entity in shelf.entities:
1101                        assert isinstance(entity, ShelfLibraryItemMinified)
1102                        item: MediaItemType | None = None
1103                        if media_type in [MediaType.PODCAST, MediaType.AUDIOBOOK]:
1104                            item = await self.mass.music.get_library_item_by_prov_id(
1105                                media_type=media_type,
1106                                provider_instance_id_or_domain=self.instance_id,
1107                                item_id=entity.id_,
1108                            )
1109                        elif media_type == MediaType.PODCAST_EPISODE:
1110                            podcast_id = entity.id_
1111                            if entity.recent_episode is None:
1112                                continue
1113                            _cover_path = None
1114                            _cover_version = None
1115                            _podcast_title = None
1116                            if isinstance(entity, ShelfLibraryItemMinifiedPodcast):
1117                                _cover_path = entity.media.cover_path
1118                                _cover_version = entity.updated_at
1119                                _podcast_title = entity.media.metadata.title
1120                            # we only have a PodcastEpisode here, with limited information
1121                            item = parse_podcast_episode(
1122                                episode=entity.recent_episode,
1123                                prov_podcast_id=podcast_id,
1124                                prov_podcast_name=_podcast_title,
1125                                instance_id=self.instance_id,
1126                                domain=self.domain,
1127                                token=self._client.token,
1128                                base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
1129                                cover_path=_cover_path,
1130                                cover_version=_cover_version,
1131                            )
1132                        if item is not None:
1133                            items.append(item)
1134                case AbsShelfId.RECENT_SERIES | AbsShelfId.CONTINUE_SERIES:
1135                    # We jump into a browse folder here if we have SeriesShelf, set path up as if
1136                    # browse function used.
1137                    if isinstance(shelf, ShelfSeries):
1138                        for entity in shelf.entities:
1139                            assert isinstance(entity, SeriesShelf)
1140                            if len(entity.books) == 0:
1141                                continue
1142                            path = (
1143                                f"{self.instance_id}://"
1144                                f"{AbsBrowsePaths.LIBRARIES_BOOK} {library_id}/"
1145                                f"{AbsBrowsePaths.SERIES}/{entity.id_}"
1146                            )
1147                            items.append(
1148                                BrowseFolder(
1149                                    item_id=entity.id_,
1150                                    name=entity.name,
1151                                    provider=self.instance_id,
1152                                    path=path,
1153                                )
1154                            )
1155                    elif isinstance(shelf, ShelfBook) and media_type == MediaType.AUDIOBOOK:
1156                        # Single books, must be audiobooks
1157                        for entity in shelf.entities:
1158                            item = await self.mass.music.get_library_item_by_prov_id(
1159                                media_type=media_type,
1160                                provider_instance_id_or_domain=self.instance_id,
1161                                item_id=entity.id_,
1162                            )
1163                            if item is not None:
1164                                items.append(item)
1165                case AbsShelfId.NEWEST_AUTHORS:
1166                    # same as for series, use a folder
1167                    for entity in shelf.entities:
1168                        assert isinstance(entity, AuthorExpanded)
1169                        if entity.num_books == 0:
1170                            continue
1171                        path = (
1172                            f"{self.instance_id}://"
1173                            f"{AbsBrowsePaths.LIBRARIES_BOOK} {library_id}/"
1174                            f"{AbsBrowsePaths.AUTHORS}/{entity.id_}"
1175                        )
1176                        items.append(
1177                            BrowseFolder(
1178                                item_id=entity.id_,
1179                                name=entity.name,
1180                                provider=self.instance_id,
1181                                path=path,
1182                            )
1183                        )
1184            if not items:
1185                continue
1186
1187            # add collected items
1188            assert isinstance(shelf.id_, AbsShelfId)
1189            items_collected = items_by_shelf_id.get(shelf.id_, [])
1190            items_collected.append(items)
1191            items_by_shelf_id[shelf.id_] = items_collected
1192
1193    @handle_refresh_token
1194    async def on_played(
1195        self,
1196        media_type: MediaType,
1197        prov_item_id: str,
1198        fully_played: bool,
1199        position: int,
1200        media_item: MediaItemType,
1201        is_playing: bool = False,
1202    ) -> None:
1203        """
1204        Update progress in Audiobookshelf.
1205
1206        In our case media_type may have 3 values:
1207            - PODCAST
1208            - PODCAST_EPISODE
1209            - AUDIOBOOK
1210        We ignore PODCAST (function is called on adding a podcast with position=None)
1211
1212        """
1213
1214        async def _update_by_session(session_helper: SessionHelper, duration: int) -> bool:
1215            now = time.time()
1216            time_listened = now - session_helper.last_sync_time
1217            if time_listened > PLAYBACK_REPORT_INTERVAL_SECONDS * 2 + 10:
1218                # See player_queues controller, we get an update every 30s, and immediately on pause
1219                # or play.
1220                # We reset after two missed updates, as this indicates a trigger after a longer
1221                # absence and should not count into abs' statistics
1222                self.logger.debug("Resetting time_listened due to longer absence.")
1223                time_listened = 0.0
1224            try:
1225                await self._client.sync_open_session(
1226                    session_id=session_helper.abs_session_id,
1227                    parameters=SyncOpenSessionParameters(
1228                        current_time=position,
1229                        time_listened=time_listened,
1230                        duration=duration,
1231                    ),
1232                )
1233                session_helper.last_sync_time = now
1234                session_helper.failed_sync_count = 0
1235                self.logger.debug("Synced playback session, position %s s.", position)
1236                return True
1237            except AbsSessionSyncError:
1238                session_helper.failed_sync_count += 1
1239                if session_helper.failed_sync_count >= 5:
1240                    self.logger.warning(
1241                        "Unable to sync session %s after %s attempts - "
1242                        "falling back to non-session approach.",
1243                        session_helper.abs_session_id,
1244                        session_helper.failed_sync_count,
1245                    )
1246                    self.sessions.pop(prov_item_id, None)
1247            return False
1248
1249        if media_type == MediaType.PODCAST_EPISODE:
1250            abs_podcast_id, abs_episode_id = prov_item_id.split(" ")
1251
1252            # guard, see progress guard class docstrings for explanation
1253            if not self.progress_guard.guard_ok_mass(
1254                item_id=abs_podcast_id, episode_id=abs_episode_id
1255            ):
1256                return
1257            self.progress_guard.add_progress(item_id=abs_podcast_id, episode_id=abs_episode_id)
1258
1259            if media_item is None or not isinstance(media_item, PodcastEpisode):
1260                return
1261
1262            if fully_played and position < media_item.duration - PLAYBACK_REPORT_INTERVAL_SECONDS:
1263                # faulty position update
1264                # occurs sometimes, if a player disconnects unexpectedly, or reports
1265                # a false position - seen this for MC players, but not for sendspin
1266                return
1267
1268            if position == 0 and not fully_played:
1269                # marked unplayed
1270                mp = await self._client.get_my_media_progress(
1271                    item_id=abs_podcast_id, episode_id=abs_episode_id
1272                )
1273                if mp is not None:
1274                    await self._client.remove_my_media_progress(media_progress_id=mp.id_)
1275                    self.logger.debug(f"Removed media progress of {media_type.value}.")
1276                    return
1277
1278            duration = media_item.duration
1279            updated = False
1280            if session_helper := self.sessions.get(prov_item_id):
1281                updated = await _update_by_session(session_helper=session_helper, duration=duration)
1282            if not updated:
1283                self.logger.debug(
1284                    f"Updating media progress of {media_type.value}, title {media_item.name}."
1285                )
1286                await self._client.update_my_media_progress(
1287                    item_id=abs_podcast_id,
1288                    episode_id=abs_episode_id,
1289                    duration_seconds=duration,
1290                    progress_seconds=position,
1291                    is_finished=fully_played,
1292                )
1293
1294        if media_type == MediaType.AUDIOBOOK:
1295            # guard, see progress guard class docstrings for explanation
1296            if not self.progress_guard.guard_ok_mass(item_id=prov_item_id):
1297                return
1298            self.progress_guard.add_progress(item_id=prov_item_id)
1299
1300            if media_item is None or not isinstance(media_item, Audiobook):
1301                return
1302
1303            if fully_played and position < media_item.duration - PLAYBACK_REPORT_INTERVAL_SECONDS:
1304                # faulty position update, see above
1305                return
1306
1307            if position == 0 and not fully_played:
1308                # marked unplayed
1309                mp = await self._client.get_my_media_progress(item_id=prov_item_id)
1310                if mp is not None:
1311                    await self._client.remove_my_media_progress(media_progress_id=mp.id_)
1312                    self.logger.debug(f"Removed media progress of {media_type.value}.")
1313                return
1314
1315            duration = media_item.duration
1316            updated = False
1317            if session_helper := self.sessions.get(prov_item_id):
1318                updated = await _update_by_session(session_helper=session_helper, duration=duration)
1319            if not updated:
1320                self.logger.debug(f"Updating {media_type.value} named {media_item.name} progress")
1321                await self._client.update_my_media_progress(
1322                    item_id=prov_item_id,
1323                    duration_seconds=duration,
1324                    progress_seconds=position,
1325                    is_finished=fully_played,
1326                )
1327
1328    @handle_refresh_token
1329    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1330        """
1331        Browse for audiobookshelf.
1332
1333        Generates this view:
1334        Library_Name_A (Audiobooks)
1335            Audiobooks
1336                Audiobook_1
1337                Audiobook_2
1338            Series
1339                Series_1
1340                    Audiobook_1
1341                    Audiobook_2
1342                Series_2
1343                    Audiobook_3
1344                    Audiobook_4
1345            Collections
1346                Collection_1
1347                    Audiobook_1
1348                    Audiobook_2
1349                Collection_2
1350                    Audiobook_3
1351                    Audiobook_4
1352            Authors
1353                Author_1
1354                    Series_1
1355                    Audiobook_1
1356                    Audiobook_2
1357                Author_2
1358                    Audiobook_3
1359        Library_Name_B (Podcasts)
1360            Podcast_1
1361            Podcast_2
1362        """
1363        # ruff: noqa: PLR0911 # to many return
1364        item_path = path.split("://", 1)[1]
1365        if not item_path:
1366            return self._browse_root()
1367        sub_path = item_path.split("/")
1368        lib_key, lib_id = sub_path[0].split(" ")
1369        if len(sub_path) == 1:
1370            if lib_key == AbsBrowsePaths.LIBRARIES_PODCAST:
1371                return self._browse_lib_podcasts(current_path=path)
1372            return self._browse_lib_audiobooks(current_path=path)
1373        if len(sub_path) == 2:
1374            item_key = sub_path[1]
1375            match item_key:
1376                case AbsBrowsePaths.AUTHORS:
1377                    return await self._browse_authors(current_path=path, library_id=lib_id)
1378                case AbsBrowsePaths.NARRATORS:
1379                    return await self._browse_narrators(current_path=path, library_id=lib_id)
1380                case AbsBrowsePaths.SERIES:
1381                    return await self._browse_series(current_path=path, library_id=lib_id)
1382                case AbsBrowsePaths.COLLECTIONS:
1383                    return await self._browse_collections(current_path=path, library_id=lib_id)
1384                case AbsBrowsePaths.AUDIOBOOKS:
1385                    return await self._browse_books(library_id=lib_id)
1386                case AbsBrowsePaths.PODCASTS:
1387                    return await self._browse_podcasts(library_id=lib_id)
1388                case AbsBrowsePaths.PLAYLISTS:
1389                    return await self._browse_playlists(library_id=lib_id, browse_path=lib_key)
1390        elif len(sub_path) == 3:
1391            item_key, item_id = sub_path[1:3]
1392            match item_key:
1393                case AbsBrowsePaths.AUTHORS:
1394                    return await self._browse_author_books(current_path=path, author_id=item_id)
1395                case AbsBrowsePaths.NARRATORS:
1396                    return await self._browse_narrator_books(
1397                        library_id=lib_id, narrator_filter_str=item_id
1398                    )
1399                case AbsBrowsePaths.SERIES:
1400                    return await self._browse_series_books(series_id=item_id)
1401                case AbsBrowsePaths.COLLECTIONS:
1402                    return await self._browse_collection_books(collection_id=item_id)
1403        elif len(sub_path) == 4:
1404            # series within author
1405            series_id = sub_path[3]
1406            return await self._browse_series_books(series_id=series_id)
1407        return []
1408
1409    def _browse_root(self, append_mediatype_suffix: bool = True) -> Sequence[BrowseFolder]:
1410        items = []
1411
1412        def _get_folder(
1413            path: str, lib_id: str, lib_name: str, translation_key: str | None = None
1414        ) -> BrowseFolder:
1415            return BrowseFolder(
1416                item_id=lib_id,
1417                name=lib_name,
1418                translation_key=translation_key,
1419                translation_params=[lib_name],
1420                provider=self.instance_id,
1421                path=f"{self.instance_id}://{path}",
1422            )
1423
1424        if len(self.libraries.audiobooks) == 0 and len(self.libraries.podcasts) == 0:
1425            self._log_no_libraries()
1426            return []
1427
1428        translation_key: str | None
1429        for lib_id, lib in self.libraries.audiobooks.items():
1430            path = f"{AbsBrowsePaths.LIBRARIES_BOOK} {lib_id}"
1431            translation_key = None
1432            if append_mediatype_suffix:
1433                translation_key = AbsBrowseItemsBookTranslationKey.AUDIOBOOKS_LIBRARY
1434            items.append(
1435                _get_folder(path, lib_id, lib_name=lib.name, translation_key=translation_key)
1436            )
1437        for lib_id, lib in self.libraries.podcasts.items():
1438            path = f"{AbsBrowsePaths.LIBRARIES_PODCAST} {lib_id}"
1439            translation_key = None
1440            if append_mediatype_suffix:
1441                translation_key = AbsBrowseItemsPodcastTranslationKey.PODCASTS_LIBRARY
1442            items.append(
1443                _get_folder(path, lib_id, lib_name=lib.name, translation_key=translation_key)
1444            )
1445        return items
1446
1447    def _browse_lib_podcasts(self, current_path: str) -> Sequence[BrowseFolder]:
1448        items = []
1449        for translation_key in AbsBrowseItemsPodcastTranslationKey:
1450            if "library" in translation_key:
1451                continue
1452            path = current_path + "/" + ABS_BROWSE_ITEMS_PODCAST_TO_PATH[translation_key]
1453            items.append(
1454                BrowseFolder(
1455                    item_id=translation_key.lower(),
1456                    name="",
1457                    translation_key=translation_key,
1458                    provider=self.instance_id,
1459                    path=path,
1460                )
1461            )
1462        return items
1463
1464    async def _browse_podcasts(self, library_id: str) -> list[MediaItemType]:
1465        """Browse podcasts."""
1466        if len(self.libraries.podcasts[library_id].item_ids) == 0:
1467            self._log_no_helper_item_ids()
1468        items = []
1469        for podcast_id in self.libraries.podcasts[library_id].item_ids:
1470            mass_item = await self.mass.music.get_library_item_by_prov_id(
1471                media_type=MediaType.PODCAST,
1472                item_id=podcast_id,
1473                provider_instance_id_or_domain=self.instance_id,
1474            )
1475            if mass_item is not None:
1476                items.append(mass_item)
1477        return sorted(items, key=lambda x: x.name)
1478
1479    def _browse_lib_audiobooks(self, current_path: str) -> Sequence[BrowseFolder]:
1480        items = []
1481        for translation_key in AbsBrowseItemsBookTranslationKey:
1482            if "library" in translation_key or "entry" in translation_key:
1483                continue
1484            path = current_path + "/" + ABS_BROWSE_ITEMS_BOOK_TO_PATH[translation_key]
1485            items.append(
1486                BrowseFolder(
1487                    item_id=translation_key.lower(),
1488                    name="",
1489                    translation_key=translation_key,
1490                    provider=self.instance_id,
1491                    path=path,
1492                )
1493            )
1494        return items
1495
1496    async def _browse_authors(self, current_path: str, library_id: str) -> Sequence[BrowseFolder]:
1497        abs_authors = await self._client.get_library_authors(library_id=library_id)
1498        items = []
1499        for author in abs_authors:
1500            path = f"{current_path}/{author.id_}"
1501            items.append(
1502                BrowseFolder(
1503                    item_id=author.id_,
1504                    name=author.name,
1505                    provider=self.instance_id,
1506                    path=path,
1507                )
1508            )
1509
1510        return sorted(items, key=lambda x: x.name)
1511
1512    async def _browse_narrators(self, current_path: str, library_id: str) -> Sequence[BrowseFolder]:
1513        abs_narrators = await self._client.get_library_narrators(library_id=library_id)
1514        items = []
1515        for narrator in abs_narrators:
1516            path = f"{current_path}/{narrator.id_}"
1517            items.append(
1518                BrowseFolder(
1519                    item_id=narrator.id_,
1520                    name=narrator.name,
1521                    provider=self.instance_id,
1522                    path=path,
1523                )
1524            )
1525
1526        return sorted(items, key=lambda x: x.name)
1527
1528    async def _browse_series(self, current_path: str, library_id: str) -> Sequence[BrowseFolder]:
1529        items = []
1530        async for response in self._client.get_library_series(library_id=library_id):
1531            if not response.results:
1532                break
1533            for abs_series in response.results:
1534                path = f"{current_path}/{abs_series.id_}"
1535                items.append(
1536                    BrowseFolder(
1537                        item_id=abs_series.id_,
1538                        name=abs_series.name,
1539                        provider=self.instance_id,
1540                        path=path,
1541                    )
1542                )
1543
1544        return sorted(items, key=lambda x: x.name)
1545
1546    async def _browse_collections(
1547        self, current_path: str, library_id: str
1548    ) -> Sequence[BrowseFolder]:
1549        items = []
1550        async for response in self._client.get_library_collections(library_id=library_id):
1551            if not response.results:
1552                break
1553            for abs_collection in response.results:
1554                path = f"{current_path}/{abs_collection.id_}"
1555                items.append(
1556                    BrowseFolder(
1557                        item_id=abs_collection.id_,
1558                        name=abs_collection.name,
1559                        provider=self.instance_id,
1560                        path=path,
1561                    )
1562                )
1563        return sorted(items, key=lambda x: x.name)
1564
1565    @handle_refresh_token
1566    async def _browse_playlists(self, library_id: str, browse_path: str) -> Sequence[MediaItemType]:
1567        items = []
1568        if browse_path == AbsBrowsePaths.LIBRARIES_PODCAST:
1569            playlists = self.libraries.playlists_podcasts
1570            if len(self.libraries.playlists_podcasts) == 0:
1571                self._log_no_helper_item_ids()
1572        elif browse_path == AbsBrowsePaths.LIBRARIES_BOOK:
1573            playlists = self.libraries.playlists_audiobooks
1574            if len(self.libraries.playlists_audiobooks) == 0:
1575                self._log_no_helper_item_ids()
1576        else:
1577            raise RuntimeError("Unknown media type in browse playlist.")
1578        for playlist_id in playlists[library_id]:
1579            mass_item = await self.mass.music.get_library_item_by_prov_id(
1580                media_type=MediaType.PLAYLIST,
1581                item_id=playlist_id,
1582                provider_instance_id_or_domain=self.instance_id,
1583            )
1584            if mass_item is not None:
1585                items.append(mass_item)
1586        return sorted(items, key=lambda x: x.name)
1587
1588    async def _browse_books(self, library_id: str) -> Sequence[MediaItemType]:
1589        if len(self.libraries.audiobooks[library_id].item_ids) == 0:
1590            self._log_no_helper_item_ids()
1591        items = []
1592        for book_id in self.libraries.audiobooks[library_id].item_ids:
1593            mass_item = await self.mass.music.get_library_item_by_prov_id(
1594                media_type=MediaType.AUDIOBOOK,
1595                item_id=book_id,
1596                provider_instance_id_or_domain=self.instance_id,
1597            )
1598            if mass_item is not None:
1599                items.append(mass_item)
1600        return sorted(items, key=lambda x: x.name)
1601
1602    async def _browse_author_books(
1603        self, current_path: str, author_id: str
1604    ) -> Sequence[MediaItemType | BrowseFolder]:
1605        items: list[MediaItemType | BrowseFolder] = []
1606
1607        abs_author = await self._client.get_author(
1608            author_id=author_id, include_items=True, include_series=True
1609        )
1610        if not isinstance(abs_author, AbsAuthorWithItemsAndSeries):
1611            raise TypeError("Unexpected type of author.")
1612
1613        book_ids = {x.id_ for x in abs_author.library_items}
1614        series_book_ids = set()
1615
1616        for series in abs_author.series:
1617            series_book_ids.update([x.id_ for x in series.items])
1618            path = f"{current_path}/{series.id_}"
1619            items.append(
1620                BrowseFolder(
1621                    item_id=series.id_,
1622                    name=series.name,
1623                    translation_key="series_entry",
1624                    translation_params=[series.name],
1625                    provider=self.instance_id,
1626                    path=path,
1627                )
1628            )
1629        book_ids = book_ids.difference(series_book_ids)
1630        for book_id in book_ids:
1631            mass_item = await self.mass.music.get_library_item_by_prov_id(
1632                media_type=MediaType.AUDIOBOOK,
1633                item_id=book_id,
1634                provider_instance_id_or_domain=self.instance_id,
1635            )
1636            if mass_item is not None:
1637                items.append(mass_item)
1638
1639        return items
1640
1641    async def _browse_narrator_books(
1642        self, library_id: str, narrator_filter_str: str
1643    ) -> Sequence[Audiobook]:
1644        items: list[Audiobook] = []
1645        async for response in self._client.get_library_items(
1646            library_id=library_id, filter_str=f"narrators.{narrator_filter_str}"
1647        ):
1648            if not response.results:
1649                break
1650            for item in response.results:
1651                mass_item = await self.mass.music.get_library_item_by_prov_id(
1652                    media_type=MediaType.AUDIOBOOK,
1653                    item_id=item.id_,
1654                    provider_instance_id_or_domain=self.instance_id,
1655                )
1656                if mass_item is not None:
1657                    mass_item = cast("Audiobook", mass_item)
1658                    items.append(mass_item)
1659
1660        return sorted(items, key=lambda x: x.name)
1661
1662    async def _browse_series_books(self, series_id: str) -> Sequence[MediaItemType]:
1663        items = []
1664
1665        abs_series = await self._client.get_series(series_id=series_id, include_progress=True)
1666        if not isinstance(abs_series, AbsSeriesWithProgress):
1667            raise TypeError("Unexpected series type.")
1668
1669        for book_id in abs_series.progress.library_item_ids:
1670            # these are sorted in abs by sequence
1671            mass_item = await self.mass.music.get_library_item_by_prov_id(
1672                media_type=MediaType.AUDIOBOOK,
1673                item_id=book_id,
1674                provider_instance_id_or_domain=self.instance_id,
1675            )
1676            if mass_item is not None:
1677                items.append(mass_item)
1678
1679        return items
1680
1681    async def _browse_collection_books(self, collection_id: str) -> Sequence[MediaItemType]:
1682        items = []
1683        abs_collection = await self._client.get_collection(collection_id=collection_id)
1684        for book in abs_collection.books:
1685            mass_item = await self.mass.music.get_library_item_by_prov_id(
1686                media_type=MediaType.AUDIOBOOK,
1687                item_id=book.id_,
1688                provider_instance_id_or_domain=self.instance_id,
1689            )
1690            if mass_item is not None:
1691                items.append(mass_item)
1692        return items
1693
1694    async def _socket_abs_item_changed(
1695        self, items: LibraryItemExpanded | list[LibraryItemExpanded]
1696    ) -> None:
1697        """For added and updated."""
1698        abs_items = [items] if isinstance(items, LibraryItemExpanded) else items
1699        for abs_item in abs_items:
1700            if isinstance(abs_item, LibraryItemExpandedBook):
1701                # If the book has no audiofiles, we skip -> ebook only.
1702                if len(abs_item.media.tracks) == 0:
1703                    continue
1704                self.logger.debug(
1705                    'Updated book "%s" via socket.', abs_item.media.metadata.title or ""
1706                )
1707                await self.mass.music.audiobooks.add_item_to_library(
1708                    parse_audiobook(
1709                        abs_audiobook=abs_item,
1710                        audiobook_narrators=await self._get_audiobook_narrators(abs_item),
1711                        instance_id=self.instance_id,
1712                        domain=self.domain,
1713                        token=self._client.token,
1714                        base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
1715                    ),
1716                    overwrite_existing=True,
1717                )
1718                lib = self.libraries.audiobooks.get(abs_item.library_id, None)
1719                if lib is not None:
1720                    lib.item_ids.add(abs_item.id_)
1721            elif isinstance(abs_item, LibraryItemExpandedPodcast):
1722                self.logger.debug(
1723                    'Updated podcast "%s" via socket.', abs_item.media.metadata.title or ""
1724                )
1725                mass_podcast = parse_podcast(
1726                    abs_podcast=abs_item,
1727                    instance_id=self.instance_id,
1728                    domain=self.domain,
1729                    token=self._client.token,
1730                    base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
1731                )
1732                if not (
1733                    bool(self.config.get_value(CONF_HIDE_EMPTY_PODCASTS))
1734                    and mass_podcast.total_episodes == 0
1735                ):
1736                    await self.mass.music.podcasts.add_item_to_library(
1737                        mass_podcast,
1738                        overwrite_existing=True,
1739                    )
1740                    lib = self.libraries.podcasts.get(abs_item.library_id, None)
1741                    if lib is not None:
1742                        lib.item_ids.add(abs_item.id_)
1743        await self._cache_set_helper_libraries()
1744
1745    async def _socket_abs_item_removed(self, item: LibraryItemRemoved) -> None:
1746        """Item removed."""
1747        media_type: MediaType | None = None
1748        for lib in self.libraries.audiobooks.values():
1749            if item.id_ in lib.item_ids:
1750                media_type = MediaType.AUDIOBOOK
1751                lib.item_ids.remove(item.id_)
1752                break
1753        for lib in self.libraries.podcasts.values():
1754            if item.id_ in lib.item_ids:
1755                media_type = MediaType.PODCAST
1756                lib.item_ids.remove(item.id_)
1757                break
1758
1759        if media_type is not None:
1760            mass_item = await self.mass.music.get_library_item_by_prov_id(
1761                media_type=media_type,
1762                item_id=item.id_,
1763                provider_instance_id_or_domain=self.instance_id,
1764            )
1765            if mass_item is not None:
1766                await self.mass.music.remove_item_from_library(
1767                    media_type=media_type, library_item_id=mass_item.item_id
1768                )
1769                self.logger.debug('Removed %s "%s" via socket.', media_type.value, mass_item.name)
1770
1771        await self._cache_set_helper_libraries()
1772
1773    async def _socket_abs_user_item_progress_updated(
1774        self, id_: str, progress: MediaProgress
1775    ) -> None:
1776        """
1777        To update continue listening.
1778
1779        ABS reports every 15s and immediately on play state change.
1780        This callback is called per item if a progress is changed:
1781            - a change in position
1782            - the item is finished
1783        But it is _not_called, if a progress is reset/ discarded.
1784        """
1785        # guard, see progress guard class docstrings for explanation
1786        if not self.progress_guard.guard_ok_abs(abs_progress=progress):
1787            return
1788
1789        known_ids = self._get_all_known_item_ids()
1790        if progress.library_item_id not in known_ids:
1791            return
1792
1793        self.logger.debug(f"Updated progress of item {progress.library_item_id} via socket.")
1794
1795        if progress.episode_id is None:
1796            await self._update_playlog_book(progress)
1797            return
1798        await self._update_playlog_episode(progress)
1799
1800    async def _socket_abs_playlist_changed(self, abs_playlist: AbsPlaylistExpanded) -> None:
1801        if time.time() - self.playlist_last < 5:
1802            return
1803        if abs_playlist.library_id in self.libraries.audiobooks:
1804            media_type = MediaType.AUDIOBOOK
1805        elif abs_playlist.library_id in self.libraries.podcasts:
1806            media_type = MediaType.PODCAST_EPISODE
1807        else:
1808            return
1809        async with self.playlist_lock:
1810            parsed_playlist = parse_playlist(
1811                abs_playlist=abs_playlist,
1812                instance_id=self.instance_id,
1813                domain=self.domain,
1814                token=self._client.token,
1815                base_url=str(self.get_setup_value(CONF_URL)).rstrip("/"),
1816                owner=self.abs_username,
1817                media_type=media_type,
1818            )
1819            ma_library_playlist = await self.mass.music.get_library_item_by_prov_id(
1820                media_type=MediaType.PLAYLIST,
1821                item_id=abs_playlist.id_,
1822                provider_instance_id_or_domain=self.instance_id,
1823            )
1824            if ma_library_playlist is not None and isinstance(ma_library_playlist, Playlist):
1825                await self.mass.music.playlists.update_item_in_library(
1826                    item_id=ma_library_playlist.item_id, update=parsed_playlist, overwrite=True
1827                )
1828            else:
1829                await self.mass.music.playlists.add_item_to_library(item=parsed_playlist)
1830            if media_type == MediaType.AUDIOBOOK:
1831                self.libraries.playlists_audiobooks[abs_playlist.library_id].add(abs_playlist.id_)
1832            elif media_type == MediaType.PODCAST_EPISODE:
1833                self.libraries.playlists_podcasts[abs_playlist.library_id].add(abs_playlist.id_)
1834        await self._cache_set_helper_libraries()
1835
1836    async def _socket_abs_playlist_removed(self, abs_playlist: AbsPlaylistExpanded) -> None:
1837        if time.time() - self.playlist_last < 5:
1838            return
1839        if mass_item := await self.mass.music.get_library_item_by_prov_id(
1840            media_type=MediaType.PLAYLIST,
1841            item_id=abs_playlist.id_,
1842            provider_instance_id_or_domain=self.instance_id,
1843        ):
1844            async with self.playlist_lock:
1845                await self.mass.music.playlists.remove_item_from_library(item_id=mass_item.item_id)
1846                playlist_set = self.libraries.playlists_audiobooks.get(abs_playlist.library_id)
1847                if playlist_set is None:
1848                    playlist_set = self.libraries.playlists_podcasts.get(abs_playlist.library_id)
1849                if playlist_set is not None:
1850                    with suppress(KeyError):
1851                        playlist_set.remove(abs_playlist.id_)
1852        await self._cache_set_helper_libraries()
1853
1854    async def _socket_abs_refresh_token_expired(self) -> None:
1855        await self.reauthenticate()
1856
1857    async def reauthenticate(self) -> None:
1858        """Reauthorize the abs session config if refresh token expired."""
1859        # some safe guarding should that function be called simultaneously
1860        if self.reauthenticate_lock.locked() or time.time() - self.reauthenticate_last < 5:
1861            while True:
1862                if not self.reauthenticate_lock.locked():
1863                    return
1864                await asyncio.sleep(0.5)
1865        async with self.reauthenticate_lock:
1866            await self._client.session_config.authenticate(
1867                username=str(self.get_setup_value(CONF_USERNAME)),
1868                password=str(self.get_setup_value(CONF_PASSWORD)),
1869            )
1870            self.reauthenticate_last = time.time()
1871
1872    def _get_all_known_item_ids(self) -> set[str]:
1873        known_ids = set()
1874        for lib in self.libraries.podcasts.values():
1875            known_ids.update(lib.item_ids)
1876        for lib in self.libraries.audiobooks.values():
1877            known_ids.update(lib.item_ids)
1878
1879        return known_ids
1880
1881    async def _set_playlog_from_user(self, user: User) -> None:
1882        """
1883        Update on user callback.
1884
1885        User holds also all media progresses specific to that user.
1886
1887        The function 'guard_ok_abs' uses the timestamp of the last update in abs, thus after an
1888        initial progress update, an unchanged update will not trigger a (useless) playlog update.
1889
1890        We do not sync removed progresses for the sake of simplicity.
1891        """
1892        await self._set_playlog_from_user_sync(user.media_progress)
1893
1894    async def _set_playlog_from_user_sync(self, progresses: list[MediaProgress]) -> None:
1895        # for debugging
1896        __updated_items = 0
1897
1898        known_ids = self._get_all_known_item_ids()
1899        abs_ids_with_progress = set()
1900
1901        for progress in progresses:
1902            # save progress ids for later
1903            ma_item_id = (
1904                progress.library_item_id
1905                if progress.episode_id is None
1906                else f"{progress.library_item_id} {progress.episode_id}"
1907            )
1908            abs_ids_with_progress.add(ma_item_id)
1909
1910            # Guard. Also makes sure, that we don't write to db again if no state change happened.
1911            # This is achieved by adding a Helper Progress in the update playlog functions, which
1912            # then has the most recent timestamp. If a subsequent progress sent by abs has an older
1913            # timestamp, we do not update again.
1914            if not self.progress_guard.guard_ok_abs(progress):
1915                continue
1916            if progress.current_time is not None:
1917                if (
1918                    int(progress.current_time) != 0
1919                    and not progress.current_time >= PLAYBACK_REPORT_INTERVAL_SECONDS
1920                ):
1921                    # same as mass default, only > 30s
1922                    continue
1923            if progress.library_item_id not in known_ids:
1924                continue
1925            __updated_items += 1
1926            if progress.episode_id is None:
1927                await self._update_playlog_book(progress)
1928            else:
1929                await self._update_playlog_episode(progress)
1930        self.logger.debug(f"Updated {__updated_items} from full playlog.")
1931
1932        # Get MA's known progresses of ABS.
1933        # In ABS the user may discard a progress, which removes the progress completely.
1934        # There is no socket notification for this event.
1935        ma_playlog_state = await self.mass.music.get_playlog_provider_item_ids(
1936            provider_instance_id=self.instance_id
1937        )
1938        ma_ids_with_progress = {x for _, x in ma_playlog_state}
1939        discarded_progress_ids = ma_ids_with_progress.difference(abs_ids_with_progress)
1940        for discarded_progress_id in discarded_progress_ids:
1941            if len(discarded_progress_id.split(" ")) == 1:
1942                if discarded_item := await self.mass.music.get_library_item_by_prov_id(
1943                    media_type=MediaType.AUDIOBOOK,
1944                    item_id=discarded_progress_id,
1945                    provider_instance_id_or_domain=self.instance_id,
1946                ):
1947                    self.progress_guard.add_progress(discarded_progress_id)
1948                    await self.mass.music.mark_item_unplayed(discarded_item)
1949            else:
1950                with suppress(MediaNotFoundError):
1951                    discarded_item = await self.get_podcast_episode(
1952                        prov_episode_id=discarded_progress_id, add_progress=False
1953                    )
1954                    self.progress_guard.add_progress(*discarded_progress_id.split(" "))
1955                    await self.mass.music.mark_item_unplayed(discarded_item)
1956            self.logger.debug("Discarded item %s ", discarded_progress_id)
1957
1958    async def _update_playlog_book(self, progress: MediaProgress) -> None:
1959        # helper progress also ensures no useless progress updates,
1960        # see comment above
1961        self.progress_guard.add_progress(progress.library_item_id)
1962        if progress.current_time is None:
1963            return
1964        mass_audiobook = await self.mass.music.get_library_item_by_prov_id(
1965            media_type=MediaType.AUDIOBOOK,
1966            item_id=progress.library_item_id,
1967            provider_instance_id_or_domain=self.instance_id,
1968        )
1969        if mass_audiobook is None:
1970            return
1971        if int(progress.current_time) == 0 and not progress.is_finished:
1972            await self.mass.music.mark_item_unplayed(mass_audiobook)
1973        else:
1974            await self.mass.music.mark_item_played(
1975                mass_audiobook,
1976                fully_played=progress.is_finished,
1977                seconds_played=int(progress.current_time),
1978                user_initiated=False,
1979            )
1980
1981    async def _update_playlog_episode(self, progress: MediaProgress) -> None:
1982        # helper progress also ensures no useless progress updates,
1983        # see comment above
1984        self.progress_guard.add_progress(progress.library_item_id, progress.episode_id)
1985        if progress.current_time is None:
1986            return
1987        _episode_id = f"{progress.library_item_id} {progress.episode_id}"
1988        try:
1989            # need to obtain full podcast, and then search for episode
1990            mass_episode = await self.get_podcast_episode(_episode_id, add_progress=False)
1991        except MediaNotFoundError:
1992            return
1993        if int(progress.current_time) == 0 and not progress.is_finished:
1994            await self.mass.music.mark_item_unplayed(mass_episode)
1995        else:
1996            await self.mass.music.mark_item_played(
1997                mass_episode,
1998                fully_played=progress.is_finished,
1999                seconds_played=int(progress.current_time),
2000                user_initiated=False,
2001            )
2002
2003    async def _update_book_narrators(self, library_id: str) -> None:
2004        # narrators are not expanded in ABS' response, so acquire them here
2005        narrators = await self._client.get_library_narrators(library_id=library_id)
2006        audiobook_narrators: dict[str, set[NarratorHelper]] = {}
2007        for narrator in narrators:
2008            async for response in self._client.get_library_items(
2009                library_id=library_id, filter_str=f"narrators.{narrator.id_}"
2010            ):
2011                if not response.results:
2012                    break
2013                for item in response.results:
2014                    narrator_set = audiobook_narrators.get(item.id_, set())
2015                    narrator_set.add(NarratorHelper(id_=narrator.id_, name=narrator.name))
2016                    audiobook_narrators[item.id_] = narrator_set
2017        self.libraries.audiobook_narrators = {
2018            **self.libraries.audiobook_narrators,
2019            **audiobook_narrators,
2020        }
2021
2022    async def _get_audiobook_narrators(
2023        self, book: AbsLibraryItemExpandedBook
2024    ) -> set[NarratorHelper]:
2025        """Get narrators of an audiobook, either from cache or API calls."""
2026        if cached_narrators := self.libraries.audiobook_narrators.get(book.id_):
2027            return cached_narrators
2028        await self._update_book_narrators(book.library_id)
2029        return self.libraries.audiobook_narrators.get(book.id_, set())
2030
2031    async def _cache_set_helper_libraries(self) -> None:
2032        await self.mass.cache.set(
2033            key=CACHE_KEY_LIBRARIES,
2034            provider=self.instance_id,
2035            category=CACHE_CATEGORY_LIBRARIES,
2036            data=self.libraries.to_dict(),
2037        )
2038
2039    def _sync_library_keys(self, libraries: list[Any]) -> None:
2040        """Prune deleted and add new library keys, preserving cached item_ids."""
2041        current_book_ids = {
2042            lib.id_ for lib in libraries if lib.media_type == AbsLibraryMediaType.BOOK
2043        }
2044        current_podcast_ids = {
2045            lib.id_ for lib in libraries if lib.media_type == AbsLibraryMediaType.PODCAST
2046        }
2047        for stale_id in set(self.libraries.audiobooks) - current_book_ids:
2048            del self.libraries.audiobooks[stale_id]
2049        for stale_id in set(self.libraries.podcasts) - current_podcast_ids:
2050            del self.libraries.podcasts[stale_id]
2051        for library in libraries:
2052            if library.media_type == AbsLibraryMediaType.BOOK:
2053                self.libraries.audiobooks.setdefault(library.id_, LibraryHelper(name=library.name))
2054            elif library.media_type == AbsLibraryMediaType.PODCAST:
2055                self.libraries.podcasts.setdefault(library.id_, LibraryHelper(name=library.name))
2056
2057    def _log_no_libraries(self) -> None:
2058        self.logger.error("There are no libraries visible to the Audiobookshelf provider.")
2059
2060    def _log_no_helper_item_ids(self) -> None:
2061        self.logger.warning(
2062            "Cached item ids are missing. "
2063            "Please trigger a full resync of the Audiobookshelf provider manually."
2064        )
2065
2066    @handle_refresh_token
2067    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
2068        """Fetch the personalized views of all libraries and parse them into shelf folders."""
2069        # We have to avoid "flooding" the home page, which becomes especially troublesome if users
2070        # have multiple libraries. Instead we collect per ShelfId, and make sure, that we always get
2071        # roughly the same amount of items per row, no matter the amount of libraries
2072        # List of list (one list per lib) here, such that we can pick the items per lib later.
2073        items_by_shelf_id: dict[AbsShelfId, list[list[MediaItemType | BrowseFolder]]] = {}
2074
2075        all_libraries = {**self.libraries.audiobooks, **self.libraries.podcasts}
2076        max_items_per_row = 20
2077        num_libraries = len(all_libraries)
2078
2079        if num_libraries == 0:
2080            self._log_no_libraries()
2081            return []
2082
2083        limit_items_per_lib = max_items_per_row // num_libraries
2084        limit_items_per_lib = 1 if limit_items_per_lib == 0 else limit_items_per_lib
2085
2086        for library_id in all_libraries:
2087            shelves = await self._client.get_library_personalized_view(
2088                library_id=library_id, limit=limit_items_per_lib
2089            )
2090            await self._recommendations_iter_shelves(shelves, library_id, items_by_shelf_id)
2091
2092        folders: list[RecommendationFolder] = []
2093        for shelf_id, item_lists in items_by_shelf_id.items():
2094            # we have something like [[A, B], [C, D, E], [F]]
2095            # and want [A, C, F, B, D, E]
2096            recommendation_items = [
2097                x
2098                for x in itertools.chain.from_iterable(itertools.zip_longest(*item_lists))
2099                if x is not None
2100            ][:max_items_per_row]
2101
2102            # shelf ids follow pattern:
2103            # recently-added
2104            # newest-episodes
2105            # etc
2106            name = f"{shelf_id.capitalize().replace('-', ' ')}"
2107            folders.append(
2108                RecommendationFolder(
2109                    item_id=f"{shelf_id}",
2110                    name=name,
2111                    icon=ABS_SHELF_ID_ICONS.get(shelf_id),
2112                    translation_key=ABS_SHELF_ID_TRANSLATION_KEY.get(shelf_id),
2113                    items=UniqueList(recommendation_items),
2114                    provider=self.instance_id,
2115                )
2116            )
2117
2118        return folders
2119
2120    def _browse_recommendation_row(self) -> RecommendationFolder:
2121        """Build the static browse row descriptor, without items."""
2122        translation_key = "libraries"
2123        if len(self.libraries.audiobooks) <= 1 and len(self.libraries.podcasts) == 0:
2124            translation_key = "library"
2125        return RecommendationFolder(
2126            item_id="browse",
2127            name="Libraries",
2128            icon="mdi-bookshelf",
2129            translation_key=translation_key,
2130            provider=self.instance_id,
2131        )
2132
2133    def _browse_recommendation_items(
2134        self,
2135    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
2136        """Build the items of the browse row from the known libraries."""
2137        # Browse "recommendation" for convenience. If the user has
2138        # multiple audiobook libraries, we return a listing of them.
2139        # If there is only a single audiobook library, we add the folders
2140        # from _browse_lib_audiobooks, i.e. Authors, Narrators etc.
2141        # Podcast libs do not have filter folders, so always the root folders.
2142        browse_items: list[MediaItemType | BrowseFolder] = []
2143        if len(self.libraries.audiobooks) <= 1:
2144            # audiobooklibs are first, and we have at max 1 audiobook lib
2145            _browse_root = self._browse_root(append_mediatype_suffix=False)
2146            if len(self.libraries.audiobooks) == 0:
2147                browse_items.extend(_browse_root)
2148            else:
2149                assert isinstance(_browse_root[0], BrowseFolder)
2150                _path = _browse_root[0].path
2151                browse_items.extend(self._browse_lib_audiobooks(current_path=_path))
2152                # add podcast roots
2153                browse_items.extend(_browse_root[1:])
2154        else:
2155            browse_items = list(self._browse_root())
2156        return UniqueList(browse_items)
2157
2158    async def _cleanup_open_sessions_loop(self) -> None:
2159        """Close unused open sessions."""
2160        while True:
2161            await asyncio.sleep(STREAMDETAILS_EXPIRATION_S)
2162            current_time = time.time()
2163
2164            async with self.create_session_lock:
2165                sessions_to_close = [
2166                    (session_key, session)
2167                    for session_key, session in self.sessions.items()
2168                    if current_time - session.last_sync_time > (STREAMDETAILS_EXPIRATION_S * 2)
2169                ]
2170                results = await asyncio.gather(
2171                    *(
2172                        self._client.close_open_session(session_id=session.abs_session_id)
2173                        for (_, session) in sessions_to_close
2174                    ),
2175                    return_exceptions=True,
2176                )
2177
2178                for (session_key, session), result in zip(sessions_to_close, results, strict=True):
2179                    if isinstance(result, Exception):
2180                        self.logger.warning(
2181                            "Failed to close session %s: %s",
2182                            session.abs_session_id,
2183                            result,
2184                        )
2185                    else:
2186                        self.logger.debug(
2187                            "Closed session %s",
2188                            session.abs_session_id,
2189                        )
2190                    # We do not try again after a failure. _get_playback_session verifies if a session
2191                    # exists.
2192                    self.sessions.pop(session_key, None)
2193