Workaround for "Youtube Music playlist stalls on uploaded music" music-assistant/support#4469

music_assistant/providers/ytmusic/__init__.py

@@ -328,15 +328,15 @@ async def get_library_podcasts(self) -> AsyncGenerator[Podcast, None]:
     @use_cache(3600 * 24 * 30)  # Cache for 30 days
     async def get_album(self, prov_album_id) -> Album:
         """Get full album details by id."""
-        if album_obj := await get_album(prov_album_id=prov_album_id, language=self.language):
+        if album_obj := await get_album(headers=self._headers, prov_album_id=prov_album_id, language=self.language, user=self._yt_user):
             return self._parse_album(album_obj=album_obj, album_id=prov_album_id)
         msg = f"Item {prov_album_id} not found"
         raise MediaNotFoundError(msg)
 
     @use_cache(3600 * 24 * 30)  # Cache for 30 days
     async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
         """Get album tracks for given album id."""
-        album_obj = await get_album(prov_album_id=prov_album_id, language=self.language)
+        album_obj = await get_album(headers=self._headers, prov_album_id=prov_album_id, language=self.language, user=self._yt_user)
         if not album_obj.get("tracks"):
             return []
         tracks = []

music_assistant/providers/ytmusic/helpers.py

@@ -37,16 +37,25 @@ def _get_artist():
     return await asyncio.to_thread(_get_artist)
 
 
-async def get_album(prov_album_id: str, language: str = "en") -> dict[str, str]:
+async def get_album(headers: dict[str, str], prov_album_id: str, language: str = "en", user: str | None = None) -> dict[str, str]:
     """Async wrapper around the ytmusicapi get_album function."""
 
     def _get_album():
-        ytm = ytmusicapi.YTMusic(language=language)
-        album = ytm.get_album(browseId=prov_album_id)
+        if prov_album_id.startswith("FEmusic_library_privately_owned_release"):
+            ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
+            album = ytm.get_library_upload_album(browseId=prov_album_id)
+        else:
+            ytm = ytmusicapi.YTMusic(language=language)
+            album = ytm.get_album(browseId=prov_album_id)
+
         if "audioPlaylistId" in album:
             # Track id's from album tracks do not match with actual album tracks. E.g. a track
             # points to the videoId of the original version, while we want the album version
-            album_playlist = ytm.get_playlist(playlistId=album["audioPlaylistId"], limit=None)
+            try:
+                album_playlist = ytm.get_playlist(playlistId=album["audioPlaylistId"], limit=None)
+            except:
+                return album
+
             # Do some basic checks
             if len(album_playlist.get("tracks", [])) != len(album.get("tracks", [])):
                 return album
@@ -57,8 +66,7 @@ def _get_album():
                     album_track["videoId"] = playlist_track["videoId"]
                     album_track["isAvailable"] = playlist_track.get("isAvailable", True)
                     album_track["likeStatus"] = playlist_track.get("likeStatus", "INDIFFERENT")
-            return album
-        return ytm.get_album(browseId=prov_album_id)
+        return album
 
     return await asyncio.to_thread(_get_album)
 
@@ -238,7 +246,7 @@ async def library_add_remove_album(
     headers: dict[str, str], prov_item_id: str, add: bool = True, user: str | None = None
 ) -> bool:
     """Add or remove an album or playlist to the user's library."""
-    album = await get_album(prov_album_id=prov_item_id)
+    album = await get_album(headers=headers, prov_album_id=prov_item_id, user=user)
 
     def _library_add_remove_album():
         ytm = ytmusicapi.YTMusic(auth=headers, user=user)