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