Improve Spotify stream start time and fallback

music_assistant/providers/spotify/provider.py

@@ -202,7 +202,6 @@ async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
             if item and item["id"]:
                 yield parse_playlist(item, self)
 
-    # ruff: noqa: PLR0915
     @use_cache()
     async def search(
         self, search_query: str, media_types: list[MediaType] | None = None, limit: int = 5
@@ -214,9 +213,32 @@ async def search(
         :param limit: Number of items to return in the search (per type).
         """
         searchresult = SearchResults()
-        searchtypes = []
         if media_types is None:
             return searchresult
+
+        searchtype = self._build_search_types(media_types)
+        if not searchtype:
+            return searchresult
+
+        search_query = search_query.replace("'", "")
+        offset = 0
+        page_limit = min(limit, 50)
+
+        while True:
+            api_result = await self._get_data(
+                "search", q=search_query, type=searchtype, limit=page_limit, offset=offset
+            )
+            items_received = self._process_search_results(api_result, searchresult)
+
+            offset += page_limit
+            if offset >= limit or items_received < page_limit:
+                break
+
+        return searchresult
+
+    def _build_search_types(self, media_types: list[MediaType]) -> str:
+        """Build comma-separated search types string from media types."""
+        searchtypes = []
         if MediaType.ARTIST in media_types:
             searchtypes.append("artist")
         if MediaType.ALBUM in media_types:
@@ -229,75 +251,76 @@ async def search(
             searchtypes.append("show")
         if MediaType.AUDIOBOOK in media_types and self.audiobooks_supported:
             searchtypes.append("audiobook")
-        if not searchtypes:
-            return searchresult
-        searchtype = ",".join(searchtypes)
-        search_query = search_query.replace("'", "")
-        offset = 0
-        page_limit = min(limit, 50)
-        while True:
-            items_received = 0
-            api_result = await self._get_data(
-                "search", q=search_query, type=searchtype, limit=page_limit, offset=offset
-            )
-            if "artists" in api_result:
-                artists = [
-                    parse_artist(item, self)
-                    for item in api_result["artists"]["items"]
-                    if (item and item["id"] and item["name"])
-                ]
-                searchresult.artists = [*searchresult.artists, *artists]
-                items_received += len(api_result["artists"]["items"])
-            if "albums" in api_result:
-                albums = [
-                    parse_album(item, self)
-                    for item in api_result["albums"]["items"]
-                    if (item and item["id"])
-                ]
-                searchresult.albums = [*searchresult.albums, *albums]
-                items_received += len(api_result["albums"]["items"])
-            if "tracks" in api_result:
-                tracks = [
-                    parse_track(item, self)
-                    for item in api_result["tracks"]["items"]
-                    if (item and item["id"])
-                ]
-                searchresult.tracks = [*searchresult.tracks, *tracks]
-                items_received += len(api_result["tracks"]["items"])
-            if "playlists" in api_result:
-                playlists = [
-                    parse_playlist(item, self)
-                    for item in api_result["playlists"]["items"]
-                    if (item and item["id"])
-                ]
-                searchresult.playlists = [*searchresult.playlists, *playlists]
-                items_received += len(api_result["playlists"]["items"])
-            if "shows" in api_result:
-                podcasts = []
-                for item in api_result["shows"]["items"]:
-                    if not (item and item["id"]):
-                        continue
-                    # Filter out audiobooks - they have a distinctive description format
-                    description = item.get("description", "")
-                    if description.startswith("Author(s):") and "Narrator(s):" in description:
-                        continue
-                    podcasts.append(parse_podcast(item, self))
-                searchresult.podcasts = [*searchresult.podcasts, *podcasts]
-                items_received += len(api_result["shows"]["items"])
-            if "audiobooks" in api_result and self.audiobooks_supported:
-                audiobooks = [
-                    parse_audiobook(item, self)
-                    for item in api_result["audiobooks"]["items"]
-                    if (item and item["id"])
-                ]
-                searchresult.audiobooks = [*searchresult.audiobooks, *audiobooks]
-                items_received += len(api_result["audiobooks"]["items"])
-            offset += page_limit
-            if offset >= limit:
-                break
-            if items_received < page_limit:
-                break
-        return searchresult
+        return ",".join(searchtypes)
+
+    def _process_search_results(
+        self, api_result: dict[str, Any], searchresult: SearchResults
+    ) -> int:
+        """Process API search results and update searchresult object.
+
+        Returns the total number of items received.
+        """
+        items_received = 0
+
+        if "artists" in api_result:
+            artists = [
+                parse_artist(item, self)
+                for item in api_result["artists"]["items"]
+                if (item and item["id"] and item["name"])
+            ]
+            searchresult.artists = [*searchresult.artists, *artists]
+            items_received += len(api_result["artists"]["items"])
+
+        if "albums" in api_result:
+            albums = [
+                parse_album(item, self)
+                for item in api_result["albums"]["items"]
+                if (item and item["id"])
+            ]
+            searchresult.albums = [*searchresult.albums, *albums]
+            items_received += len(api_result["albums"]["items"])
+
+        if "tracks" in api_result:
+            tracks = [
+                parse_track(item, self)
+                for item in api_result["tracks"]["items"]
+                if (item and item["id"])
+            ]
+            searchresult.tracks = [*searchresult.tracks, *tracks]
+            items_received += len(api_result["tracks"]["items"])
+
+        if "playlists" in api_result:
+            playlists = [
+                parse_playlist(item, self)
+                for item in api_result["playlists"]["items"]
+                if (item and item["id"])
+            ]
+            searchresult.playlists = [*searchresult.playlists, *playlists]
+            items_received += len(api_result["playlists"]["items"])
+
+        if "shows" in api_result:
+            podcasts = []
+            for item in api_result["shows"]["items"]:
+                if not (item and item["id"]):
+                    continue
+                # Filter out audiobooks - they have a distinctive description format
+                description = item.get("description", "")
+                if description.startswith("Author(s):") and "Narrator(s):" in description:
+                    continue
+                podcasts.append(parse_podcast(item, self))
+            searchresult.podcasts = [*searchresult.podcasts, *podcasts]
+            items_received += len(api_result["shows"]["items"])
+
+        if "audiobooks" in api_result and self.audiobooks_supported:
+            audiobooks = [
+                parse_audiobook(item, self)
+                for item in api_result["audiobooks"]["items"]
+                if (item and item["id"])
+            ]
+            searchresult.audiobooks = [*searchresult.audiobooks, *audiobooks]
+            items_received += len(api_result["audiobooks"]["items"])
+
+        return items_received
 
     @use_cache()
     async def get_artist(self, prov_artist_id: str) -> Artist:
@@ -642,20 +665,13 @@ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[
     async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
         """Return content details for the given track/episode/audiobook when it will be streamed."""
         if media_type == MediaType.AUDIOBOOK and self.audiobooks_supported:
-            chapters_data = await self._get_audiobook_chapters_data(item_id)
-            if not chapters_data:
-                raise MediaNotFoundError(f"No chapters found for audiobook {item_id}")
-
-            # Calculate total duration and convert to seconds for StreamDetails
-            total_duration_ms = sum(chapter.get("duration_ms", 0) for chapter in chapters_data)
-            duration_seconds = total_duration_ms // 1000
-
-            # Create chapter URIs for streaming
-            chapter_uris = []
-            for chapter in chapters_data:
-                chapter_id = chapter["id"]
-                chapter_uri = f"spotify://episode:{chapter_id}"
-                chapter_uris.append(chapter_uri)
+            # Don't fetch chapters here - do it lazily during streaming
+            # Just get basic info for duration estimate
+            try:
+                audiobook_obj = await self._get_data(f"audiobooks/{item_id}")
+                duration_seconds = audiobook_obj.get("duration_ms", 0) // 1000
+            except Exception:
+                duration_seconds = 0
 
             return StreamDetails(
                 item_id=item_id,
@@ -666,7 +682,7 @@ async def get_stream_details(self, item_id: str, media_type: MediaType) -> Strea
                 allow_seek=True,
                 can_seek=True,
                 duration=duration_seconds,
-                data={"chapters": chapter_uris, "chapters_data": chapters_data},
+                data={"fetch_chapters_on_stream": True},  # Flag to fetch during streaming
             )
 
         # For all other media types (tracks, podcast episodes)
@@ -685,20 +701,23 @@ async def get_audio_stream(
     ) -> AsyncGenerator[bytes, None]:
         """Get audio stream from Spotify via librespot."""
         if streamdetails.media_type == MediaType.AUDIOBOOK and isinstance(streamdetails.data, dict):
-            chapter_uris = streamdetails.data.get("chapters", [])
-            chapters_data = streamdetails.data.get("chapters_data", [])
+            # Fetch chapters NOW if not already provided
+            if streamdetails.data.get("fetch_chapters_on_stream"):
+                chapters_data = await self._get_audiobook_chapters_data(streamdetails.item_id)
+                chapter_uris = [f"spotify://episode:{ch['id']}" for ch in chapters_data]
+            else:
+                chapter_uris = streamdetails.data.get("chapters", [])
+                chapters_data = streamdetails.data.get("chapters_data", [])
 
-            # Calculate which chapter to start from based on seek_position
+            # Calculate start chapter
             seek_position_ms = seek_position * 1000
             current_seek_ms = seek_position_ms
             start_chapter = 0
 
             if seek_position > 0 and chapters_data:
                 accumulated_duration_ms = 0
-
                 for i, chapter_data in enumerate(chapters_data):
                     chapter_duration_ms = chapter_data.get("duration_ms", 0)
-
                     if accumulated_duration_ms + chapter_duration_ms > seek_position_ms:
                         start_chapter = i
                         current_seek_ms = seek_position_ms - accumulated_duration_ms
@@ -708,20 +727,59 @@ async def get_audio_stream(
                     start_chapter = len(chapter_uris) - 1
                     current_seek_ms = 0
 
-            # Convert back to seconds for librespot
             current_seek_seconds = int(current_seek_ms // 1000)
 
-            # Stream chapters starting from the calculated position
-            for i in range(start_chapter, len(chapter_uris)):
-                chapter_uri = chapter_uris[i]
-                chapter_seek = current_seek_seconds if i == start_chapter else 0
-
-                try:
-                    async for chunk in self.streamer.stream_spotify_uri(chapter_uri, chapter_seek):
+            # Try streaming with quick_fail for faster fallback detection
+            first_chunk_received = False
+            try:
+                for i in range(start_chapter, len(chapter_uris)):
+                    chapter_uri = chapter_uris[i]
+                    chapter_seek = current_seek_seconds if i == start_chapter else 0
+
+                    # Use quick_fail=True for first chapter to enable faster fallback
+                    use_quick_fail = i == start_chapter
+                    async for chunk in self.streamer.stream_spotify_uri(
+                        chapter_uri, chapter_seek, quick_fail=use_quick_fail
+                    ):
+                        first_chunk_received = True
                         yield chunk
-                except Exception as e:
-                    self.logger.error(f"Chapter {i + 1} streaming failed: {e}")
-                    continue
+
+                    # If we haven't received any chunks by now, trigger fallback
+                    if not first_chunk_received:
+                        raise Exception(f"No audio data received for chapter {i + 1}")
+
+                return  # Success
+
+            except Exception as e:
+                # Try fallback if we haven't successfully streamed any data yet
+                if not first_chunk_received:
+                    self.logger.warning(
+                        f"Audiobook streaming failed on {self.instance_id}: "
+                        f"{e}. Trying other instances..."
+                    )
+
+                    for provider in self.mass.music.providers:
+                        if (
+                            provider.domain == "spotify"
+                            and provider.instance_id != self.instance_id
+                            and isinstance(provider, SpotifyProvider)
+                            and provider.audiobooks_supported
+                        ):
+                            try:
+                                async for chunk in provider.get_audio_stream(
+                                    streamdetails, seek_position
+                                ):
+                                    yield chunk
+                                return
+                            except Exception as retry_error:
+                                self.logger.debug(
+                                    f"Instance {provider.instance_id} also failed: {retry_error}"
+                                )
+                                continue
+
+                raise UnsupportedFeaturedException(
+                    f"No Spotify instance can play this audiobook. Original error: {e}"
+                )
         else:
             # Handle normal tracks and podcast episodes
             async for chunk in self.streamer.get_audio_stream(streamdetails, seek_position):
@@ -747,6 +805,7 @@ async def login(self, force_refresh: bool = False) -> dict[str, Any]:
             "refresh_token": refresh_token,
             "client_id": client_id,
         }
+        err = "Unknown error"
         for _ in range(2):
             async with self.mass.http_session.post(
                 "https://accounts.spotify.com/api/token", data=params

music_assistant/providers/spotify/streaming.py

@@ -41,11 +41,17 @@ async def get_audio_stream(
             yield chunk
 
     async def stream_spotify_uri(
-        self, spotify_uri: str, seek_position: int = 0
+        self, spotify_uri: str, seek_position: int = 0, quick_fail: bool = False
     ) -> AsyncGenerator[bytes, None]:
         """Stream a Spotify URI using librespot.
 
         This internal method handles the entire process, from authentication to playback.
+
+        Args:
+            spotify_uri: The Spotify URI to stream
+            seek_position: Position in seconds to start from
+            quick_fail: If True, use shorter timeout for faster failure
+                                 detection (useful for fallback scenarios)
         """
         # Validate that librespot binary is available
         if not self.provider._librespot_bin:
@@ -70,9 +76,15 @@ async def stream_spotify_uri(
         if seek_position:
             args += ["--start-position", str(int(seek_position))]
 
-        # we retry twice in case librespot fails to start
-        for attempt in (1, 2):
-            log_librespot = self.provider.logger.isEnabledFor(VERBOSE_LOG_LEVEL) or attempt == 2
+        # Use shorter timeout if quick_fail is enabled (for audiobook fallback scenarios)
+        base_timeout = 3 if quick_fail else 5
+        max_attempts = 1 if quick_fail else 2
+
+        # we retry twice in case librespot fails to start (or once if quick_fail)
+        for attempt in range(1, max_attempts + 1):
+            log_librespot = (
+                self.provider.logger.isEnabledFor(VERBOSE_LOG_LEVEL) or attempt == max_attempts
+            )
             async with AsyncProcess(
                 args,
                 stdout=True,
@@ -82,13 +94,15 @@ async def stream_spotify_uri(
                 # get first chunk with timeout, to catch the issue where librespot is not starting
                 # which seems to happen from time to time (but rarely)
                 try:
-                    chunk = await asyncio.wait_for(librespot_proc.read(64000), timeout=10 * attempt)
+                    chunk = await asyncio.wait_for(
+                        librespot_proc.read(64000), timeout=base_timeout * attempt
+                    )
                     if not chunk:
                         raise AudioError(f"No audio data received from librespot for {spotify_uri}")
                     yield chunk
                 except (TimeoutError, AudioError):
                     err_mesg = f"No audio received from librespot within timeout for {spotify_uri}"
-                    if attempt == 2:
+                    if attempt == max_attempts:
                         raise AudioError(err_mesg)
                     self.provider.logger.warning("%s - will retry once", err_mesg)
                     continue