/
/
/
1"""Recommendations and station helpers for Apple Music."""
2
3from __future__ import annotations
4
5import hashlib
6from typing import TYPE_CHECKING, cast
7
8from aiohttp import ClientResponseError
9from music_assistant_models.enums import MediaType
10from music_assistant_models.errors import MediaNotFoundError
11from music_assistant_models.media_items import (
12 Artist,
13 ItemMapping,
14 Playlist,
15 RecommendationFolder,
16 Track,
17)
18
19from music_assistant.controllers.cache import use_cache
20
21from .parsers import parse_artist, parse_station_as_playlist, parse_track
22
23if TYPE_CHECKING:
24 from .provider import AppleMusicProvider
25
26
27def _slugify_title(title: str) -> str:
28 """Return a stable, alphanumeric slug for a recommendation folder title."""
29 slug = "".join(c for c in title.lower().replace(" ", "_") if c.isalnum() or c == "_")
30 if not slug:
31 # Fall back to a hash for titles that yield no alphanumeric characters.
32 slug = hashlib.md5(title.encode(), usedforsecurity=False).hexdigest()[:12]
33 return slug
34
35
36class AppleMusicRecommendationManager:
37 """Handles recommendations, stations, and similar-track lookups."""
38
39 def __init__(self, provider: AppleMusicProvider) -> None:
40 """Initialize recommendation manager."""
41 self.provider = provider
42 self._station_id_to_name: dict[str, str] = {}
43 self._station_name_to_id: dict[str, str] = {}
44 self.mass = provider.mass
45 self.instance_id = provider.instance_id
46 self.domain = provider.domain
47 self.api = provider.api_client
48 self.logger = provider.logger
49
50 @use_cache(3600 * 24, allow_expired_cache=True)
51 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
52 """
53 Retrieve tracks similar to the provided track.
54
55 :param prov_track_id: The Apple Music track ID.
56 :param limit: Maximum number of tracks to return.
57 """
58 if limit <= 0:
59 return []
60 endpoint = f"me/stations/next-tracks/ra.{prov_track_id}"
61 try:
62 response = await self.api.post_data(endpoint, include="artists")
63 except ClientResponseError as err:
64 if err.status == 500:
65 self.logger.debug("Similar tracks unavailable for %s (%s)", prov_track_id, endpoint)
66 return []
67 raise
68 if not response:
69 return []
70 tracks = [track for track in response.get("data", []) if track and track.get("id")][:limit]
71 if not tracks:
72 return []
73 track_ids = [track["id"] for track in tracks]
74 rating_response = await self.api.get_ratings(track_ids, MediaType.TRACK)
75 return [
76 parse_track(self.provider, track, rating_response.get(track["id"])) for track in tracks
77 ]
78
79 @use_cache(3600 * 24)
80 async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
81 """Retrieve a list of artists similar to the provided artist via Apple Music similar-artists view."""
82 storefront = self.provider._storefront
83 response = await self.api.get_data(
84 f"catalog/{storefront}/artists/{prov_artist_id}",
85 views="similar-artists",
86 )
87 data = response.get("data", [])
88 if not data:
89 return []
90 similar = data[0].get("views", {}).get("similar-artists", {}).get("data", [])
91 artists: list[Artist] = []
92 for artist_obj in similar[:limit]:
93 parsed = parse_artist(self.provider, artist_obj)
94 if isinstance(parsed, Artist):
95 artists.append(parsed)
96 return artists
97
98 async def get_station_playlist(self, station_id: str) -> Playlist:
99 """Fetch name and artwork for a radio station and return it as a dynamic Playlist."""
100 try:
101 station_response = await self.api.get_data(
102 f"catalog/{self.provider._storefront}/stations/{station_id}"
103 )
104 station_obj = station_response["data"][0]
105 station_obj["id"] = station_id
106 return parse_station_as_playlist(self.provider, station_obj)
107 except MediaNotFoundError, KeyError, IndexError:
108 return parse_station_as_playlist(self.provider, {"id": station_id})
109
110 async def get_personal_recommendations(self) -> list[RecommendationFolder]:
111 """Fetch personal recommendations grouped into folders by section title."""
112 response = await self.api.get_data(
113 "me/recommendations?include[personal-recommendation]=contents"
114 )
115 seen: set[str] = set()
116 folders: dict[str, RecommendationFolder] = {}
117 # Reset maps so stale entries from previous fetches are not kept.
118 self._station_id_to_name.clear()
119 self._station_name_to_id.clear()
120 for recommendation in response.get("data", []):
121 rec_id = recommendation.get("id", "")
122 title = (
123 recommendation.get("attributes", {}).get("title", {}).get("stringForDisplay", "")
124 )
125 if not rec_id or not title:
126 continue
127 contents = recommendation.get("relationships", {}).get("contents", {})
128 for item in contents.get("data", []):
129 if item.get("type") != "stations":
130 continue
131 station_id = item.get("id")
132 if not station_id or station_id in seen:
133 continue
134 attributes = item.get("attributes", {})
135 if attributes.get("isLive", False):
136 # Live broadcast stations require Widevine DRM; skip them.
137 continue
138 seen.add(station_id)
139 if attributes.get("name"):
140 playlist = parse_station_as_playlist(self.provider, item)
141 else:
142 playlist = await self.provider.get_playlist(station_id)
143 if playlist.name == station_id:
144 continue
145 if playlist.name and playlist.name != station_id:
146 self._station_id_to_name[station_id] = playlist.name
147 self._station_name_to_id[playlist.name] = station_id
148 if title not in folders:
149 folders[title] = RecommendationFolder(
150 item_id=_slugify_title(title),
151 provider=self.provider.instance_id,
152 name=title,
153 )
154 folders[title].items.append(playlist)
155 return list(folders.values())
156
157 async def resolve_station_id(self, stale_id: str) -> str | None:
158 """
159 Return the current station ID for a stale one.
160
161 :param stale_id: The outdated station ID that may have been rotated by Apple.
162 """
163 station_name = self._station_id_to_name.get(stale_id)
164 if not station_name:
165 # Maps may be empty after a process restart; populate from the cached payload first.
166 self._populate_station_maps(await self.provider._recommendation_payload())
167 station_name = self._station_id_to_name.get(stale_id)
168 if not station_name:
169 return None
170 # Apple rotates station ids: refresh through the mixin cache so the fresh payload
171 # (which rebuilds the maps) is also what rows/items serve afterwards.
172 await self.provider._refresh_recommendation_payload()
173 return self._station_name_to_id.get(station_name)
174
175 async def browse_stations(self) -> list[ItemMapping | Playlist]:
176 """Return recommended radio stations from personal recommendations."""
177 return cast(
178 "list[ItemMapping | Playlist]",
179 [
180 item
181 for folder in await self.provider._recommendation_payload()
182 for item in folder.items
183 ],
184 )
185
186 def _populate_station_maps(self, folders: list[RecommendationFolder]) -> None:
187 """
188 Populate the station name maps from payload folders, if they are empty.
189
190 After a process restart the payload may be served from the persistent cache
191 without running get_personal_recommendations, leaving the maps empty; the
192 folder items (stations parsed as playlists) carry the id/name pairs.
193 """
194 if self._station_id_to_name:
195 return
196 for folder in folders:
197 for item in folder.items:
198 if item.name and item.name != item.item_id:
199 self._station_id_to_name[item.item_id] = item.name
200 self._station_name_to_id[item.name] = item.item_id
201