music-assistant-server

1"""Library management for Apple Music."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.errors import MusicAssistantError
9from music_assistant_models.media_items import Track
10
11from .helpers.utils import is_catalog_id, is_library_id, translate_media_type_to_apple_type
12from .parsers import parse_album, parse_artist, parse_playlist, parse_track
13
14if TYPE_CHECKING:
15    from collections.abc import AsyncGenerator
16
17    from music_assistant_models.media_items import Album, Artist, MediaItemType, Playlist
18
19    from .provider import AppleMusicProvider
20
21# 100 is the maximum Apple accepts; the heavy includes only cost ~20% latency per page.
22_TRACK_PAGE_SIZE = 100
23
24# Detail lookups for weak-mapped library tracks go out in batches of this many ids.
25_DETAIL_BATCH_SIZE = 100
26
27# Catalog enrichment batch size: 300 (the documented max) returns a 504, so cap at 150. This also
28# bounds the in-flight window, keeping a ~100k library from being materialized at once.
29_TRACK_SYNC_WINDOW = 150
30
31# Limit search fallback attempts per window to avoid rate limits/latency when many IDs are deprecated.
32_MAX_SEARCH_FALLBACK_PER_WINDOW = 10
33
34
35class AppleMusicLibraryManager:
36    """Manages Apple Music library operations."""
37
38    def __init__(self, provider: AppleMusicProvider) -> None:
39        """Initialize library manager."""
40        self.provider = provider
41        self.api = provider.api_client
42        self.logger = provider.logger
43
44    async def get_library_artists(self) -> AsyncGenerator[Artist]:
45        """Retrieve library artists from the provider."""
46        endpoint = "me/library/artists"
47        for item in await self.api.get_all_items(
48            endpoint, include="catalog", extend="editorialNotes"
49        ):
50            if item and item["id"]:
51                yield cast("Artist", parse_artist(self.provider, item))
52
53    async def get_library_albums(self) -> AsyncGenerator[Album]:
54        """Retrieve library albums from the provider."""
55        endpoint = "me/library/albums"
56        album_items = await self.api.get_all_items(
57            endpoint, include="catalog,artists", extend="editorialNotes"
58        )
59        album_catalog_item_ids = [
60            item["id"]
61            for item in album_items
62            if item and item["id"] and not is_library_id(item["id"])
63        ]
64        album_library_item_ids = [
65            item["id"] for item in album_items if item and item["id"] and is_library_id(item["id"])
66        ]
67        rating_catalog_response = await self.api.get_ratings(
68            album_catalog_item_ids, MediaType.ALBUM
69        )
70        rating_library_response = await self.api.get_ratings(
71            album_library_item_ids, MediaType.ALBUM
72        )
73        for item in album_items:
74            if item and item["id"]:
75                is_favourite = (
76                    rating_catalog_response.get(item["id"])
77                    if not is_library_id(item["id"])
78                    else rating_library_response.get(item["id"])
79                )
80                album = parse_album(self.provider, item, is_favourite)
81                if album:
82                    yield cast("Album", album)
83
84    async def get_library_tracks(self) -> AsyncGenerator[Track]:
85        """Retrieve library tracks from the provider."""
86        # Enrich and yield in bounded windows so the full library is never held in memory at once.
87        catalog_items: dict[str, dict[str, Any]] = {}
88        library_only_items: list[dict[str, Any]] = []
89        async for item in self.api.iter_all_items(
90            "me/library/songs", include="catalog,albums,artists", page_size=_TRACK_PAGE_SIZE
91        ):
92            catalog_id = item.get("attributes", {}).get("playParams", {}).get("catalogId")
93            if not catalog_id:
94                library_only_items.append(item)
95            else:
96                catalog_items[catalog_id] = item
97            if len(catalog_items) >= _TRACK_SYNC_WINDOW:
98                async for track in self._flush_catalog_tracks(catalog_items):
99                    yield track
100                catalog_items = {}
101            if len(library_only_items) >= _TRACK_SYNC_WINDOW:
102                async for track in self._flush_library_only_tracks(library_only_items):
103                    yield track
104                library_only_items = []
105        async for track in self._flush_catalog_tracks(catalog_items):
106            yield track
107        async for track in self._flush_library_only_tracks(library_only_items):
108            yield track
109
110    def _track_has_weak_album_mapping(self, track: Track) -> bool:
111        """Return True for missing or name-only album mapping."""
112        if not track.album:
113            return True
114        album_item_id = track.album.item_id
115        return (
116            album_item_id == track.album.name
117            and not is_library_id(album_item_id)
118            and not is_catalog_id(album_item_id)
119        )
120
121    def _apply_album_detail(
122        self,
123        item: dict[str, Any],
124        parsed_track: Track,
125        detail: dict[str, Any],
126        is_favourite: bool | None,
127    ) -> Track:
128        """Return the detail-based track when it resolves the album the listing lacked."""
129        detailed_track = parse_track(self.provider, detail, is_favourite)
130        if self._track_has_weak_album_mapping(detailed_track):
131            # Keep detail album fallback if list had no album.
132            if not parsed_track.album and detailed_track.album:
133                return detailed_track
134            self.logger.debug(
135                "Library song %s still has no resolvable album mapping after detail fetch",
136                item["id"],
137            )
138            return parsed_track
139        return detailed_track
140
141    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
142        """Retrieve playlists from the provider."""
143        endpoint = "me/library/playlists"
144        playlist_items = await self.api.get_all_items(endpoint)
145        playlist_library_item_ids = [
146            item["id"]
147            for item in playlist_items
148            if item and item["id"] and is_library_id(item["id"])
149        ]
150        rating_library_response = await self.api.get_ratings(
151            playlist_library_item_ids, MediaType.PLAYLIST
152        )
153        for item in playlist_items:
154            is_favourite = rating_library_response.get(item["id"], False)
155            # Fetch catalog metadata, but keep library ID for write operations.
156            if item["attributes"]["hasCatalog"]:
157                yield await self.provider.media_manager.get_playlist(
158                    item["attributes"]["playParams"]["globalId"],
159                    is_favourite,
160                    can_edit_hint=item["attributes"].get("canEdit"),
161                    library_id_override=item["id"] if is_library_id(item["id"]) else None,
162                )
163            elif item and item["id"]:
164                yield parse_playlist(self.provider, item, is_favourite)
165
166    async def library_add(self, item: MediaItemType) -> None:
167        """Add item to library."""
168        if item.media_type == MediaType.ARTIST:
169            # The POST /v1/me/library endpoint does not support ids[artists];
170            # artists appear in the library implicitly via their albums/songs.
171            self.logger.debug(
172                "Skipping library_add for artist %s: Apple Music does not support "
173                "adding artists directly via the API.",
174                item.name,
175            )
176            return
177        item_type = translate_media_type_to_apple_type(item.media_type)
178        kwargs = {f"ids[{item_type}]": item.item_id}
179        await self.api.post_data("me/library", **kwargs)
180
181    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> None:
182        """Remove item from library."""
183        self.logger.debug(
184            "Deleting items from your library is not yet supported by the Apple Music API. "
185            f"Skipping deletion of {media_type} - {prov_item_id}."
186        )
187
188    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
189        """Add track(s) to playlist."""
190        endpoint = f"me/library/playlists/{prov_playlist_id}/tracks"
191        data = {
192            "data": [
193                {
194                    "id": track_id,
195                    "type": "library-songs" if is_library_id(track_id) else "songs",
196                }
197                for track_id in prov_track_ids
198            ]
199        }
200        await self.api.post_data(endpoint, data=data)
201
202    async def remove_playlist_tracks(
203        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
204    ) -> None:
205        """Remove track(s) from playlist."""
206        message = (
207            "Removing tracks from playlists is not supported by the Apple Music "
208            "API. Make sure to delete them using the Apple Music app."
209        )
210        raise MusicAssistantError(message)
211
212    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
213        """Set the favorite status of an item."""
214        data = {
215            "type": "ratings",
216            "attributes": {"value": 1 if favorite else -1},
217        }
218        item_type = translate_media_type_to_apple_type(media_type)
219        if is_catalog_id(prov_item_id):
220            endpoint = f"me/ratings/{item_type}/{prov_item_id}"
221        else:
222            endpoint = f"me/ratings/library-{item_type}/{prov_item_id}"
223        await self.api.put_data(endpoint, data=data)
224
225    async def _flush_catalog_tracks(
226        self, library_items_by_catalog_id: dict[str, dict[str, Any]]
227    ) -> AsyncGenerator[Track]:
228        """Enrich one window of catalog-backed library tracks with catalog detail and yield them."""
229        if not library_items_by_catalog_id:
230            return
231        catalog_ids = list(library_items_by_catalog_id)
232        catalog_endpoint = f"catalog/{self.provider._storefront}/songs"
233        response = await self.api.get_data(
234            catalog_endpoint, ids=",".join(catalog_ids), include="artists,albums"
235        )
236        rating_response = await self.api.get_ratings(catalog_ids, MediaType.TRACK)
237        returned_catalog_ids: set[str] = set()
238        for item in response.get("data", []):
239            returned_catalog_ids.add(item["id"])
240            is_favourite = rating_response.get(item["id"])
241            parsed_track = parse_track(self.provider, item, is_favourite)
242            if self._track_has_weak_album_mapping(parsed_track) and (
243                library_item := library_items_by_catalog_id.get(item["id"])
244            ):
245                parsed_library_track = parse_track(self.provider, library_item, is_favourite)
246                if parsed_library_track.album and not self._track_has_weak_album_mapping(
247                    parsed_library_track
248                ):
249                    parsed_track.album = parsed_library_track.album
250            yield parsed_track
251        # Handle deprecated catalog IDs: search replacement with per-window limit
252        search_attempts = 0
253        for missing_catalog_id in (cid for cid in catalog_ids if cid not in returned_catalog_ids):
254            if library_item := library_items_by_catalog_id.get(missing_catalog_id):
255                library_item_id = library_item.get("id")
256                is_favourite = rating_response.get(missing_catalog_id)
257
258                # Limit search attempts per window to avoid API rate limits
259                if search_attempts >= _MAX_SEARCH_FALLBACK_PER_WINDOW:
260                    # Mark remaining as unavailable without attempting search
261                    parsed_track = parse_track(self.provider, library_item, is_favourite)
262                    for mapping in parsed_track.provider_mappings:
263                        if mapping.provider_instance == self.provider.instance_id:
264                            mapping.available = False
265                    self.logger.debug(
266                        "Skipping search fallback for %s (reached window limit of %d searches)",
267                        library_item_id,
268                        _MAX_SEARCH_FALLBACK_PER_WINDOW,
269                    )
270                    yield parsed_track
271                    continue
272
273                search_attempts += 1
274
275                # Try to find current catalog version via search
276                replacement_track = await self._try_search_replacement_for_deprecated_track(
277                    library_item, is_favourite
278                )
279
280                if replacement_track:
281                    yield replacement_track
282                else:
283                    # No replacement found - yield library-only track but mark unavailable
284                    # (these often have corrupt streams from Apple's deprecated catalog versions)
285                    parsed_track = parse_track(self.provider, library_item, is_favourite)
286                    for mapping in parsed_track.provider_mappings:
287                        if mapping.provider_instance == self.provider.instance_id:
288                            mapping.available = False
289                    self.logger.debug(
290                        "Library track %s references deprecated catalog ID %s - marked unavailable",
291                        library_item_id,
292                        missing_catalog_id,
293                    )
294                    yield parsed_track
295
296    async def _try_search_replacement_for_deprecated_track(
297        self, library_item: dict[str, Any], is_favourite: bool | None
298    ) -> Track | None:
299        """
300        Try to find a current catalog version for a deprecated library track via search.
301
302        Returns the replacement track if found, None otherwise.
303        """
304        attributes = library_item.get("attributes", {})
305        track_name = attributes.get("name")
306        artist_name = attributes.get("artistName")
307        album_name = attributes.get("albumName")
308
309        if not track_name or not artist_name:
310            return None
311
312        # Search for track: "Artist Track"
313        search_query = f"{artist_name} {track_name}"
314        try:
315            search_results = await self.provider.media_manager.search(
316                search_query, [MediaType.TRACK], limit=10
317            )
318
319            if not search_results.tracks:
320                return None
321
322            # Try to find exact match (case-insensitive)
323            track_name_lower = track_name.lower()
324            artist_name_lower = artist_name.lower()
325            album_name_lower = album_name.lower() if album_name else None
326
327            for track in search_results.tracks:
328                # Skip ItemMapping entries (only interested in full Track objects)
329                if not isinstance(track, Track):
330                    continue
331
332                # Check track name match
333                if track.name.lower() != track_name_lower:
334                    continue
335
336                # Check artist match
337                if not any(a.name.lower() == artist_name_lower for a in track.artists):
338                    continue
339
340                # If we have album info, require album match
341                if album_name_lower:
342                    if not track.album or track.album.name.lower() != album_name_lower:
343                        # Album mismatch or missing - might be a different version/remaster
344                        continue
345
346                # Found a match! Update favorite status and return
347                track.favorite = is_favourite or False
348                self.logger.debug(
349                    "Found replacement catalog track %s for deprecated library track %s",
350                    track.item_id,
351                    library_item.get("id"),
352                )
353                return track
354
355            return None
356
357        except Exception as err:
358            self.logger.debug(
359                "Search fallback failed for track '%s' by '%s': %s",
360                track_name,
361                artist_name,
362                err,
363                exc_info=True,
364            )
365            return None
366
367    async def _flush_library_only_tracks(
368        self, library_only_items: list[dict[str, Any]]
369    ) -> AsyncGenerator[Track]:
370        """Enrich one window of library-only tracks (no catalog id) and yield them."""
371        if not library_only_items:
372            return
373        library_ids = [item["id"] for item in library_only_items if item and item["id"]]
374        rating_response = await self.api.get_ratings(library_ids, MediaType.TRACK)
375        parsed_tracks = [
376            (item, parse_track(self.provider, item, rating_response.get(item["id"])))
377            for item in library_only_items
378        ]
379        details = await self._fetch_library_song_details(
380            [
381                item["id"]
382                for item, track in parsed_tracks
383                if self._track_has_weak_album_mapping(track)
384            ]
385        )
386        for item, parsed_track in parsed_tracks:
387            if (detail := details.get(item["id"])) is None:
388                yield parsed_track
389                continue
390            yield self._apply_album_detail(
391                item, parsed_track, detail, rating_response.get(item["id"])
392            )
393
394    async def _fetch_library_song_details(
395        self, library_ids: list[str]
396    ) -> dict[str, dict[str, Any]]:
397        """Return the detailed library-song items for the given ids, keyed by id."""
398        details: dict[str, dict[str, Any]] = {}
399        for offset in range(0, len(library_ids), _DETAIL_BATCH_SIZE):
400            batch = library_ids[offset : offset + _DETAIL_BATCH_SIZE]
401            try:
402                response = await self.api.get_data(
403                    "me/library/songs", ids=",".join(batch), include="catalog,albums,artists"
404                )
405            except MusicAssistantError as err:
406                # the listing parse stays usable, so a failed batch only costs album detail
407                self.logger.warning(
408                    "Unable to fetch library song details for %s tracks: %s", len(batch), err
409                )
410                continue
411            details.update(
412                {item["id"]: item for item in response.get("data", []) if item.get("id")}
413            )
414        return details
415