music-assistant-server

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