def parse_playlist(playlist_obj: dict[str, Any], provider: SpotifyProvider) -> Playlist:
"""Parse spotify playlist object to generic layout."""
+ owner_id = playlist_obj["owner"].get("id", "")
is_editable = (
- provider._sp_user is not None and playlist_obj["owner"]["id"] == provider._sp_user["id"]
+ provider._sp_user is not None and owner_id == provider._sp_user["id"]
) or playlist_obj["collaborative"]
+ # Spotify-owned playlists (Daily Mix, Discover Weekly, etc.) are personalized per user
+ is_spotify_owned = owner_id.lower() == "spotify"
+
# Get owner name with fallback
owner_name = playlist_obj["owner"].get("display_name")
if owner_name is None and provider._sp_user is not None:
owner_name = provider._sp_user["display_name"]
+ # Mark as unique if user-owned/editable OR if it's a Spotify personalized playlist
+ is_unique = is_editable or is_spotify_owned
+
playlist = Playlist(
item_id=playlist_obj["id"],
provider=provider.instance_id,
provider_domain=provider.domain,
provider_instance=provider.instance_id,
url=playlist_obj["external_urls"]["spotify"],
- is_unique=is_editable, # user-owned playlists are unique
+ is_unique=is_unique,
)
},
is_editable=is_editable,
yield audiobook
async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
- """Retrieve playlists from the provider."""
+ """Retrieve playlists from the provider.
+
+ Note: We use the global session here because playlists like "Daily Mix"
+ are only returned when using the non-dev (global) token.
+ """
yield await self._get_liked_songs_playlist()
- async for item in self._get_all_items("me/playlists"):
+ async for item in self._get_all_items("me/playlists", use_global_session=True):
if item and item["id"]:
yield parse_playlist(item, self)
if prov_playlist_id == self._get_liked_songs_playlist_id():
return await self._get_liked_songs_playlist()
- playlist_obj = await self._get_data(f"playlists/{prov_playlist_id}")
+ # Use global session for Spotify-owned playlists (e.g., Daily Mix)
+ # as they may not be accessible via the dev token
+ use_global = await self._is_spotify_owned_playlist(prov_playlist_id)
+ playlist_obj = await self._get_data(
+ f"playlists/{prov_playlist_id}", use_global_session=use_global
+ )
return parse_playlist(playlist_obj, self)
@use_cache()
if prov_playlist_id == self._get_liked_songs_playlist_id()
else f"playlists/{prov_playlist_id}/tracks"
)
+ # Use global session for liked songs or Spotify-owned playlists (e.g., Daily Mix)
+ use_global = is_liked_songs or await self._is_spotify_owned_playlist(prov_playlist_id)
# do single request to get the etag (which we use as checksum for caching)
- cache_checksum = await self._get_etag(
- uri, limit=1, offset=0, use_global_session=is_liked_songs
- )
+ cache_checksum = await self._get_etag(uri, limit=1, offset=0, use_global_session=use_global)
page_size = 50
offset = page * page_size
spotify_result = await self._get_data_with_caching(
- uri, cache_checksum, limit=page_size, offset=offset, use_global_session=is_liked_songs
+ uri, cache_checksum, limit=page_size, offset=offset, use_global_session=use_global
)
for index, item in enumerate(spotify_result["items"], 1):
if not (item and item["track"] and item["track"]["id"]):
# Don't unload - we can still use the global session
self.dev_session_active = False
self.logger.warning(str(err))
+ raise
# make sure that our updated creds get stored in memory + config
self._auth_info_dev = auth_info
return liked_songs
+ @use_cache(86400 * 90)
+ async def _is_spotify_owned_playlist(self, prov_playlist_id: str) -> bool:
+ """Check if a playlist is owned by Spotify.
+
+ Spotify-owned playlists (e.g., Daily Mix, Discover Weekly) are only accessible
+ via the global token, not through developer API tokens.
+
+ :param prov_playlist_id: The Spotify playlist ID.
+ :returns: True if the playlist is owned by Spotify.
+ """
+ if prov_playlist_id == self._get_liked_songs_playlist_id():
+ return False
+ try:
+ # We need to use global session here to actually get the playlist info
+ # because if it's a Spotify-owned playlist, the dev session won't have access
+ playlist_obj = await self._get_data(
+ f"playlists/{prov_playlist_id}",
+ fields="owner.id",
+ use_global_session=True,
+ )
+ owner_id = playlist_obj.get("owner", {}).get("id", "").lower()
+ return bool(owner_id == "spotify")
+ except MediaNotFoundError:
+ return False
+
async def _add_audiobook_chapters(self, audiobook: Audiobook) -> None:
"""Add chapter metadata to an audiobook from Spotify API data."""
try: