/
/
/
1"""MusicMe provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import json
8import re
9import time
10import urllib.parse
11from collections.abc import Sequence
12from typing import TYPE_CHECKING, Any
13
14import aiohttp
15from aiohttp import ClientConnectionError, ClientResponseError
16from music_assistant_models.enums import (
17 ContentType,
18 ImageType,
19 MediaType,
20 StreamType,
21)
22from music_assistant_models.errors import (
23 LoginFailed,
24 MediaNotFoundError,
25 RateLimited,
26 ResourceTemporarilyUnavailable,
27 SetupFailedError,
28)
29from music_assistant_models.media_items import (
30 Album,
31 Artist,
32 AudioFormat,
33 BrowseFolder,
34 ItemMapping,
35 MediaItemImage,
36 MediaItemType,
37 Playlist,
38 ProviderMapping,
39 Radio,
40 RecommendationFolder,
41 SearchResults,
42 Track,
43 UniqueList,
44)
45from music_assistant_models.streamdetails import StreamDetails
46
47from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER, CONF_PASSWORD, CONF_USERNAME
48from music_assistant.controllers.cache import use_cache
49from music_assistant.helpers.aiohttp_client import create_clientsession
50from music_assistant.helpers.json import json_loads
51from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
52from music_assistant.models.music_provider import MusicProvider
53
54from .constants import (
55 CLIENT_JSON,
56 DATASERVICE_BASE,
57 LOGIN_URL,
58 PARTNER_ID,
59 STREAM_BASE,
60 VALID_ID_RE,
61 WEB_BASE,
62)
63from .helpers import decrypt
64
65if TYPE_CHECKING:
66 from collections.abc import AsyncGenerator
67
68 from music_assistant_models.config_entries import ConfigEntry
69
70
71class MusicMeProvider(MusicProvider):
72 """Provider for the MusicMe streaming service."""
73
74 _user_id: str | None = None
75 http_session: aiohttp.ClientSession
76 throttler: ThrottlerManager
77
78 @property
79 def supported_media_types(self) -> set[MediaType]:
80 """Return the media types this provider can serve."""
81 # full catalogue access via search/browse, but only playlists as library items
82 return {MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST}
83
84 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
85 """Return Config entries to setup this provider."""
86 return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
87
88 async def handle_async_init(self) -> None:
89 """Handle async initialization of the provider."""
90 if not self.get_setup_value(CONF_USERNAME) or not self.get_setup_value(CONF_PASSWORD):
91 msg = "Missing MusicMe email or password"
92 raise SetupFailedError(msg)
93 # Dedicated session with its own cookie jar to support multi-instance
94 # (each instance has its own MusicMe login cookies)
95 self.http_session = create_clientsession(self.mass, cookie_jar=aiohttp.CookieJar())
96 self.throttler = ThrottlerManager(rate_limit=1, period=1)
97 await self._login()
98
99 async def unload(self, is_removed: bool = False) -> None:
100 """Handle unload/close of the provider."""
101 if hasattr(self, "http_session") and not self.http_session.closed:
102 await self.http_session.close()
103
104 # ---- search ----
105
106 @use_cache(3600 * 3)
107 async def search(
108 self, search_query: str, media_types: list[MediaType], limit: int = 10
109 ) -> SearchResults:
110 """
111 Perform search on MusicMe.
112
113 Uses a two-pass strategy: the dataservice API first (fast, popular matches),
114 then a web search fallback for niche queries.
115
116 :param search_query: Search query.
117 :param media_types: A list of media_types to include.
118 :param limit: Number of items to return in the search (per type).
119 """
120 result = SearchResults()
121 encoded_query = urllib.parse.quote_plus(search_query)
122
123 data = await self._api_get(
124 f"/search?query={encoded_query}"
125 f"&resources=artists{{maxResults:{limit}}}"
126 f",albums{{maxResults:{limit}}}"
127 f",tracks{{maxResults:{limit}}}"
128 )
129 if data and "results" in data:
130 res = data["results"]
131 if MediaType.ARTIST in media_types:
132 artists = res.get("artists", [])
133 if isinstance(artists, list):
134 result.artists = [self._parse_artist(a) for a in artists if a.get("id")]
135 if MediaType.ALBUM in media_types:
136 albums = res.get("albums", [])
137 if isinstance(albums, list):
138 result.albums = [self._parse_album(a) for a in albums if a.get("barcode")]
139 if MediaType.TRACK in media_types:
140 tracks = res.get("tracks", [])
141 if isinstance(tracks, list):
142 result.tracks = [
143 self._parse_track(t)
144 for t in tracks
145 if t.get("barcode") and t.get("streamable") == 2
146 ]
147
148 if not result.artists and not result.albums and not result.tracks:
149 self.logger.debug("API search empty, trying web search fallback")
150 web_result = await self._web_search_fallback(search_query, media_types, limit)
151 if web_result:
152 result = web_result
153
154 return result
155
156 # ---- browse ----
157
158 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
159 """
160 Browse MusicMe content.
161
162 :param path: The path to browse.
163 """
164 sub_path = ""
165 if path and "://" in path:
166 sub_path = path.split("://", 1)[1]
167 if not path or not sub_path:
168 return [
169 BrowseFolder(
170 item_id="home",
171 provider=self.instance_id,
172 path=f"{self.instance_id}://home",
173 name="Featured",
174 translation_key="featured",
175 ),
176 BrowseFolder(
177 item_id="news",
178 provider=self.instance_id,
179 path=f"{self.instance_id}://news",
180 name="New releases",
181 translation_key="new_releases",
182 ),
183 BrowseFolder(
184 item_id="tops",
185 provider=self.instance_id,
186 path=f"{self.instance_id}://tops",
187 name="Top artists",
188 translation_key="top_artists",
189 ),
190 BrowseFolder(
191 item_id="radios",
192 provider=self.instance_id,
193 path=f"{self.instance_id}://radios",
194 name="Themed radios",
195 translation_key="themed_radios",
196 ),
197 ]
198
199 parts = [p for p in sub_path.split("/") if p]
200 section = parts[0] if parts else ""
201 sub_id = parts[1] if len(parts) > 1 else None
202
203 if section == "home":
204 return await self._browse_home()
205 if section == "news":
206 return await self._browse_news(sub_id)
207 if section == "tops":
208 return await self._browse_tops()
209 if section == "radios":
210 return await self._browse_radio_themes(sub_id)
211 return []
212
213 # ---- recommendations ----
214
215 async def get_recommendations(self) -> list[RecommendationFolder]:
216 """Get this provider's available recommendation rows, without items."""
217 return [
218 RecommendationFolder(
219 name="Featured",
220 translation_key="featured",
221 item_id=f"{self.instance_id}_home",
222 provider=self.instance_id,
223 icon="mdi-star",
224 ),
225 RecommendationFolder(
226 name="New releases",
227 translation_key="new_releases",
228 item_id=f"{self.instance_id}_news",
229 provider=self.instance_id,
230 icon="mdi-new-box",
231 ),
232 RecommendationFolder(
233 name="Top artists",
234 translation_key="top_artists",
235 item_id=f"{self.instance_id}_tops",
236 provider=self.instance_id,
237 icon="mdi-trending-up",
238 ),
239 RecommendationFolder(
240 name="Radios",
241 translation_key="radios",
242 item_id=f"{self.instance_id}_radios",
243 provider=self.instance_id,
244 icon="mdi-radio",
245 ),
246 ]
247
248 async def get_recommendation_items(
249 self, item_id: str
250 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
251 """
252 Get the items for a single recommendation row.
253
254 :param item_id: The item_id of the row, as returned by get_recommendations.
255 """
256 folder: RecommendationFolder | None = None
257 if item_id == f"{self.instance_id}_home":
258 folder = await self._get_home_recommendations()
259 elif item_id == f"{self.instance_id}_news":
260 folder = await self._get_news_recommendations()
261 elif item_id == f"{self.instance_id}_tops":
262 folder = await self._get_tops_recommendations()
263 elif item_id == f"{self.instance_id}_radios":
264 folder = await self._get_radios_recommendations()
265 if folder is None:
266 return UniqueList()
267 return folder.items
268
269 @use_cache(3600)
270 async def _browse_home(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
271 """Browse the MusicMe homepage."""
272 data = await self._api_get("/home")
273 if not data:
274 return []
275 items: list[MediaItemType | ItemMapping | BrowseFolder] = []
276 for entry in data.get("results", {}).get("items", []):
277 if entry.get("id"):
278 items.append(self._parse_radio(entry))
279 return items
280
281 @use_cache(3600)
282 async def _browse_news(
283 self, style_id: str | None = None
284 ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
285 """Browse new releases, optionally filtered by style."""
286 sid = style_id or "0"
287 data = await self._api_get(
288 f"/news/{sid}?filters={{style:{sid}}}"
289 f"&resources=albums{{maxResults:20}},focus-albums,styles"
290 )
291 if not data:
292 return []
293 results: list[MediaItemType | ItemMapping | BrowseFolder] = []
294
295 if not style_id:
296 for style in data.get("results", {}).get("styles", []):
297 s_id = str(style.get("id", ""))
298 s_name = style.get("name", "")
299 if s_id and s_name and s_id != "0":
300 results.append(
301 BrowseFolder(
302 item_id=f"news_{s_id}",
303 provider=self.instance_id,
304 path=f"{self.instance_id}://news/{s_id}",
305 name=s_name,
306 )
307 )
308
309 for album_obj in data.get("results", {}).get("albums", []):
310 if album_obj.get("barcode") and album_obj.get("streamable", 0) == 2:
311 results.append(self._parse_album(album_obj))
312 return results
313
314 @use_cache(3600)
315 async def _browse_tops(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
316 """Browse top artists."""
317 data = await self._api_get(
318 "/tops?filters={style:0}&resources=styles,artists{maxResults:20},videos{maxResults:0}"
319 )
320 if not data:
321 return []
322 results: list[MediaItemType | ItemMapping | BrowseFolder] = []
323 for artist_obj in data.get("results", {}).get("artists", []):
324 if artist_obj.get("id"):
325 results.append(self._parse_artist(artist_obj))
326 return results
327
328 @use_cache(3600)
329 async def _browse_radio_themes(
330 self, theme_id: str | None = None
331 ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
332 """Browse radios grouped by theme."""
333 tid = theme_id or "0"
334 data = await self._api_get(
335 f"/radios?filters={{theme:{tid}}}&resources=themes,theme-airplays{{maxResults:50}},home"
336 )
337 if not data:
338 return []
339 results: list[MediaItemType | ItemMapping | BrowseFolder] = []
340
341 if not theme_id:
342 for theme in data.get("results", {}).get("themes", []):
343 t_id = str(theme.get("id", ""))
344 t_name = theme.get("name", "")
345 if t_id and t_name and t_id != "0":
346 results.append(
347 BrowseFolder(
348 item_id=f"radios_{t_id}",
349 provider=self.instance_id,
350 path=f"{self.instance_id}://radios/{t_id}",
351 name=f"{t_name} ({theme.get('radiocount', '')})",
352 )
353 )
354
355 for item in data.get("results", {}).get("theme-airplays", []):
356 if item.get("id"):
357 results.append(self._parse_radio(item))
358 return results
359
360 # ---- library ----
361
362 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
363 """Retrieve library playlists from MusicMe."""
364 data = await self._api_get('/playlists?resources=home{details:"list"}')
365 if not data:
366 return
367 items: Any = data.get("results", {})
368 if isinstance(items, dict) and "items" in items:
369 items = items["items"]
370 if isinstance(items, dict) and "results" in items:
371 items = items["results"]
372 if isinstance(items, list):
373 for pl in items:
374 if pl.get("id"):
375 yield self._parse_playlist(pl)
376
377 # ---- item detail getters ----
378
379 @use_cache(3600 * 24 * 30)
380 async def get_artist(self, prov_artist_id: str) -> Artist:
381 """Get full artist details by id."""
382 data = await self._api_get(
383 f"/artist/{prov_artist_id}"
384 f"?resources=albums{{maxResults:0}},related-artists{{maxResults:0}}"
385 )
386 if not data or "item" not in data:
387 msg = f"Artist {prov_artist_id} not found"
388 raise MediaNotFoundError(msg)
389 return self._parse_artist(data["item"])
390
391 @use_cache(3600 * 24 * 30)
392 async def get_album(self, prov_album_id: str) -> Album:
393 """Get full album details by id."""
394 data = await self._api_get(f"/album/{prov_album_id}?resources=tracks")
395 if not data or "item" not in data:
396 msg = f"Album {prov_album_id} not found"
397 raise MediaNotFoundError(msg)
398 return self._parse_album(data["item"])
399
400 @use_cache(3600 * 24 * 30)
401 async def get_track(self, prov_track_id: str) -> Track:
402 """
403 Get full track details by id.
404
405 MusicMe doesn't have a single-track endpoint, so we fetch the parent album.
406 """
407 album_barcode = prov_track_id.split("-", maxsplit=1)[0]
408 data = await self._api_get(f"/album/{album_barcode}?resources=tracks")
409 if not data:
410 msg = f"Track {prov_track_id} not found"
411 raise MediaNotFoundError(msg)
412 for track_obj in data.get("results", {}).get("tracks", []):
413 if track_obj.get("barcode") == prov_track_id:
414 return self._parse_track(track_obj)
415 msg = f"Track {prov_track_id} not found in album {album_barcode}"
416 raise MediaNotFoundError(msg)
417
418 @use_cache(3600 * 24 * 30)
419 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
420 """Get full playlist details by id."""
421 data = await self._api_get(f"/playlist/{prov_playlist_id}?resources=tracks")
422 if not data:
423 msg = f"Playlist {prov_playlist_id} not found"
424 raise MediaNotFoundError(msg)
425 item = data.get("item", data)
426 return self._parse_playlist(item)
427
428 @use_cache(3600 * 24 * 14, allow_expired_cache=True)
429 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
430 """Get all album tracks for given album id."""
431 data = await self._api_get(f"/album/{prov_album_id}?resources=tracks")
432 if not data:
433 return []
434 tracks = data.get("results", {}).get("tracks", [])
435 return [self._parse_track(t) for t in tracks if t.get("barcode")]
436
437 @use_cache(3600 * 24 * 7, allow_expired_cache=True)
438 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
439 """
440 Get a list of top/popular tracks for the given artist.
441
442 MusicMe has no dedicated top-tracks endpoint, so we collect the first
443 tracks from the artist's most recent albums (sorted by date desc).
444 """
445 data = await self._api_get(f"/artist/{prov_artist_id}?resources=albums{{maxResults:5}}")
446 if not data:
447 return []
448 albums = data.get("results", {}).get("albums", [])
449 streamable_barcodes = [
450 a["barcode"] for a in albums if a.get("barcode") and a.get("streamable", 0) == 2
451 ]
452 album_results = await asyncio.gather(
453 *(self._api_get(f"/album/{bc}?resources=tracks") for bc in streamable_barcodes)
454 )
455 top_tracks: list[Track] = []
456 for album_data in album_results:
457 if not album_data:
458 continue
459 for t in album_data.get("results", {}).get("tracks", []):
460 if t.get("barcode") and t.get("streamable", 0) == 2:
461 top_tracks.append(self._parse_track(t))
462 if len(top_tracks) >= 20:
463 break
464 if len(top_tracks) >= 20:
465 break
466 return top_tracks
467
468 @use_cache(3600 * 24 * 14, allow_expired_cache=True)
469 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
470 """Get a list of albums for the given artist."""
471 data = await self._api_get(f"/artist/{prov_artist_id}?resources=albums{{maxResults:50}}")
472 if not data:
473 return []
474 albums = data.get("results", {}).get("albums", [])
475 return [self._parse_album(a) for a in albums if a.get("barcode")]
476
477 @use_cache(3600 * 3, allow_expired_cache=True)
478 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
479 """Get playlist tracks."""
480 if page > 0:
481 return []
482 data = await self._api_get(f"/playlist/{prov_playlist_id}?resources=tracks")
483 if not data:
484 return []
485 tracks = data.get("results", {}).get("tracks", [])
486 return [self._parse_track(t) for t in tracks if t.get("barcode")]
487
488 # ---- streaming ----
489
490 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
491 """
492 Return the content details for the given track when it will be streamed.
493
494 :param item_id: The MusicMe track barcode, or radio airplay ID.
495 :param media_type: The media type (TRACK or RADIO).
496 """
497 if media_type == MediaType.RADIO:
498 track_barcode = await self._get_radio_track(item_id)
499 else:
500 track_barcode = item_id
501
502 stream_url = await self._get_stream_url(track_barcode)
503 return StreamDetails(
504 item_id=item_id,
505 provider=self.instance_id,
506 audio_format=AudioFormat(
507 content_type=ContentType.MP4,
508 codec_type=ContentType.AAC,
509 sample_rate=44100,
510 bit_depth=16,
511 channels=2,
512 ),
513 stream_type=StreamType.HTTP,
514 path=stream_url,
515 can_seek=True,
516 allow_seek=True,
517 )
518
519 async def _get_radio_track(self, radio_id: str) -> str:
520 """Get a streamable track barcode from a MusicMe radio/airplay."""
521 data = await self._api_get(f"/airplay/{radio_id}?resources=tracks")
522 if not data:
523 msg = f"Radio {radio_id} returned no data"
524 raise MediaNotFoundError(msg)
525 tracks = data.get("results", {}).get("tracks", [])
526 for t in tracks:
527 if t.get("barcode") and t.get("streamable", 0) == 2:
528 return str(t["barcode"])
529 msg = f"No streamable tracks in radio {radio_id}"
530 raise MediaNotFoundError(msg)
531
532 # ---- parsers ----
533
534 def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
535 """Parse a MusicMe artist object to a Music Assistant Artist."""
536 artist_id = str(artist_obj.get("id", ""))
537 artist = Artist(
538 item_id=artist_id,
539 provider=self.instance_id,
540 name=artist_obj.get("name", "Unknown"),
541 provider_mappings={
542 ProviderMapping(
543 item_id=artist_id,
544 provider_domain=self.domain,
545 provider_instance=self.instance_id,
546 url=f"{WEB_BASE}{artist_obj['uri']}" if artist_obj.get("uri") else "",
547 )
548 },
549 )
550 # Use square artist images (r288) instead of the banner-shaped eyestoeye/144x56
551 if artist_id:
552 artist.metadata.add_image(
553 MediaItemImage(
554 type=ImageType.THUMB,
555 path=f"https://covers-ng3.hosting-media.net/art/r288/{artist_id}.jpg",
556 provider=self.instance_id,
557 remotely_accessible=True,
558 )
559 )
560 return artist
561
562 def _parse_album(self, album_obj: dict[str, Any]) -> Album:
563 """Parse a MusicMe album object to a Music Assistant Album."""
564 barcode = str(album_obj.get("barcode", ""))
565 album = Album(
566 item_id=barcode,
567 provider=self.instance_id,
568 name=album_obj.get("name", album_obj.get("title", "Unknown")),
569 provider_mappings={
570 ProviderMapping(
571 item_id=barcode,
572 provider_domain=self.domain,
573 provider_instance=self.instance_id,
574 available=album_obj.get("streamable", 0) == 2,
575 audio_format=AudioFormat(
576 content_type=ContentType.MP4,
577 codec_type=ContentType.AAC,
578 sample_rate=44100,
579 bit_depth=16,
580 ),
581 url=f"{WEB_BASE}{album_obj['uri']}" if album_obj.get("uri") else "",
582 )
583 },
584 )
585 for art_obj in album_obj.get("artists", []):
586 if art_obj.get("id"):
587 album.artists.append(self._parse_artist(art_obj))
588 if cover := album_obj.get("cover_url"):
589 album.metadata.add_image(
590 MediaItemImage(
591 type=ImageType.THUMB,
592 path=cover,
593 provider=self.instance_id,
594 remotely_accessible=True,
595 )
596 )
597 if streetdate := album_obj.get("streetdate"):
598 with contextlib.suppress(ValueError, IndexError):
599 album.year = int(streetdate[:4])
600 return album
601
602 def _parse_track(self, track_obj: dict[str, Any]) -> Track:
603 """Parse a MusicMe track object to a Music Assistant Track."""
604 barcode = str(track_obj.get("barcode", ""))
605 track = Track(
606 item_id=barcode,
607 provider=self.instance_id,
608 name=track_obj.get("title", "Unknown"),
609 duration=track_obj.get("duration", 0),
610 provider_mappings={
611 ProviderMapping(
612 item_id=barcode,
613 provider_domain=self.domain,
614 provider_instance=self.instance_id,
615 available=track_obj.get("streamable", 0) == 2,
616 audio_format=AudioFormat(
617 content_type=ContentType.MP4,
618 codec_type=ContentType.AAC,
619 sample_rate=44100,
620 bit_depth=16,
621 ),
622 )
623 },
624 )
625 # Parse disc and track number from barcode format: {albumBarcode}-{disc}_{track}
626 if "-" in barcode and "_" in barcode:
627 try:
628 disc_track = barcode.split("-", 1)[1]
629 disc_track_parts = disc_track.split("_")
630 track.disc_number = int(disc_track_parts[0])
631 track.track_number = int(disc_track_parts[1])
632 except ValueError, IndexError:
633 pass
634 for art_obj in track_obj.get("artists", []):
635 if art_obj.get("id"):
636 track.artists.append(self._parse_artist(art_obj))
637 if track_obj.get("album"):
638 album_barcode = barcode.split("-", maxsplit=1)[0]
639 track.album = ItemMapping(
640 item_id=album_barcode,
641 provider=self.instance_id,
642 name=track_obj["album"],
643 media_type=MediaType.ALBUM,
644 )
645 if cover := track_obj.get("cover_url"):
646 track.metadata.add_image(
647 MediaItemImage(
648 type=ImageType.THUMB,
649 path=cover,
650 provider=self.instance_id,
651 remotely_accessible=True,
652 )
653 )
654 return track
655
656 def _parse_radio(self, radio_obj: dict[str, Any]) -> Radio:
657 """Parse a MusicMe radio/airplay object to a Music Assistant Radio."""
658 radio_id = str(radio_obj.get("id", ""))
659 radio = Radio(
660 item_id=radio_id,
661 provider=self.instance_id,
662 name=radio_obj.get("name", "Radio"),
663 provider_mappings={
664 ProviderMapping(
665 item_id=radio_id,
666 provider_domain=self.domain,
667 provider_instance=self.instance_id,
668 )
669 },
670 )
671 img = radio_obj.get("tile_url") or radio_obj.get("img_url")
672 if img:
673 radio.metadata.add_image(
674 MediaItemImage(
675 type=ImageType.THUMB,
676 path=img,
677 provider=self.instance_id,
678 remotely_accessible=True,
679 )
680 )
681 preview_artists = radio_obj.get("preview_artists", [])
682 if preview_artists:
683 names = [a.get("name", "") for a in preview_artists[:8] if a.get("name")]
684 if names:
685 radio.metadata.description = ", ".join(names) + "..."
686 return radio
687
688 def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
689 """Parse a MusicMe playlist object to a Music Assistant Playlist."""
690 playlist_id = str(playlist_obj.get("id", ""))
691 playlist = Playlist(
692 item_id=playlist_id,
693 provider=self.instance_id,
694 name=playlist_obj.get("name", playlist_obj.get("title", "Playlist")),
695 provider_mappings={
696 ProviderMapping(
697 item_id=playlist_id,
698 provider_domain=self.domain,
699 provider_instance=self.instance_id,
700 )
701 },
702 )
703 if cover := playlist_obj.get("cover_url"):
704 playlist.metadata.add_image(
705 MediaItemImage(
706 type=ImageType.THUMB,
707 path=cover,
708 provider=self.instance_id,
709 remotely_accessible=True,
710 )
711 )
712 return playlist
713
714 # ---- private helpers ----
715
716 async def _web_search_fallback(
717 self,
718 search_query: str,
719 media_types: list[MediaType],
720 limit: int = 10,
721 ) -> SearchResults | None:
722 """
723 Fall back to MusicMe's PHP web search when the dataservice API returns nothing.
724
725 :param search_query: Search query.
726 :param media_types: A list of media_types to include.
727 :param limit: Number of items to return.
728 """
729 encoded = urllib.parse.quote_plus(search_query)
730 url = f"{WEB_BASE}/search.php?ambsearch={encoded}&mmz=all"
731
732 try:
733 async with self.http_session.get(url, allow_redirects=True) as resp:
734 if resp.status != 200:
735 return None
736 raw_bytes = await resp.read()
737 content = raw_bytes.decode("latin-1", errors="replace")
738 except (ClientConnectionError, ClientResponseError) as err:
739 self.logger.debug("Web search fallback failed: %s", err)
740 return None
741
742 result = SearchResults()
743
744 match = re.search(r"window\.playerInit\s*=\s*(\{.*?\});", content, re.DOTALL)
745 if not match:
746 return None
747
748 player_init = json_loads(match.group(1))
749 content_on_load = player_init.get("contentOnLoad", {})
750 content_type = content_on_load.get("type", "")
751 content_id = content_on_load.get("id")
752
753 if not content_id or not VALID_ID_RE.match(str(content_id)):
754 return None
755
756 if content_type == "artist" and MediaType.ARTIST in media_types:
757 artist_data = await self._api_get(
758 f"/artist/{content_id}"
759 f"?resources=albums{{maxResults:{limit}}},related-artists{{maxResults:0}}"
760 )
761 if artist_data and "item" in artist_data:
762 result.artists = [self._parse_artist(artist_data["item"])]
763 if MediaType.ALBUM in media_types:
764 albums = artist_data.get("results", {}).get("albums", [])
765 result.albums = [self._parse_album(a) for a in albums if a.get("barcode")]
766
767 elif content_type == "album" and MediaType.ALBUM in media_types:
768 album_data = await self._api_get(f"/album/{content_id}?resources=tracks")
769 if album_data and "item" in album_data:
770 result.albums = [self._parse_album(album_data["item"])]
771 if MediaType.TRACK in media_types:
772 tracks = album_data.get("results", {}).get("tracks", [])
773 result.tracks = [
774 self._parse_track(t)
775 for t in tracks[:limit]
776 if t.get("barcode") and t.get("streamable") == 2
777 ]
778
779 return result if (result.artists or result.albums or result.tracks) else None
780
781 async def _login(self) -> None:
782 """Authenticate with MusicMe via web login and extract the userId."""
783 email = self.get_setup_value(CONF_USERNAME)
784 password = self.get_setup_value(CONF_PASSWORD)
785
786 try:
787 async with self.http_session.post(
788 LOGIN_URL,
789 data={
790 "email": email,
791 "password": password,
792 "act": "login",
793 "login_cookie": "1",
794 },
795 allow_redirects=True,
796 ) as resp:
797 resp.raise_for_status()
798 except (ClientResponseError, ClientConnectionError) as err:
799 msg = f"MusicMe login request failed: {err}"
800 raise LoginFailed(msg) from err
801
802 try:
803 async with self.http_session.get(f"{WEB_BASE}/?f=1") as resp:
804 resp.raise_for_status()
805 raw_bytes = await resp.read()
806 content = raw_bytes.decode("latin-1", errors="replace")
807 except (ClientResponseError, ClientConnectionError) as err:
808 msg = f"MusicMe page load failed: {err}"
809 raise LoginFailed(msg) from err
810
811 match = re.search(r"window\.playerInit\s*=\s*(\{.*?\});", content, re.DOTALL)
812 if not match:
813 msg = "Login failed â could not find playerInit in MusicMe response"
814 raise LoginFailed(msg)
815
816 player_init = json_loads(match.group(1))
817 self._user_id = player_init.get("userId")
818 if not self._user_id:
819 msg = "Login failed â no userId in MusicMe playerInit"
820 raise LoginFailed(msg)
821
822 self.logger.info(
823 "Successfully logged in to MusicMe (catalog=%s)",
824 player_init.get("catalogSize", "?"),
825 )
826
827 def _build_base_params(self) -> str:
828 """Build the common query string for dataservice API calls."""
829 parts = ["format=json", f"partnerid={PARTNER_ID}", f"client={CLIENT_JSON}"]
830 if self._user_id:
831 parts.append(f"userid={self._user_id}")
832 return "&".join(parts)
833
834 @throttle_with_retries
835 async def _api_get(self, endpoint: str) -> dict[str, Any] | None:
836 """
837 Make a GET request to MusicMe dataservice and decrypt the response.
838
839 :param endpoint: API endpoint path (e.g. '/search?query=...')
840 """
841 url = f"{DATASERVICE_BASE}{endpoint}"
842 sep = "&" if "?" in url else "?"
843 url = f"{url}{sep}{self._build_base_params()}"
844
845 self.logger.debug("GET %s", endpoint.split("?", maxsplit=1)[0])
846 try:
847 async with self.http_session.get(url) as response:
848 if response.status == 429:
849 try:
850 backoff = min(int(response.headers.get("Retry-After", 10)), 300)
851 except ValueError, TypeError:
852 backoff = 10
853 raise RateLimited("Rate limited", backoff_time=backoff)
854 if response.status in (502, 503):
855 raise ResourceTemporarilyUnavailable(
856 "Server temporarily unavailable", backoff_time=30
857 )
858 if response.status == 404:
859 return None
860 response.raise_for_status()
861 raw = await response.text()
862 except ClientConnectionError as err:
863 path = endpoint.split("?", maxsplit=1)[0]
864 raise ResourceTemporarilyUnavailable(
865 f"Connection error for {path}", backoff_time=10
866 ) from err
867
868 try:
869 result: dict[str, Any] = json_loads(decrypt(raw.strip()))
870 return result
871 except (ValueError, json.JSONDecodeError, IndexError, UnicodeDecodeError) as err:
872 self.logger.warning("Failed to decrypt API response: %s", err)
873 return None
874
875 async def _get_stream_url(self, track_barcode: str) -> str:
876 """
877 Get a fresh streaming URL for a track.
878
879 Tickets are single-use and ephemeral â must be generated right before playback.
880 """
881 nocache = int(time.time() * 1000)
882 data = await self._api_get(f"/getstream/{PARTNER_ID}?ref={track_barcode}&nocache={nocache}")
883 if not data or "ticket" not in data:
884 msg = f"No streaming ticket for {track_barcode}"
885 raise MediaNotFoundError(msg)
886
887 # Double-encrypted: API response already decrypted, ticket needs a second pass
888 try:
889 ticket = decrypt(data["ticket"])
890 except (ValueError, TypeError, IndexError, UnicodeDecodeError) as err:
891 msg = f"Unable to decode streaming ticket for {track_barcode}"
892 raise MediaNotFoundError(msg) from err
893 return f"{STREAM_BASE}/{PARTNER_ID}/{ticket}.mp4"
894
895 @use_cache(3600 * 3, base_class=RecommendationFolder)
896 async def _get_home_recommendations(self) -> RecommendationFolder | None:
897 """Get the Featured recommendation folder, or None when unavailable."""
898 data = await self._api_get("/home")
899 if not data:
900 return None
901 items: list[MediaItemType | ItemMapping | BrowseFolder] = [
902 self._parse_radio(entry)
903 for entry in data.get("results", {}).get("items", [])
904 if entry.get("id")
905 ]
906 if not items:
907 return None
908 return RecommendationFolder(
909 name="Featured",
910 translation_key="featured",
911 item_id=f"{self.instance_id}_home",
912 provider=self.instance_id,
913 icon="mdi-star",
914 items=UniqueList(items),
915 )
916
917 @use_cache(3600 * 3, base_class=RecommendationFolder)
918 async def _get_news_recommendations(self) -> RecommendationFolder | None:
919 """Get the New releases recommendation folder, or None when unavailable."""
920 data = await self._api_get(
921 "/news/0?filters={style:0}&resources=albums{maxResults:10},focus-albums,styles"
922 )
923 if not data:
924 return None
925 items: list[MediaItemType | ItemMapping | BrowseFolder] = [
926 self._parse_album(album_obj)
927 for album_obj in data.get("results", {}).get("albums", [])
928 if album_obj.get("barcode") and album_obj.get("streamable", 0) == 2
929 ]
930 if not items:
931 return None
932 return RecommendationFolder(
933 name="New releases",
934 translation_key="new_releases",
935 item_id=f"{self.instance_id}_news",
936 provider=self.instance_id,
937 icon="mdi-new-box",
938 items=UniqueList(items),
939 )
940
941 @use_cache(3600 * 3, base_class=RecommendationFolder)
942 async def _get_tops_recommendations(self) -> RecommendationFolder | None:
943 """Get the Top artists recommendation folder, or None when unavailable."""
944 data = await self._api_get(
945 "/tops?filters={style:0}&resources=styles,artists{maxResults:10},videos{maxResults:0}"
946 )
947 if not data:
948 return None
949 items: list[MediaItemType | ItemMapping | BrowseFolder] = [
950 self._parse_artist(artist_obj)
951 for artist_obj in data.get("results", {}).get("artists", [])
952 if artist_obj.get("id")
953 ]
954 if not items:
955 return None
956 return RecommendationFolder(
957 name="Top artists",
958 translation_key="top_artists",
959 item_id=f"{self.instance_id}_tops",
960 provider=self.instance_id,
961 icon="mdi-trending-up",
962 items=UniqueList(items),
963 )
964
965 @use_cache(3600 * 3, base_class=RecommendationFolder)
966 async def _get_radios_recommendations(self) -> RecommendationFolder | None:
967 """Get the Radios recommendation folder, or None when unavailable."""
968 data = await self._api_get(
969 "/radios?filters={theme:0}&resources=themes,theme-airplays{maxResults:10},home"
970 )
971 if not data:
972 return None
973 items: list[MediaItemType | ItemMapping | BrowseFolder] = [
974 self._parse_radio(item)
975 for item in data.get("results", {}).get("theme-airplays", [])[:10]
976 if item.get("id")
977 ]
978 if not items:
979 return None
980 return RecommendationFolder(
981 name="Radios",
982 translation_key="radios",
983 item_id=f"{self.instance_id}_radios",
984 provider=self.instance_id,
985 icon="mdi-radio",
986 items=UniqueList(items),
987 )
988