Improvement of MusicMe music provider for API signin, and fix the parsing of the "streamable" fields and playlists ids

music_assistant/providers/musicme/__init__.py

@@ -9,7 +9,7 @@
 
 from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
 
-from .constants import SUPPORTED_FEATURES
+from .constants import SIGNIN_BY_API, SUPPORTED_FEATURES
 from .provider import MusicMeProvider
 
 if TYPE_CHECKING:
@@ -42,6 +42,14 @@ async def get_config_entries(
     """
     # ruff: noqa: ARG001
     return (
+        ConfigEntry(
+            key=SIGNIN_BY_API,
+            type=ConfigEntryType.BOOLEAN,
+            label="Signin by API",
+            required=False,
+            default_value=False,
+            description="If checked, the signin will be made by API call, if unchecked, login will be made by HTTP authentication.",
+        ),
         ConfigEntry(
             key=CONF_USERNAME,
             type=ConfigEntryType.STRING,

music_assistant/providers/musicme/constants.py

@@ -14,6 +14,7 @@
     ProviderFeature.RECOMMENDATIONS,
 }
 
+SIGNIN_BY_API = "signin_by_api" # Custom key for ConfigEntry
 DATASERVICE_BASE = "https://dataservice.musicme.com/dataservice/v3"
 STREAM_BASE = "https://stream.hosting-media.net/musicme"
 WEB_BASE = "https://www.musicme.com"

music_assistant/providers/musicme/provider.py

@@ -55,6 +55,7 @@
     DATASERVICE_BASE,
     LOGIN_URL,
     PARTNER_ID,
+    SIGNIN_BY_API,
     STREAM_BASE,
     VALID_ID_RE,
     WEB_BASE,
@@ -81,7 +82,12 @@ async def handle_async_init(self) -> None:
         # (each instance has its own MusicMe login cookies)
         self.http_session = create_clientsession(self.mass, cookie_jar=aiohttp.CookieJar())
         self.throttler = ThrottlerManager(rate_limit=1, period=1)
-        await self._login()
+        if self.config.get_value(SIGNIN_BY_API):
+            self.logger.info("Trying to signin by API call")
+            await self._signin_by_api()
+        else:
+            self.logger.info("Trying to login by HTTP authentication")
+            await self._login()
 
     async def unload(self, is_removed: bool = False) -> None:
         """Handle unload/close of the provider."""
@@ -128,7 +134,7 @@ async def search(
                     result.tracks = [
                         self._parse_track(t)
                         for t in tracks
-                        if t.get("barcode") and t.get("streamable") == 2
+                        if t.get("barcode") and t.get("streamable") != 0
                     ]
 
         if not result.artists and not result.albums and not result.tracks:
@@ -232,7 +238,7 @@ async def _browse_news(
                     )
 
         for album_obj in data.get("results", {}).get("albums", []):
-            if album_obj.get("barcode") and album_obj.get("streamable", 0) == 2:
+            if album_obj.get("barcode") and album_obj.get("streamable", 0) != 0:
                 results.append(self._parse_album(album_obj))
         return results
 
@@ -324,7 +330,7 @@ async def recommendations(self) -> list[RecommendationFolder]:
                 icon="mdi-new-box",
             )
             for album_obj in news_data.get("results", {}).get("albums", []):
-                if album_obj.get("barcode") and album_obj.get("streamable", 0) == 2:
+                if album_obj.get("barcode") and album_obj.get("streamable", 0) != 0:
                     folder.items.append(self._parse_album(album_obj))
             if folder.items:
                 result.append(folder)
@@ -445,7 +451,7 @@ async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
             return []
         albums = data.get("results", {}).get("albums", [])
         streamable_barcodes = [
-            a["barcode"] for a in albums if a.get("barcode") and a.get("streamable", 0) == 2
+            a["barcode"] for a in albums if a.get("barcode") and a.get("streamable", 0) != 0
         ]
         album_results = await asyncio.gather(
             *(self._api_get(f"/album/{bc}?resources=tracks") for bc in streamable_barcodes)
@@ -455,7 +461,7 @@ async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
             if not album_data:
                 continue
             for t in album_data.get("results", {}).get("tracks", []):
-                if t.get("barcode") and t.get("streamable", 0) == 2:
+                if t.get("barcode") and t.get("streamable", 0) != 0:
                     top_tracks.append(self._parse_track(t))
                 if len(top_tracks) >= 20:
                     break
@@ -521,7 +527,7 @@ async def _get_radio_track(self, radio_id: str) -> str:
             raise MediaNotFoundError(msg)
         tracks = data.get("results", {}).get("tracks", [])
         for t in tracks:
-            if t.get("barcode") and t.get("streamable", 0) == 2:
+            if t.get("barcode") and t.get("streamable", 0) != 0:
                 return str(t["barcode"])
         msg = f"No streamable tracks in radio {radio_id}"
         raise MediaNotFoundError(msg)
@@ -568,7 +574,7 @@ def _parse_album(self, album_obj: dict[str, Any]) -> Album:
                     item_id=barcode,
                     provider_domain=self.domain,
                     provider_instance=self.instance_id,
-                    available=album_obj.get("streamable", 0) == 2,
+                    available=album_obj.get("streamable", 0) != 0,
                     audio_format=AudioFormat(
                         content_type=ContentType.MP4,
                         codec_type=ContentType.AAC,
@@ -609,7 +615,7 @@ def _parse_track(self, track_obj: dict[str, Any]) -> Track:
                     item_id=barcode,
                     provider_domain=self.domain,
                     provider_instance=self.instance_id,
-                    available=track_obj.get("streamable", 0) == 2,
+                    available=track_obj.get("streamable", 0) != 0,
                     audio_format=AudioFormat(
                         content_type=ContentType.MP4,
                         codec_type=ContentType.AAC,
@@ -684,7 +690,7 @@ def _parse_radio(self, radio_obj: dict[str, Any]) -> Radio:
 
     def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
         """Parse a MusicMe playlist object to a Music Assistant Playlist."""
-        playlist_id = str(playlist_obj.get("id", ""))
+        playlist_id = (str(playlist_obj.get("id", ""))).removeprefix("pl-")
         playlist = Playlist(
             item_id=playlist_id,
             provider=self.instance_id,
@@ -769,11 +775,32 @@ async def _web_search_fallback(
                     result.tracks = [
                         self._parse_track(t)
                         for t in tracks[:limit]
-                        if t.get("barcode") and t.get("streamable") == 2
+                        if t.get("barcode") and t.get("streamable") != 0
                     ]
 
         return result if (result.artists or result.albums or result.tracks) else None
 
+    async def _signin_by_api(self) -> None:
+        login = urllib.parse.quote_plus(str(self.config.get_value(CONF_USERNAME)))
+        password = urllib.parse.quote_plus(str(self.config.get_value(CONF_PASSWORD)))
+
+        response = await self._api_get(
+            f"/medialibrary/signin"
+            f"?channel=65777&lang=fr&client=%7B%22type%22%3A%22desktop-web%22%2C%22context%22%3A%22pro.bib.musicme.com%22%7D"
+            f"&key=sKTBA7ybW3nvCUQ6&nocrypt=0&login={login}&password={password}"
+        )
+
+        if response and "results" in response:
+            results = response.get("results", {})
+            if results and "user" in results:
+                self._user_id = results.get("user").get("id")
+
+        if not self._user_id:
+            msg = "Login failed — no user.id in MusicMe API response"
+            raise LoginFailed(msg)
+
+        self.logger.info("Successfully logged in to MusicMe")
+
     async def _login(self) -> None:
         """Authenticate with MusicMe via web login and extract the userId."""
         email = self.config.get_value(CONF_USERNAME)

tests/providers/musicme/test_parsers.py

@@ -152,8 +152,8 @@ def test_album_streamable(self, provider: MusicMeProvider) -> None:
         assert mapping.available is True
 
     def test_album_not_streamable(self, provider: MusicMeProvider) -> None:
-        """Test album with streamable != 2 is marked unavailable."""
-        obj = {**ALBUM_OBJ, "streamable": 1}
+        """Test album with streamable != 0 is marked unavailable."""
+        obj = {**ALBUM_OBJ, "streamable": 0}
         album = provider._parse_album(obj)
         mapping = next(iter(album.provider_mappings))
         assert mapping.available is False

tests/providers/musicme/test_streaming.py

@@ -72,7 +72,7 @@ async def test_returns_first_streamable_track(self, provider: MusicMeProvider) -
             return_value={
                 "results": {
                     "tracks": [
-                        {"barcode": "AAA-01_01", "streamable": 1},
+                        {"barcode": "AAA-01_01", "streamable": 0},
                         {"barcode": "BBB-01_01", "streamable": 2},
                         {"barcode": "CCC-01_01", "streamable": 2},
                     ]