/
/
/
1"""Helpers for Apple Music playlist browsing."""
2
3from __future__ import annotations
4
5from collections.abc import Sequence
6from dataclasses import dataclass, replace
7from typing import TYPE_CHECKING, Any, cast
8
9from music_assistant_models.enums import MediaType
10from music_assistant_models.errors import MediaNotFoundError
11from music_assistant_models.media_items import BrowseFolder, Playlist, ProviderMapping
12
13from music_assistant.providers.apple_music.helpers.utils import is_library_id
14from music_assistant.providers.apple_music.parsers import parse_playlist
15
16if TYPE_CHECKING:
17 from music_assistant.providers.apple_music.provider import AppleMusicProvider
18
19ROOT_PLAYLIST_FOLDER_ID = "p.playlistsroot"
20# Apple exposes the entire playlist hierarchy under this synthetic root. We walk the
21# tree lazily, fetching the exact branch the user opens instead of preloading.
22
23
24@dataclass(slots=True)
25class AppleMusicPlaylistFolder:
26 """Lightweight representation of a folder node returned by Apple."""
27
28 item_id: str
29 name: str
30
31
32def _folder_path_segment(name: str) -> str:
33 """Return human-readable, path-safe breadcrumb text."""
34 return (name.strip() or "Folder").replace("/", "-").replace("|", "-")
35
36
37def _extract_playlist_folder_id(path_parts: list[str]) -> str | None:
38 """Extract the active folder id from a playlist browse path."""
39 if not path_parts:
40 return None
41 last_segment = path_parts[-1]
42 if "|" in last_segment:
43 return last_segment.rsplit("|", 1)[1]
44 return last_segment
45
46
47def _folder_nodes(
48 provider: AppleMusicProvider,
49 folders: list[AppleMusicPlaylistFolder],
50 base_path: str,
51) -> list[BrowseFolder]:
52 """Convert folder metadata returned by the API into browse nodes."""
53 normalized_base = base_path.rstrip("/")
54 items: list[BrowseFolder] = []
55 for folder in folders:
56 folder_name = folder.name or "Folder"
57 segment_name = _folder_path_segment(folder_name)
58 segment = f"{segment_name}|{folder.item_id}"
59 items.append(
60 BrowseFolder(
61 item_id=f"folder:{folder.item_id}",
62 provider=provider.instance_id,
63 path=f"{normalized_base}/{segment}",
64 name=folder_name,
65 )
66 )
67 return items
68
69
70async def _fetch_playlist_folder_children(
71 provider: AppleMusicProvider,
72 folder_id: str | None = None,
73) -> tuple[list[AppleMusicPlaylistFolder], list[Playlist]]:
74 """Fetch folders/playlists for a single branch of the Apple Music tree."""
75 apple_folder_id = folder_id or ROOT_PLAYLIST_FOLDER_ID
76 endpoint = f"me/library/playlist-folders/{apple_folder_id}/children"
77 try:
78 children = await provider.api_client.get_all_items(endpoint)
79 except MediaNotFoundError:
80 children = []
81 folders: list[AppleMusicPlaylistFolder] = []
82 playlist_entries: list[dict[str, Any]] = []
83 library_playlist_ids: list[str] = []
84 for child in children:
85 child_id = child.get("id")
86 if not child_id:
87 continue
88 child_type = child.get("type")
89 attributes = child.get("attributes") or {}
90 if child_type == "library-playlist-folders":
91 folders.append(
92 AppleMusicPlaylistFolder(
93 item_id=child_id,
94 name=attributes.get("name") or "Folder",
95 )
96 )
97 elif child_type == "library-playlists":
98 playlist_entries.append(child)
99 if is_library_id(child_id):
100 library_playlist_ids.append(child_id)
101 ratings: dict[str, bool] = {}
102 if library_playlist_ids:
103 ratings = await provider.api_client.get_ratings(library_playlist_ids, MediaType.PLAYLIST)
104 playlists: list[Playlist] = []
105 for playlist_entry in playlist_entries:
106 playlist_id = cast("str", playlist_entry.get("id"))
107 is_favourite = ratings.get(playlist_id, False)
108 attributes = playlist_entry.get("attributes") or {}
109 play_params = attributes.get("playParams") or {}
110 global_id = play_params.get("globalId")
111
112 # Start with the original entry, potentially modify it below
113 playlist_obj = playlist_entry
114
115 if attributes.get("hasCatalog") and global_id and not is_library_id(global_id):
116 try:
117 playlist = await provider.get_playlist(global_id, is_favourite)
118 except MediaNotFoundError:
119 provider.logger.debug(
120 "Catalog playlist %s not found, falling back to library metadata",
121 global_id,
122 )
123 playlist_obj = _playlist_without_global_id(playlist_obj)
124 else:
125 playlists.append(_apply_library_id(playlist, playlist_id, provider))
126 continue
127 playlists.append(parse_playlist(provider, playlist_obj, is_favourite))
128 playlists.sort(key=lambda item: (item.name or "").casefold())
129 folders.sort(key=lambda folder: folder.name.casefold())
130 return folders, playlists
131
132
133def _playlist_without_global_id(playlist_obj: dict[str, Any]) -> dict[str, Any]:
134 """
135 Return a shallow copy without a catalog ID.
136
137 Some folders report `hasCatalog=True` but their catalog playlist fetch fails.
138 When that happens we strip the bogus `globalId` so downstream parsing sticks
139 to the library ID (which *can* be resolved).
140 """
141 new_obj = dict(playlist_obj)
142 attributes = dict(new_obj.get("attributes") or {})
143 play_params = dict(attributes.get("playParams") or {})
144 play_params.pop("globalId", None)
145 attributes["playParams"] = play_params
146 new_obj["attributes"] = attributes
147 return new_obj
148
149
150def _apply_library_id(
151 playlist: Playlist, library_id: str, provider: AppleMusicProvider
152) -> Playlist:
153 """
154 Return a copy of `playlist` that always points to the library endpoint.
155
156 `get_playlist` is cached, so mutating the original object would leak those
157 changes to other consumers of the cached catalog playlist. Instead we clone
158 the dataclass with `replace`, swap the ids for this provider instance, and
159 keep the cached object untouched.
160 """
161 new_mappings: set[ProviderMapping] = set()
162 for mapping in playlist.provider_mappings:
163 if mapping.provider_instance == provider.instance_id:
164 new_mappings.add(replace(mapping, item_id=library_id))
165 else:
166 new_mappings.add(mapping)
167 return replace(
168 playlist,
169 item_id=library_id,
170 provider=provider.instance_id,
171 provider_mappings=new_mappings,
172 )
173
174
175async def browse_playlists(
176 provider: AppleMusicProvider, path: str, path_parts: list[str]
177) -> Sequence[BrowseFolder | Playlist]:
178 """Handle playlist browsing for the Apple Music provider."""
179 folder_id: str | None = None
180 base_path = f"{provider.instance_id}://playlists"
181 if len(path_parts) > 1:
182 folder_id = _extract_playlist_folder_id(path_parts[1:])
183 base_path = path.rstrip("/")
184 folders, playlists = await _fetch_playlist_folder_children(provider, folder_id)
185 folder_nodes = _folder_nodes(provider, folders, base_path)
186 return [*folder_nodes, *playlists]
187