music-assistant-server

1"""NetEase Cloud Music provider implementation (MVP)."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import re
8import time
9from collections.abc import AsyncGenerator, Sequence
10from contextlib import suppress
11from datetime import UTC, datetime
12from typing import TYPE_CHECKING, Any
13from urllib.parse import parse_qs, urlparse
14
15from aiohttp import ClientError, ClientSession, ClientTimeout
16from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
17from music_assistant_models.enums import (
18    ConfigEntryType,
19    ContentType,
20    ImageType,
21    MediaType,
22    ProviderFeature,
23    StreamType,
24)
25from music_assistant_models.errors import (
26    InvalidDataError,
27    LoginFailed,
28    MediaNotFoundError,
29    ResourceTemporarilyUnavailable,
30    UnplayableMediaError,
31    UnsupportedFeaturedException,
32)
33from music_assistant_models.media_items import (
34    Album,
35    Artist,
36    AudioFormat,
37    ItemMapping,
38    MediaItemImage,
39    Playlist,
40    ProviderMapping,
41    RecommendationFolder,
42    SearchResults,
43    Track,
44    UniqueList,
45)
46from music_assistant_models.streamdetails import StreamDetails
47
48from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
49from music_assistant.controllers.cache import use_cache
50from music_assistant.helpers.track_filter import filter_tracks
51from music_assistant.models.music_provider import MusicProvider
52
53from .constants import (
54    CONF_API_BASE_URL,
55    CONF_COOKIE,
56    CONF_QUALITY,
57    CONF_UID,
58    DEFAULT_API_BASE_URL,
59    QUALITY_EXHIGH,
60    QUALITY_HIGHER,
61    QUALITY_HIRES,
62    QUALITY_JYEFFECT,
63    QUALITY_JYMASTER,
64    QUALITY_LOSSLESS,
65    QUALITY_STANDARD,
66)
67
68if TYPE_CHECKING:
69    from music_assistant_models.config_entries import ProviderConfig
70    from music_assistant_models.media_items import BrowseFolder, MediaItemType
71    from music_assistant_models.provider import ProviderManifest
72
73    from music_assistant.mass import MusicAssistant
74    from music_assistant.models import ProviderInstanceType
75
76
77SUPPORTED_FEATURES = {
78    ProviderFeature.LIBRARY_ARTISTS,
79    ProviderFeature.LIBRARY_ALBUMS,
80    ProviderFeature.LIBRARY_TRACKS,
81    ProviderFeature.LIBRARY_PLAYLISTS,
82    ProviderFeature.RECOMMENDATIONS,
83    ProviderFeature.SEARCH,
84    ProviderFeature.ARTIST_ALBUMS,
85    ProviderFeature.ARTIST_TOPTRACKS,
86    ProviderFeature.LYRICS,
87}
88
89_HTTP_TIMEOUT = ClientTimeout(total=20)
90_LRC_TIMESTAMP_PATTERN = re.compile(r"\[\d{1,2}:\d{2}(?:\.\d{1,3})?\]")
91_LRC_META_TAG_PATTERN = re.compile(r"^\[[a-zA-Z]+:.*\]$")
92_RECOMMEND_NEWSONG_TTL = 60 * 30
93_RECOMMEND_PLAYLIST_TTL = 60 * 60
94_RECOMMEND_DAILY_TTL = 60 * 30
95_RECOMMEND_PERSONAL_FM_TTL = 60 * 5
96_RECOMMEND_HEART_MODE_TTL = 60 * 60
97CACHE_CATEGORY_RECOMMENDATIONS = 1
98# NetEase song-detail payload uses this bit in `hr`/`h` mark metadata to indicate
99# that the track has a Hi-Res tier in catalog metadata.
100# Value observed from NeteaseCloudMusicApi-compatible responses.
101_HIRES_MARK_FLAG = 17179869184
102_PLAYLIST_PERSONAL_FM_ID = "personal_fm_dynamic"
103_PLAYLIST_HEART_MODE_PREFIX = "heart_mode_dynamic"
104_NCM_PROVIDER_ICON_URL = (
105    "https://raw.githubusercontent.com/NeteaseCloudMusicApiEnhanced/"
106    "api-enhanced/main/public/docs/netease.png"
107)
108
109
110def _to_positive_int(value: Any) -> int:
111    """Convert unknown value to positive int, otherwise return 0."""
112    if isinstance(value, bool):
113        return 0
114    if isinstance(value, (int, float)):
115        parsed = int(value)
116        return max(0, parsed)
117    if isinstance(value, str):
118        stripped = value.strip()
119        if not stripped:
120            return 0
121        with suppress(ValueError):
122            parsed = int(float(stripped))
123            return max(0, parsed)
124    return 0
125
126
127def _lrc_to_plain_text(lrc_text: str) -> str:
128    """Convert timestamped lrc to plain multi-line lyric text."""
129    lines: list[str] = []
130    for raw_line in lrc_text.splitlines():
131        line = _LRC_TIMESTAMP_PATTERN.sub("", raw_line).strip()
132        if not line or _LRC_META_TAG_PATTERN.match(line):
133            continue
134        lines.append(line)
135    return "\n".join(lines).strip()
136
137
138def _extract_song_image_url(song_obj: dict[str, Any]) -> str | None:
139    """Extract best-effort cover image URL from a song payload object."""
140    album_raw = (
141        song_obj.get("al") if isinstance(song_obj.get("al"), dict) else song_obj.get("album")
142    )
143    album_data = album_raw if isinstance(album_raw, dict) else {}
144    for candidate in (
145        album_data.get("picUrl"),
146        album_data.get("coverUrl"),
147        song_obj.get("picUrl"),
148        song_obj.get("albumPic"),
149    ):
150        if isinstance(candidate, str) and candidate.strip():
151            return candidate.strip()
152    return None
153
154
155def _parse_track_duration_seconds(song_obj: dict[str, Any]) -> int:
156    """Parse track duration in seconds from NCM payload fields with known units."""
157    # Fields documented/observed in NCM payloads as milliseconds.
158    duration_ms_candidates = (
159        song_obj.get("dt"),
160        song_obj.get("duration"),
161        song_obj.get("songTime"),
162        song_obj.get("durationMs"),
163        song_obj.get("playTime"),
164        (
165            song_obj.get("bMusic", {}).get("playTime")
166            if isinstance(song_obj.get("bMusic"), dict)
167            else None
168        ),
169    )
170    for duration_ms in duration_ms_candidates:
171        parsed_ms = _to_positive_int(duration_ms)
172        if parsed_ms > 0:
173            return parsed_ms // 1000
174
175    # Optional normalized fields that may already be in seconds.
176    duration_sec_candidates = (
177        song_obj.get("durationSec"),
178        song_obj.get("durationSeconds"),
179        song_obj.get("lengthSeconds"),
180    )
181    for duration_sec in duration_sec_candidates:
182        parsed_sec = _to_positive_int(duration_sec)
183        if parsed_sec > 0:
184            return parsed_sec
185    return 0
186
187
188class NcmApiClient:
189    """Small async client for NeteaseCloudMusicApi-compatible endpoints."""
190
191    def __init__(self, session: ClientSession, base_url: str) -> None:
192        """Initialize API client."""
193        self._session = session
194        self._base_url = base_url.rstrip("/")
195
196    async def get(
197        self,
198        path: str,
199        *,
200        params: dict[str, Any] | None = None,
201        cookie: str | None = None,
202        allow_codes: set[int] | None = None,
203    ) -> dict[str, Any]:
204        """Perform GET request and validate common NCM response format."""
205        req_params: dict[str, Any] = {}
206        if params:
207            req_params.update(params)
208        headers: dict[str, str] = {
209            "User-Agent": (
210                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
211                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
212            )
213        }
214        if cookie:
215            headers["Cookie"] = cookie
216        url = f"{self._base_url}/{path.lstrip('/')}"
217        try:
218            async with self._session.get(
219                url,
220                params=req_params,
221                headers=headers,
222                timeout=_HTTP_TIMEOUT,
223            ) as resp:
224                text = await resp.text()
225                if resp.status >= 400:
226                    raise ResourceTemporarilyUnavailable(
227                        f"Netease API HTTP {resp.status} for {path}",
228                        backoff_time=20,
229                    )
230        except TimeoutError as err:
231            raise ResourceTemporarilyUnavailable(
232                f"Netease API timeout for {url}. "
233                "Please verify API base URL and that the backend service is running.",
234                backoff_time=20,
235            ) from err
236        except ClientError as err:
237            raise ResourceTemporarilyUnavailable(
238                f"Netease API network error for {path}: {err}",
239                backoff_time=20,
240            ) from err
241        try:
242            payload = json.loads(text)
243        except json.JSONDecodeError as err:
244            raise InvalidDataError(f"Netease API returned invalid JSON for {path}") from err
245        if not isinstance(payload, dict):
246            raise InvalidDataError(f"Netease API payload is not an object for {path}")
247        code = _extract_code(payload)
248        if allow_codes and code in allow_codes:
249            return payload
250        if code not in (None, 200):
251            raise InvalidDataError(f"Netease API error code {code} for {path}")
252        return payload
253
254
255def _extract_code(payload: dict[str, Any]) -> int | None:
256    """Extract API code from payload."""
257    raw_code = payload.get("code")
258    if raw_code is None and isinstance(payload.get("data"), dict):
259        raw_code = payload["data"].get("code")
260    try:
261        return int(raw_code) if raw_code is not None else None
262    except TypeError, ValueError:
263        return None
264
265
266def _extract_data(payload: dict[str, Any]) -> dict[str, Any]:
267    """Return payload.data when it is an object, otherwise payload itself."""
268    data = payload.get("data")
269    if isinstance(data, dict):
270        return data
271    return payload
272
273
274def _extract_item_id(item: dict[str, Any], *keys: str) -> str:
275    """Return a provider item id from the first populated key."""
276    for key in keys:
277        value = item.get(key)
278        if value is None:
279            continue
280        if isinstance(value, bool) or not isinstance(value, str | int):
281            return ""
282        return str(value).strip()
283    return ""
284
285
286def _extract_cookie(payload: dict[str, Any]) -> str:
287    """Extract login cookie string from payload."""
288    data = _extract_data(payload)
289    for candidate in (data.get("cookie"), payload.get("cookie")):
290        if isinstance(candidate, str) and candidate.strip():
291            return candidate.strip()
292    return ""
293
294
295def _with_pc_os_cookie(cookie: str) -> str:
296    """
297    Return cookie string with os=pc for quality URL consistency.
298
299    Netease API may return lower-tier URLs for non-pc `os` cookies even for
300    entitled accounts. This hint only stabilizes server-side format selection;
301    entitlement still comes from upstream account/song permission checks and we
302    do not bypass locked content.
303    """
304    if not cookie.strip():
305        return cookie
306    parts = [part.strip() for part in cookie.split(";") if part.strip()]
307    kept: list[str] = []
308    os_set = False
309    for part in parts:
310        if "=" not in part:
311            kept.append(part)
312            continue
313        key, value = part.split("=", 1)
314        if key.strip().lower() == "os":
315            kept.append("os=pc")
316            os_set = True
317            continue
318        kept.append(f"{key.strip()}={value.strip()}")
319    if not os_set:
320        kept.append("os=pc")
321    return "; ".join(kept)
322
323
324async def _resolve_uid(client: NcmApiClient, cookie: str) -> str:
325    """Resolve user id from login status endpoint."""
326
327    def _as_uid(value: Any) -> str | None:
328        if value is None:
329            return None
330        text = str(value).strip()
331        return text if text and text.isdigit() else None
332
333    def _extract_uid(payload: dict[str, Any]) -> str | None:
334        data = _extract_data(payload)
335        # API variants may use one or two nested `data` wrappers.
336        containers: list[dict[str, Any]] = []
337        for candidate in (data, payload):
338            if isinstance(candidate, dict):
339                containers.append(candidate)
340                nested = candidate.get("data")
341                if isinstance(nested, dict):
342                    containers.append(nested)
343                    nested2 = nested.get("data")
344                    if isinstance(nested2, dict):
345                        containers.append(nested2)
346
347        # API variants may use different field names depending on implementation/version.
348        for container in containers:
349            profile = container.get("profile")
350            if isinstance(profile, dict):
351                for key in ("userId", "uid", "id"):
352                    if uid := _as_uid(profile.get(key)):
353                        return uid
354            account = container.get("account")
355            if isinstance(account, dict):
356                for key in ("id", "userId", "uid"):
357                    if uid := _as_uid(account.get(key)):
358                        return uid
359            for key in ("uid", "userId", "id"):
360                if uid := _as_uid(container.get(key)):
361                    return uid
362        return None
363
364    payload = await client.get(
365        "/login/status",
366        params={"timestamp": int(time.time() * 1000), "cookie": cookie},
367        cookie=cookie,
368    )
369    if uid := _extract_uid(payload):
370        return uid
371    # Fallback for API implementations that expose UID only via /user/account.
372    account_payload = await client.get(
373        "/user/account",
374        params={"timestamp": int(time.time() * 1000), "cookie": cookie},
375        cookie=cookie,
376    )
377    if uid := _extract_uid(account_payload):
378        return uid
379    raise LoginFailed("Login succeeded but user id is missing from login status")
380
381
382async def setup(
383    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
384) -> ProviderInstanceType:
385    """Initialize provider instance with given configuration."""
386    return NeteaseCloudMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
387
388
389class NeteaseCloudMusicProvider(MusicProvider):
390    """NetEase Cloud Music provider (MVP)."""
391
392    _client: NcmApiClient
393    _cookie: str
394    _uid: str
395
396    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
397        """
398        Return the configuration (options) entries for the NetEase Cloud Music provider.
399
400        Authentication runs in the interactive setup flow (see ``setup_flow.py``); the only
401        genuine option configured here is the preferred streaming quality.
402        """
403        return (
404            CONF_ENTRY_UNOFFICIAL_PROVIDER,
405            ConfigEntry(
406                key=CONF_QUALITY,
407                type=ConfigEntryType.STRING,
408                default_value=QUALITY_EXHIGH,
409                options=[
410                    ConfigValueOption(QUALITY_STANDARD),
411                    ConfigValueOption(QUALITY_HIGHER),
412                    ConfigValueOption(QUALITY_EXHIGH),
413                    ConfigValueOption(QUALITY_LOSSLESS),
414                    ConfigValueOption(QUALITY_HIRES),
415                    ConfigValueOption(QUALITY_JYEFFECT),
416                    ConfigValueOption(QUALITY_JYMASTER),
417                ],
418            ),
419        )
420
421    async def handle_async_init(self) -> None:
422        """Handle async initialization of provider."""
423        self._cookie = str(self.get_setup_value(CONF_COOKIE) or "").strip()
424        self._uid = str(self.get_setup_value(CONF_UID) or "").strip()
425        if not self._cookie:
426            raise LoginFailed("No NetEase authentication configured, please login by QR code")
427
428        api_base_url = str(self.get_setup_value(CONF_API_BASE_URL) or DEFAULT_API_BASE_URL).strip()
429        self._client = NcmApiClient(self.mass.http_session, api_base_url)
430        if not self._uid:
431            self._uid = await _resolve_uid(self._client, self._cookie)
432        self.logger.info("NetEase Cloud Music authenticated for uid %s", self._uid)
433
434    async def get_recommendations(self) -> list[RecommendationFolder]:
435        """Get this provider's available recommendation rows, without items."""
436        return [
437            RecommendationFolder(
438                item_id="recommended_radios",
439                provider=self.instance_id,
440                name="Personal Radio",
441                translation_key="personal_radio",
442                icon="mdi:radio",
443            ),
444            RecommendationFolder(
445                item_id="daily_songs",
446                provider=self.instance_id,
447                name="Recommended tracks",
448                translation_key="recommended_tracks",
449                icon="mdi:star",
450            ),
451            RecommendationFolder(
452                item_id="recommended_new_songs",
453                provider=self.instance_id,
454                name="Recommended new tracks",
455                translation_key="recommended_new_tracks",
456                icon="mdi:music-note",
457            ),
458            RecommendationFolder(
459                item_id="recommended_playlists",
460                provider=self.instance_id,
461                name="Recommended playlists",
462                translation_key="recommended_playlists",
463                icon="mdi:playlist-music",
464            ),
465        ]
466
467    async def get_recommendation_items(
468        self, item_id: str
469    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
470        """
471        Get the items for a single recommendation row.
472
473        :param item_id: The item_id of the row, as returned by get_recommendations.
474        """
475        items: UniqueList[MediaItemType | ItemMapping | BrowseFolder] = UniqueList()
476
477        if item_id == "recommended_radios":
478            return await self._build_radio_items()
479
480        if item_id == "daily_songs":
481            daily_payload = await self._get_recommend_payload_cached(
482                "daily_songs", _RECOMMEND_DAILY_TTL, "/recommend/songs"
483            )
484            daily_data = _extract_data(daily_payload)
485            daily_songs = daily_data.get("dailySongs")
486            if isinstance(daily_songs, list):
487                for song_obj in daily_songs:
488                    if not isinstance(song_obj, dict):
489                        continue
490                    with suppress(InvalidDataError):
491                        items.append(self._parse_track(song_obj))
492                daily_tracks = [item for item in items if isinstance(item, Track)]
493                await self._fill_track_durations(daily_tracks)
494            return items
495
496        if item_id == "recommended_new_songs":
497            new_song_payload = await self._get_recommend_payload_cached(
498                "recommended_newsong",
499                _RECOMMEND_NEWSONG_TTL,
500                "/personalized/newsong",
501                {"limit": 50},
502            )
503            new_song_data = _extract_data(new_song_payload)
504            raw_new_songs = new_song_data.get("result")
505            if isinstance(raw_new_songs, list):
506                for item in raw_new_songs:
507                    if not isinstance(item, dict):
508                        continue
509                    song_obj = item.get("song") if isinstance(item.get("song"), dict) else item
510                    if not isinstance(song_obj, dict):
511                        continue
512                    with suppress(InvalidDataError):
513                        items.append(self._parse_track(song_obj))
514                new_tracks = [item for item in items if isinstance(item, Track)]
515                await self._fill_track_durations(new_tracks)
516            return items
517
518        if item_id == "recommended_playlists":
519            playlist_payload = await self._get_recommend_payload_cached(
520                "recommended_playlists",
521                _RECOMMEND_PLAYLIST_TTL,
522                "/personalized",
523                {"limit": 25},
524            )
525            playlist_data = _extract_data(playlist_payload)
526            raw_playlists = playlist_data.get("result")
527            if isinstance(raw_playlists, list):
528                for playlist_obj in raw_playlists:
529                    if not isinstance(playlist_obj, dict):
530                        continue
531                    with suppress(InvalidDataError):
532                        items.append(self._parse_playlist(playlist_obj))
533            return items
534
535        return items
536
537    def _get_item_mapping(self, media_type: MediaType, item_id: str, name: str) -> ItemMapping:
538        """Create generic item mapping."""
539        return ItemMapping(
540            media_type=media_type, item_id=item_id, provider=self.instance_id, name=name
541        )
542
543    def _ensure_square_image_url(self, url: str, size: int = 500) -> str:
544        """Return image URL with square-size hint when supported by source."""
545        if not url or "param=" in url:
546            return url
547        separator = "&" if "?" in url else "?"
548        return f"{url}{separator}param={size}y{size}"
549
550    def _normalize_image_url(self, url: str) -> str:
551        """Normalize image URL for frontend compatibility."""
552        # NCM often returns http://p*.music.126.net links.
553        # In secure/ingress contexts these can be blocked as mixed content,
554        # which makes the frontend fall back to a generic provider icon.
555        if url.startswith("http://"):
556            host = urlparse(url).hostname or ""
557            if host == "music.126.net" or host.endswith(".music.126.net"):
558                return "https://" + url[len("http://") :]
559        return url
560
561    def _make_image_list(
562        self, url: str | None, *, force_square: bool = False
563    ) -> UniqueList[MediaItemImage]:
564        """Create image list for media item."""
565        if not url:
566            return UniqueList()
567        normalized = self._normalize_image_url(url)
568        image_url = self._ensure_square_image_url(normalized) if force_square else normalized
569        return UniqueList(
570            [
571                MediaItemImage(
572                    type=ImageType.THUMB,
573                    path=image_url,
574                    provider=self.instance_id,
575                    remotely_accessible=True,
576                )
577            ]
578        )
579
580    def _get_quality_obj(self, song_obj: dict[str, Any], level: str) -> dict[str, Any] | None:
581        """Map level to corresponding quality object in song/detail payload."""
582        quality_key_map = {
583            QUALITY_STANDARD: "l",
584            QUALITY_HIGHER: "m",
585            QUALITY_EXHIGH: "h",
586            QUALITY_LOSSLESS: "sq",
587            QUALITY_HIRES: "hr",
588            QUALITY_JYEFFECT: "je",
589            QUALITY_JYMASTER: "jm",
590        }
591        quality_key = quality_key_map.get(level.lower())
592        if not quality_key:
593            return None
594        quality_obj = song_obj.get(quality_key)
595        return quality_obj if isinstance(quality_obj, dict) else None
596
597    def _infer_audio_format_from_level(
598        self, level: str, quality_obj: dict[str, Any] | None = None
599    ) -> tuple[AudioFormat, str | None]:
600        """Infer best-effort AudioFormat and optional quality label from level info."""
601        level_norm = level.lower()
602        if level_norm in (QUALITY_HIRES, QUALITY_JYEFFECT, QUALITY_JYMASTER):
603            content_type = ContentType.FLAC
604            bit_depth = 24
605            details = "Hi-Res"
606        elif level_norm == QUALITY_LOSSLESS:
607            content_type = ContentType.FLAC
608            bit_depth = 16
609            details = None
610        else:
611            content_type = ContentType.MP3
612            bit_depth = 16
613            details = None
614
615        sample_rate = _to_positive_int(quality_obj.get("sr")) if quality_obj else 0
616        bit_rate = _to_positive_int(quality_obj.get("br")) if quality_obj else 0
617        return (
618            AudioFormat(
619                content_type=content_type,
620                sample_rate=sample_rate or 44100,
621                bit_depth=bit_depth,
622                bit_rate=bit_rate or None,
623            ),
624            details,
625        )
626
627    def _normalize_level_name(self, value: Any) -> str | None:
628        """Normalize any level-like value to known quality levels."""
629        if not isinstance(value, str):
630            return None
631        level = value.strip().lower()
632        aliases = {
633            "hires": QUALITY_HIRES,
634            "hi_res": QUALITY_HIRES,
635            "hi-res": QUALITY_HIRES,
636            "dolby": QUALITY_HIRES,
637            "sky": QUALITY_HIRES,
638            "jyeffect": QUALITY_JYEFFECT,
639            "jymaster": QUALITY_JYMASTER,
640            "lossless": QUALITY_LOSSLESS,
641            "exhigh": QUALITY_EXHIGH,
642            "higher": QUALITY_HIGHER,
643            "standard": QUALITY_STANDARD,
644        }
645        return aliases.get(level)
646
647    def _detect_max_quality_level(self, song_obj: dict[str, Any]) -> str:
648        """Detect highest available quality for a track from song/detail fields."""
649        level_priority = [
650            QUALITY_JYMASTER,
651            QUALITY_JYEFFECT,
652            QUALITY_HIRES,
653            QUALITY_LOSSLESS,
654            QUALITY_EXHIGH,
655            QUALITY_HIGHER,
656            QUALITY_STANDARD,
657        ]
658
659        # 1) Prefer explicit quality objects from song/detail.
660        for level in level_priority:
661            quality_obj = self._get_quality_obj(song_obj, level)
662            if isinstance(quality_obj, dict):
663                if _to_positive_int(quality_obj.get("br")) or _to_positive_int(
664                    quality_obj.get("sr")
665                ):
666                    return level
667                if quality_obj:
668                    return level
669
670        # 2) Fallback to privilege-reported max/play/download levels.
671        privilege = song_obj.get("privilege")
672        if isinstance(privilege, dict):
673            best_idx = len(level_priority)
674            best_level: str | None = None
675            for key in ("maxBrLevel", "dlLevel", "plLevel", "flLevel"):
676                normalized = self._normalize_level_name(privilege.get(key))
677                if not normalized:
678                    continue
679                idx = level_priority.index(normalized)
680                if idx < best_idx:
681                    best_idx = idx
682                    best_level = normalized
683            if best_level:
684                return best_level
685
686        # 3) Fallback to mark bit flag (Hi-Res support).
687        mark = _to_positive_int(song_obj.get("mark"))
688        if mark and (mark & _HIRES_MARK_FLAG):
689            return QUALITY_HIRES
690
691        return QUALITY_STANDARD
692
693    def _apply_track_quality_from_song_detail(self, track: Track, song_obj: dict[str, Any]) -> None:
694        """Populate mapping quality/details from detailed song object."""
695        max_level = self._detect_max_quality_level(song_obj)
696        quality_obj = self._get_quality_obj(song_obj, max_level)
697        audio_format, quality_label = self._infer_audio_format_from_level(max_level, quality_obj)
698        for mapping in track.provider_mappings:
699            if mapping.provider_instance != self.instance_id:
700                continue
701            mapping.audio_format = audio_format
702            mapping.details = quality_label
703            break
704
705    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
706        """Parse artist object."""
707        artist_id = _extract_item_id(artist_obj, "id", "artistId")
708        if not artist_id:
709            raise InvalidDataError("Artist object missing id")
710        name = str(artist_obj.get("name") or "Unknown Artist").strip()
711        artist = Artist(
712            item_id=artist_id,
713            provider=self.instance_id,
714            name=name,
715            provider_mappings={
716                ProviderMapping(
717                    item_id=artist_id,
718                    provider_domain=self.domain,
719                    provider_instance=self.instance_id,
720                    url=f"https://music.163.com/#/artist?id={artist_id}",
721                )
722            },
723        )
724        image_url = (
725            artist_obj.get("picUrl")
726            or artist_obj.get("img1v1Url")
727            or artist_obj.get("cover")
728            or artist_obj.get("avatar")
729        )
730        if isinstance(image_url, str):
731            artist.metadata.images = self._make_image_list(image_url, force_square=True)
732        return artist
733
734    def _parse_album(self, album_obj: dict[str, Any]) -> Album:
735        """Parse album object."""
736        album_id = _extract_item_id(album_obj, "id", "albumId")
737        if not album_id:
738            raise InvalidDataError("Album object missing id")
739        name = str(album_obj.get("name") or "Unknown Album").strip()
740        album = Album(
741            item_id=album_id,
742            provider=self.instance_id,
743            name=name,
744            provider_mappings={
745                ProviderMapping(
746                    item_id=album_id,
747                    provider_domain=self.domain,
748                    provider_instance=self.instance_id,
749                    url=f"https://music.163.com/#/album?id={album_id}",
750                )
751            },
752        )
753        if artists := album_obj.get("artists") or album_obj.get("ar"):
754            if isinstance(artists, list):
755                album.artists = UniqueList()
756                for artist_obj in artists:
757                    if not isinstance(artist_obj, dict):
758                        continue
759                    artist_id = str(artist_obj.get("id") or "").strip()
760                    artist_name = str(artist_obj.get("name") or "Unknown Artist").strip()
761                    if artist_id:
762                        album.artists.append(
763                            self._get_item_mapping(MediaType.ARTIST, artist_id, artist_name)
764                        )
765        image_url = (
766            album_obj.get("picUrl")
767            or album_obj.get("coverUrl")
768            or album_obj.get("blurPicUrl")
769            or album_obj.get("albumPic")
770        )
771        if isinstance(image_url, str):
772            album.metadata.images = self._make_image_list(image_url)
773        publish_time = album_obj.get("publishTime")
774        if isinstance(publish_time, int) and publish_time > 0:
775            with suppress(OSError, OverflowError, ValueError):
776                album.year = datetime.fromtimestamp(publish_time / 1000, tz=UTC).year
777        return album
778
779    def _parse_track(self, song_obj: dict[str, Any]) -> Track:
780        """Parse song object."""
781        track_id = _extract_item_id(song_obj, "id", "songId")
782        if not track_id:
783            raise InvalidDataError("Track object missing id")
784        name = str(song_obj.get("name") or "Unknown Track").strip()
785        duration = _parse_track_duration_seconds(song_obj)
786        max_level = self._detect_max_quality_level(song_obj)
787        max_quality_obj = self._get_quality_obj(song_obj, max_level)
788        max_audio_format, max_quality_label = self._infer_audio_format_from_level(
789            max_level, max_quality_obj
790        )
791        track = Track(
792            item_id=track_id,
793            provider=self.instance_id,
794            name=name,
795            duration=duration,
796            provider_mappings={
797                ProviderMapping(
798                    item_id=track_id,
799                    provider_domain=self.domain,
800                    provider_instance=self.instance_id,
801                    audio_format=max_audio_format,
802                    url=f"https://music.163.com/#/song?id={track_id}",
803                    details=max_quality_label,
804                )
805            },
806        )
807
808        artists_raw = song_obj.get("ar") or song_obj.get("artists")
809        if isinstance(artists_raw, list):
810            track.artists = UniqueList()
811            for artist_obj in artists_raw:
812                if not isinstance(artist_obj, dict):
813                    continue
814                artist_id = str(artist_obj.get("id") or "").strip()
815                artist_name = str(artist_obj.get("name") or "Unknown Artist").strip()
816                if artist_id:
817                    track.artists.append(
818                        self._get_item_mapping(MediaType.ARTIST, artist_id, artist_name)
819                    )
820
821        album_raw = song_obj.get("al") or song_obj.get("album")
822        if isinstance(album_raw, dict):
823            album_id = str(album_raw.get("id") or "").strip()
824            album_name = str(album_raw.get("name") or "Unknown Album").strip()
825            if album_id:
826                track.album = self._get_item_mapping(MediaType.ALBUM, album_id, album_name)
827            image_url = (
828                album_raw.get("picUrl")
829                or album_raw.get("coverUrl")
830                or album_raw.get("blurPicUrl")
831                or song_obj.get("picUrl")
832                or song_obj.get("albumPic")
833            )
834            if isinstance(image_url, str):
835                track.metadata.images = self._make_image_list(image_url)
836        return track
837
838    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
839        """Parse playlist object."""
840        playlist_id = _extract_item_id(playlist_obj, "id", "playlistId")
841        if not playlist_id:
842            raise InvalidDataError("Playlist object missing id")
843        name = str(playlist_obj.get("name") or "Unknown Playlist").strip()
844        playlist = Playlist(
845            item_id=playlist_id,
846            provider=self.instance_id,
847            name=name,
848            provider_mappings={
849                ProviderMapping(
850                    item_id=playlist_id,
851                    provider_domain=self.domain,
852                    provider_instance=self.instance_id,
853                    url=f"https://music.163.com/#/playlist?id={playlist_id}",
854                )
855            },
856        )
857        if isinstance(playlist_obj.get("description"), str):
858            playlist.metadata.description = str(playlist_obj["description"]).strip()
859        image_url = playlist_obj.get("coverImgUrl") or playlist_obj.get("picUrl")
860        if isinstance(image_url, str):
861            playlist.metadata.images = self._make_image_list(image_url)
862        return playlist
863
864    def _build_dynamic_playlist(
865        self,
866        item_id: str,
867        name: str,
868        translation_key: str | None = None,
869        image_url: str | None = None,
870    ) -> Playlist:
871        """Create a dynamic playlist entry for radio-like flows."""
872        playlist = Playlist(
873            item_id=item_id,
874            provider=self.instance_id,
875            name=name,
876            translation_key=translation_key,
877            provider_mappings={
878                ProviderMapping(
879                    item_id=item_id,
880                    provider_domain=self.domain,
881                    provider_instance=self.instance_id,
882                )
883            },
884            is_dynamic=True,
885        )
886        # Prefer real station/source artwork, fallback to provider icon.
887        playlist.metadata.images = self._make_image_list(image_url or _NCM_PROVIDER_ICON_URL)
888        return playlist
889
890    def _parse_heart_mode_playlist_id(self, playlist_id: str) -> tuple[str, str] | None:
891        """Parse heart mode dynamic playlist id into (seed_song_id, playlist_id)."""
892        if not playlist_id.startswith(f"{_PLAYLIST_HEART_MODE_PREFIX}:"):
893            return None
894        parts = playlist_id.split(":")
895        if len(parts) != 3:
896            return None
897        seed_song_id, source_playlist_id = parts[1], parts[2]
898        if not seed_song_id.isdigit() or not source_playlist_id.isdigit():
899            return None
900        return seed_song_id, source_playlist_id
901
902    async def _get_song_detail(self, ids: str) -> list[dict[str, Any]]:
903        """Fetch song details for one or many ids."""
904        payload = await self._client.get("/song/detail", params={"ids": ids}, cookie=self._cookie)
905        data = _extract_data(payload)
906        songs = data.get("songs")
907        if isinstance(songs, list):
908            return [item for item in songs if isinstance(item, dict)]
909        return []
910
911    async def _get_song_music_detail(self, song_id: str) -> dict[str, Any] | None:
912        """Fetch extended quality info (jm/je/hr...) for a single song."""
913        payload = await self._client.get(
914            "/song/music/detail",
915            params={"id": song_id},
916            cookie=self._cookie,
917        )
918        return _extract_data(payload)
919
920    def _merge_quality_objects(
921        self, base_song_obj: dict[str, Any], quality_song_obj: dict[str, Any]
922    ) -> dict[str, Any]:
923        """Merge quality objects from song/music/detail into song/detail object."""
924        merged = dict(base_song_obj)
925        for key in ("jm", "je", "hr", "sq", "h", "m", "l"):
926            value = quality_song_obj.get(key)
927            if isinstance(value, dict):
928                merged[key] = value
929        return merged
930
931    async def _enrich_tracks_with_cover(self, tracks: list[Track]) -> None:
932        """Enrich track cover/quality by querying song/detail in chunks."""
933        track_ids = [track.item_id for track in tracks if track.item_id]
934        if not track_ids:
935            return
936
937        details_by_id: dict[str, dict[str, Any]] = {}
938        chunk_size = 200
939        for idx in range(0, len(track_ids), chunk_size):
940            chunk = track_ids[idx : idx + chunk_size]
941            with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
942                detail_rows = await self._get_song_detail(",".join(chunk))
943                for row in detail_rows:
944                    row_id = str(row.get("id") or "").strip()
945                    if row_id:
946                        details_by_id[row_id] = row
947
948        async def _fetch_quality(track_id: str) -> tuple[str, dict[str, Any] | None]:
949            try:
950                quality_obj = await self._get_song_music_detail(track_id)
951            except InvalidDataError, ResourceTemporarilyUnavailable:
952                return track_id, None
953            return track_id, quality_obj if isinstance(quality_obj, dict) else None
954
955        semaphore = asyncio.Semaphore(8)
956
957        async def _bounded_fetch(track_id: str) -> tuple[str, dict[str, Any] | None]:
958            async with semaphore:
959                return await _fetch_quality(track_id)
960
961        quality_tasks = [
962            _bounded_fetch(track.item_id)
963            for track in tracks
964            if track.item_id and track.item_id in details_by_id
965        ]
966        quality_by_id = {
967            track_id: quality_obj
968            for track_id, quality_obj in (await asyncio.gather(*quality_tasks))
969            if isinstance(quality_obj, dict)
970        }
971
972        for track in tracks:
973            detail_obj = details_by_id.get(track.item_id)
974            if not isinstance(detail_obj, dict):
975                continue
976            quality_obj = quality_by_id.get(track.item_id)
977            if isinstance(quality_obj, dict):
978                detail_obj = self._merge_quality_objects(detail_obj, quality_obj)
979            if not track.metadata.images:
980                album_raw = detail_obj.get("al") or detail_obj.get("album")
981                if isinstance(album_raw, dict):
982                    image_url = (
983                        album_raw.get("picUrl")
984                        or album_raw.get("coverUrl")
985                        or album_raw.get("blurPicUrl")
986                        or detail_obj.get("picUrl")
987                        or detail_obj.get("albumPic")
988                    )
989                    if isinstance(image_url, str) and image_url:
990                        track.metadata.images = self._make_image_list(image_url)
991            self._apply_track_quality_from_song_detail(track, detail_obj)
992
993    async def _fill_track_durations(self, tracks: list[Track]) -> None:
994        """Fill missing track durations in bulk from song/detail."""
995        missing_tracks = [track for track in tracks if not track.duration and track.item_id]
996        if not missing_tracks:
997            return
998        track_by_id = {track.item_id: track for track in missing_tracks}
999        chunk_size = 200
1000        ids = list(track_by_id)
1001        for idx in range(0, len(ids), chunk_size):
1002            chunk = ids[idx : idx + chunk_size]
1003            rows_by_id: dict[str, dict[str, Any]] = {}
1004            with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1005                detail_rows = await self._get_song_detail(",".join(chunk))
1006                for row in detail_rows:
1007                    row_id = str(row.get("id") or "").strip()
1008                    if row_id:
1009                        rows_by_id[row_id] = row
1010            # Some API deployments do not consistently support multi-id lookup.
1011            # Fallback to single-track detail requests when needed.
1012            missing_ids = [track_id for track_id in chunk if track_id not in rows_by_id]
1013            for track_id in missing_ids:
1014                with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1015                    single_rows = await self._get_song_detail(track_id)
1016                    if single_rows and isinstance(single_rows[0], dict):
1017                        rows_by_id[track_id] = single_rows[0]
1018            for track_id in chunk:
1019                track = track_by_id.get(track_id)
1020                if not track or track.duration:
1021                    continue
1022                row_data = rows_by_id.get(track_id)
1023                if row_data is None:
1024                    continue
1025                duration = _parse_track_duration_seconds(row_data)
1026                if duration > 0:
1027                    track.duration = duration
1028
1029    def _search_plan(self, media_types: list[MediaType]) -> list[tuple[MediaType, int]]:
1030        """Build NCM search type plan from requested media types."""
1031        plan: list[tuple[MediaType, int]] = []
1032        if MediaType.TRACK in media_types:
1033            plan.append((MediaType.TRACK, 1))
1034        if MediaType.ARTIST in media_types:
1035            plan.append((MediaType.ARTIST, 100))
1036        if MediaType.ALBUM in media_types:
1037            plan.append((MediaType.ALBUM, 10))
1038        if MediaType.PLAYLIST in media_types:
1039            plan.append((MediaType.PLAYLIST, 1000))
1040        return plan
1041
1042    async def _search_single(
1043        self, search_query: str, *, type_code: int, limit: int
1044    ) -> dict[str, Any]:
1045        """Run one NCM search request by type code."""
1046        return await self._client.get(
1047            "/search",
1048            params={"keywords": search_query, "type": type_code, "limit": limit},
1049            cookie=self._cookie,
1050        )
1051
1052    def _parse_search_tracks(self, search_result: dict[str, Any], limit: int) -> list[Track]:
1053        """Parse track search result items."""
1054        tracks: list[Track] = []
1055        songs = search_result.get("songs")
1056        if not isinstance(songs, list):
1057            return tracks
1058        for song in songs[:limit]:
1059            if not isinstance(song, dict):
1060                continue
1061            with suppress(InvalidDataError):
1062                tracks.append(self._parse_track(song))
1063        return tracks
1064
1065    def _parse_search_artists(self, search_result: dict[str, Any], limit: int) -> list[Artist]:
1066        """Parse artist search result items."""
1067        artists_result: list[Artist] = []
1068        artists = search_result.get("artists")
1069        if not isinstance(artists, list):
1070            return artists_result
1071        for artist in artists[:limit]:
1072            if not isinstance(artist, dict):
1073                continue
1074            with suppress(InvalidDataError):
1075                artists_result.append(self._parse_artist(artist))
1076        return artists_result
1077
1078    def _parse_search_albums(self, search_result: dict[str, Any], limit: int) -> list[Album]:
1079        """Parse album search result items."""
1080        albums_result: list[Album] = []
1081        albums = search_result.get("albums")
1082        if not isinstance(albums, list):
1083            return albums_result
1084        for album in albums[:limit]:
1085            if not isinstance(album, dict):
1086                continue
1087            with suppress(InvalidDataError):
1088                albums_result.append(self._parse_album(album))
1089        return albums_result
1090
1091    def _parse_search_playlists(self, search_result: dict[str, Any], limit: int) -> list[Playlist]:
1092        """Parse playlist search result items."""
1093        playlists_result: list[Playlist] = []
1094        playlists = search_result.get("playlists")
1095        if not isinstance(playlists, list):
1096            return playlists_result
1097        for playlist in playlists[:limit]:
1098            if not isinstance(playlist, dict):
1099                continue
1100            with suppress(InvalidDataError):
1101                playlists_result.append(self._parse_playlist(playlist))
1102        return playlists_result
1103
1104    async def search(
1105        self,
1106        search_query: str,
1107        media_types: list[MediaType],
1108        limit: int = 5,
1109    ) -> SearchResults:
1110        """Perform search on NetEase Cloud Music."""
1111        result = SearchResults()
1112        track_results: list[Track] = []
1113        artist_results: list[Artist] = []
1114        album_results: list[Album] = []
1115        playlist_results: list[Playlist] = []
1116        search_plan = self._search_plan(media_types)
1117        responses = await asyncio.gather(
1118            *[
1119                self._search_single(search_query, type_code=type_code, limit=limit)
1120                for _, type_code in search_plan
1121            ],
1122            return_exceptions=True,
1123        )
1124        for idx, (media_type, _) in enumerate(search_plan):
1125            response = responses[idx]
1126            if isinstance(response, BaseException):
1127                self.logger.debug("NCM search failed for media_type=%s: %s", media_type, response)
1128                continue
1129            data = _extract_data(response)
1130            search_result = data.get("result")
1131            if not isinstance(search_result, dict):
1132                continue
1133            if media_type == MediaType.TRACK:
1134                track_results.extend(self._parse_search_tracks(search_result, limit))
1135            elif media_type == MediaType.ARTIST:
1136                artist_results.extend(self._parse_search_artists(search_result, limit))
1137            elif media_type == MediaType.ALBUM:
1138                album_results.extend(self._parse_search_albums(search_result, limit))
1139            elif media_type == MediaType.PLAYLIST:
1140                playlist_results.extend(self._parse_search_playlists(search_result, limit))
1141        result.tracks = track_results
1142        await self._enrich_tracks_with_cover(track_results)
1143        result.artists = artist_results
1144        result.albums = album_results
1145        result.playlists = playlist_results
1146        return result
1147
1148    @use_cache(3600 * 24)
1149    async def get_artist(self, prov_artist_id: str) -> Artist:
1150        """Get full artist details by id."""
1151        payload = await self._client.get(
1152            "/artist/detail",
1153            params={"id": prov_artist_id},
1154            cookie=self._cookie,
1155        )
1156        data = _extract_data(payload)
1157        artist_obj = data.get("artist")
1158        if not isinstance(artist_obj, dict):
1159            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
1160        return self._parse_artist(artist_obj)
1161
1162    @use_cache(3600 * 24, allow_expired_cache=True)
1163    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
1164        """Get all albums for an artist."""
1165        limit = 100
1166        offset = 0
1167        seen_album_ids: set[str] = set()
1168        albums: list[Album] = []
1169        for _ in range(50):
1170            payload = await self._client.get(
1171                "/artist/album",
1172                params={"id": prov_artist_id, "limit": limit, "offset": offset},
1173                cookie=self._cookie,
1174            )
1175            data = _extract_data(payload)
1176            raw_albums = data.get("hotAlbums") or data.get("albums")
1177            if not isinstance(raw_albums, list) or not raw_albums:
1178                break
1179
1180            for album_obj in raw_albums:
1181                if not isinstance(album_obj, dict):
1182                    continue
1183                album_id = str(album_obj.get("id") or album_obj.get("albumId") or "").strip()
1184                if album_id and album_id in seen_album_ids:
1185                    continue
1186                with suppress(InvalidDataError):
1187                    album = self._parse_album(album_obj)
1188                    albums.append(album)
1189                    seen_album_ids.add(album.item_id)
1190
1191            has_more = bool(data.get("more") or data.get("hasMore"))
1192            offset += limit
1193            if not has_more and len(raw_albums) < limit:
1194                break
1195        return albums
1196
1197    @use_cache(3600 * 24, allow_expired_cache=True)
1198    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
1199        """Get top tracks for given artist."""
1200        payload = await self._client.get(
1201            "/artist/top/song",
1202            params={"id": prov_artist_id},
1203            cookie=self._cookie,
1204        )
1205        data = _extract_data(payload)
1206        songs = data.get("songs")
1207        if not isinstance(songs, list):
1208            return []
1209        tracks: list[Track] = []
1210        for song_obj in songs:
1211            if not isinstance(song_obj, dict):
1212                continue
1213            with suppress(InvalidDataError):
1214                track = self._parse_track(song_obj)
1215                if track.duration <= 0:
1216                    # Album payload duration fields are authoritative for this endpoint;
1217                    # keep an explicit fallback here to avoid zero-length tracks.
1218                    duration = _parse_track_duration_seconds(song_obj)
1219                    if duration > 0:
1220                        track.duration = duration
1221                tracks.append(track)
1222        return tracks
1223
1224    @use_cache(3600 * 24)
1225    async def get_album(self, prov_album_id: str) -> Album:
1226        """Get full album details by id."""
1227        payload = await self._client.get(
1228            "/album",
1229            params={"id": prov_album_id},
1230            cookie=self._cookie,
1231        )
1232        data = _extract_data(payload)
1233        album_obj = data.get("album")
1234        if not isinstance(album_obj, dict):
1235            raise MediaNotFoundError(f"Album {prov_album_id} not found")
1236        return self._parse_album(album_obj)
1237
1238    @use_cache(allow_expired_cache=True)
1239    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
1240        """Get album tracks for album id."""
1241        payload = await self._client.get(
1242            "/album",
1243            params={"id": prov_album_id},
1244            cookie=self._cookie,
1245        )
1246        data = _extract_data(payload)
1247        songs = data.get("songs")
1248        if not isinstance(songs, list):
1249            return []
1250        tracks: list[Track] = []
1251        for song_obj in songs:
1252            if not isinstance(song_obj, dict):
1253                continue
1254            with suppress(InvalidDataError):
1255                tracks.append(self._parse_track(song_obj))
1256        return tracks
1257
1258    @use_cache()
1259    async def get_track(self, prov_track_id: str) -> Track:
1260        """Get full track details by id."""
1261        songs = await self._get_song_detail(prov_track_id)
1262        if not songs:
1263            raise MediaNotFoundError(f"Track {prov_track_id} not found")
1264        song_obj = songs[0]
1265        with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1266            quality_obj = await self._get_song_music_detail(prov_track_id)
1267            if isinstance(quality_obj, dict):
1268                song_obj = self._merge_quality_objects(song_obj, quality_obj)
1269        track = self._parse_track(song_obj)
1270        with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1271            lyric_payload = await self._client.get(
1272                "/lyric",
1273                params={"id": prov_track_id},
1274                cookie=self._cookie,
1275            )
1276            lyric_data = _extract_data(lyric_payload)
1277            lrc = (
1278                lyric_data.get("lrc", {}).get("lyric")
1279                if isinstance(lyric_data.get("lrc"), dict)
1280                else ""
1281            )
1282            tlyric = (
1283                lyric_data.get("tlyric", {}).get("lyric")
1284                if isinstance(lyric_data.get("tlyric"), dict)
1285                else ""
1286            )
1287            lrc_text = str(lrc or "").strip()
1288            tlyric_text = str(tlyric or "").strip()
1289            if lrc_text and _LRC_TIMESTAMP_PATTERN.search(lrc_text):
1290                track.metadata.lrc_lyrics = lrc_text
1291                track.metadata.lyrics = _lrc_to_plain_text(lrc_text) or lrc_text
1292            elif lrc_text:
1293                track.metadata.lyrics = lrc_text
1294            elif tlyric_text:
1295                track.metadata.lyrics = tlyric_text
1296        return track
1297
1298    @use_cache(3600 * 24)
1299    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
1300        """Get full playlist details by id."""
1301        if prov_playlist_id == _PLAYLIST_PERSONAL_FM_ID:
1302            return self._build_dynamic_playlist(
1303                _PLAYLIST_PERSONAL_FM_ID, "Personal FM", translation_key="personal_fm"
1304            )
1305        if heart_parts := self._parse_heart_mode_playlist_id(prov_playlist_id):
1306            seed_song_id, source_playlist_id = heart_parts
1307            return self._build_dynamic_playlist(
1308                f"{_PLAYLIST_HEART_MODE_PREFIX}:{seed_song_id}:{source_playlist_id}",
1309                "Heart Mode",
1310                translation_key="heart_mode",
1311            )
1312        if prov_playlist_id == _PLAYLIST_HEART_MODE_PREFIX:
1313            if playlist := await self._build_heart_mode_dynamic_playlist():
1314                return playlist
1315            raise MediaNotFoundError("Heart mode is currently unavailable, please try again later")
1316
1317        payload = await self._client.get(
1318            "/playlist/detail",
1319            params={"id": prov_playlist_id},
1320            cookie=self._cookie,
1321        )
1322        data = _extract_data(payload)
1323        playlist_obj = data.get("playlist")
1324        if not isinstance(playlist_obj, dict):
1325            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
1326        return self._parse_playlist(playlist_obj)
1327
1328    @use_cache(3600 * 3)
1329    async def _get_playlist_tracks_cached(
1330        self,
1331        prov_playlist_id: str,
1332        page: int = 0,
1333    ) -> Sequence[Track]:
1334        """Get playlist tracks for static playlists (cached)."""
1335        limit = 500
1336        offset = page * limit
1337        payload = await self._client.get(
1338            "/playlist/track/all",
1339            params={"id": prov_playlist_id, "limit": limit, "offset": offset},
1340            cookie=self._cookie,
1341        )
1342        data = _extract_data(payload)
1343        songs = data.get("songs")
1344        if not isinstance(songs, list):
1345            return []
1346        result: list[Track] = []
1347        for idx, song_obj in enumerate(songs, start=1):
1348            if not isinstance(song_obj, dict):
1349                continue
1350            with suppress(InvalidDataError):
1351                track = self._parse_track(song_obj)
1352                track.position = offset + idx
1353                result.append(track)
1354        return result
1355
1356    async def get_playlist_tracks(
1357        self,
1358        prov_playlist_id: str,
1359        page: int = 0,
1360    ) -> Sequence[Track]:
1361        """Get all playlist tracks for given playlist id."""
1362        if prov_playlist_id == _PLAYLIST_PERSONAL_FM_ID:
1363            if page > 0:
1364                return []
1365            tracks = await self._pick_personal_fm_tracks(fresh=True, target_count=12)
1366            for idx, track in enumerate(tracks, start=1):
1367                track.position = idx
1368            return tracks
1369        if heart_parts := self._parse_heart_mode_playlist_id(prov_playlist_id):
1370            if page > 0:
1371                return []
1372            seed_song_id, source_playlist_id = heart_parts
1373            tracks = await self._pick_heart_mode_tracks(
1374                seed_song_id,
1375                source_playlist_id,
1376                count=20,
1377            )
1378            for idx, track in enumerate(tracks, start=1):
1379                track.position = idx
1380            return tracks
1381        if prov_playlist_id == _PLAYLIST_HEART_MODE_PREFIX:
1382            if page > 0:
1383                return []
1384            if playlist := await self._build_heart_mode_dynamic_playlist():
1385                heart_parts = self._parse_heart_mode_playlist_id(playlist.item_id)
1386                if heart_parts:
1387                    seed_song_id, source_playlist_id = heart_parts
1388                    tracks = await self._pick_heart_mode_tracks(
1389                        seed_song_id,
1390                        source_playlist_id,
1391                        count=20,
1392                    )
1393                    for idx, track in enumerate(tracks, start=1):
1394                        track.position = idx
1395                    return tracks
1396            return []
1397
1398        return await self._get_playlist_tracks_cached(prov_playlist_id, page)
1399
1400    async def get_library_artists(self) -> AsyncGenerator[Artist]:
1401        """Retrieve favorite artists from NCM."""
1402        limit = 200
1403        offset = 0
1404        for _ in range(100):
1405            payload = await self._client.get(
1406                "/artist/sublist",
1407                params={"limit": limit, "offset": offset, "cookie": self._cookie},
1408                cookie=self._cookie,
1409            )
1410            data = _extract_data(payload)
1411            artists = data.get("data")
1412            if artists is None:
1413                artists = data.get("artists")
1414            if not isinstance(artists, list):
1415                raise InvalidDataError("Netease artist sublist contains no artist list")
1416            if not artists:
1417                break
1418            for artist_obj in artists:
1419                if not isinstance(artist_obj, dict):
1420                    self.report_skipped_sync_item(
1421                        MediaType.ARTIST,
1422                        None,
1423                        InvalidDataError("Netease artist sublist contains a non-object entry"),
1424                    )
1425                    continue
1426                try:
1427                    yield self._parse_artist(artist_obj)
1428                except InvalidDataError as err:
1429                    self.report_skipped_sync_item(MediaType.ARTIST, None, err)
1430            has_more = bool(data.get("more") or data.get("hasMore"))
1431            offset += limit
1432            if not has_more and len(artists) < limit:
1433                break
1434
1435    async def get_library_albums(self) -> AsyncGenerator[Album]:
1436        """Retrieve favorite albums from NCM."""
1437        limit = 200
1438        offset = 0
1439        for _ in range(100):
1440            payload = await self._client.get(
1441                "/album/sublist",
1442                params={"limit": limit, "offset": offset, "cookie": self._cookie},
1443                cookie=self._cookie,
1444            )
1445            data = _extract_data(payload)
1446            albums = data.get("data")
1447            if albums is None:
1448                albums = data.get("albums")
1449            if not isinstance(albums, list):
1450                raise InvalidDataError("Netease album sublist contains no album list")
1451            if not albums:
1452                break
1453            for album_obj in albums:
1454                if not isinstance(album_obj, dict):
1455                    self.report_skipped_sync_item(
1456                        MediaType.ALBUM,
1457                        None,
1458                        InvalidDataError("Netease album sublist contains a non-object entry"),
1459                    )
1460                    continue
1461                try:
1462                    yield self._parse_album(album_obj)
1463                except InvalidDataError as err:
1464                    self.report_skipped_sync_item(MediaType.ALBUM, None, err)
1465            has_more = bool(data.get("more") or data.get("hasMore"))
1466            offset += limit
1467            if not has_more and len(albums) < limit:
1468                break
1469
1470    async def get_library_tracks(self) -> AsyncGenerator[Track]:
1471        """Retrieve liked tracks from NCM."""
1472        payload = await self._client.get(
1473            "/likelist",
1474            params={"uid": self._uid, "cookie": self._cookie},
1475            cookie=self._cookie,
1476        )
1477        data = _extract_data(payload)
1478        ids = data.get("ids")
1479        if ids is None:
1480            ids = payload.get("ids")
1481        if not isinstance(ids, list):
1482            raise InvalidDataError("Netease liked-track response contains no id list")
1483        track_ids: list[str] = []
1484        for item in ids:
1485            item_id = str(item)
1486            if item_id.isdigit():
1487                track_ids.append(item_id)
1488                continue
1489            self.report_skipped_sync_item(
1490                MediaType.TRACK,
1491                None,
1492                InvalidDataError("Netease liked-track response contains an invalid id"),
1493            )
1494        chunk_size = 200
1495        for idx in range(0, len(track_ids), chunk_size):
1496            chunk_ids = track_ids[idx : idx + chunk_size]
1497            songs = await self._get_song_detail(",".join(chunk_ids))
1498            fetched_ids: set[str] = set()
1499            for song_obj in songs:
1500                fetched_ids.add(str(song_obj.get("id") or song_obj.get("songId") or "").strip())
1501                try:
1502                    yield self._parse_track(song_obj)
1503                except InvalidDataError as err:
1504                    self.report_skipped_sync_item(MediaType.TRACK, None, err)
1505            # the liked list is authoritative, so a track the detail call left out is still
1506            # in the library and must not be read as removed
1507            for missing_id in chunk_ids:
1508                if missing_id not in fetched_ids:
1509                    self.report_skipped_sync_item(
1510                        MediaType.TRACK,
1511                        missing_id,
1512                        MediaNotFoundError(f"NCM did not return track {missing_id}"),
1513                    )
1514
1515    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
1516        """Retrieve user playlists from NCM."""
1517        payload = await self._client.get(
1518            "/user/playlist",
1519            params={"uid": self._uid, "limit": 1000, "offset": 0, "cookie": self._cookie},
1520            cookie=self._cookie,
1521        )
1522        data = _extract_data(payload)
1523        playlists = data.get("playlist")
1524        if not isinstance(playlists, list):
1525            raise InvalidDataError("Netease user-playlist response contains no playlist list")
1526        for playlist_obj in playlists:
1527            if not isinstance(playlist_obj, dict):
1528                self.report_skipped_sync_item(
1529                    MediaType.PLAYLIST,
1530                    None,
1531                    InvalidDataError("Netease user-playlist response contains a non-object entry"),
1532                )
1533                continue
1534            try:
1535                yield self._parse_playlist(playlist_obj)
1536            except InvalidDataError as err:
1537                self.report_skipped_sync_item(MediaType.PLAYLIST, None, err)
1538
1539    async def _get_recommend_payload_cached(
1540        self,
1541        key: str,
1542        ttl: int,
1543        path: str,
1544        params: dict[str, Any] | None = None,
1545    ) -> dict[str, Any]:
1546        """Return recommendation payload from MA cache or fetch fresh."""
1547        params_key = json.dumps(params or {}, sort_keys=True, separators=(",", ":"))
1548        cache_key = f"{key}:{params_key}"
1549        cached = await self.mass.cache.get(
1550            key=cache_key,
1551            provider=self.instance_id,
1552            category=CACHE_CATEGORY_RECOMMENDATIONS,
1553            default=None,
1554        )
1555        if cached is not None:
1556            if isinstance(cached, dict):
1557                self.logger.debug("NCM recommendations %s payload cache hit", key)
1558                return cached
1559        payload = await self._client.get(path, params=params, cookie=self._cookie)
1560        await self.mass.cache.set(
1561            key=cache_key,
1562            provider=self.instance_id,
1563            category=CACHE_CATEGORY_RECOMMENDATIONS,
1564            data=payload,
1565            expiration=ttl,
1566        )
1567        return payload
1568
1569    async def _pick_personal_fm_tracks(
1570        self, *, fresh: bool = False, target_count: int = 1
1571    ) -> list[Track]:
1572        """Fetch personal FM tracks, optionally aggregate multiple fresh pulls."""
1573        if not fresh:
1574            fm_payload = await self._get_recommend_payload_cached(
1575                "personal_fm",
1576                _RECOMMEND_PERSONAL_FM_TTL,
1577                "/personal_fm",
1578            )
1579            fm_data = _extract_data(fm_payload)
1580            fm_songs = fm_data.get("data")
1581            if not isinstance(fm_songs, list) or not fm_songs:
1582                return []
1583            cached_tracks: list[Track] = []
1584            for fm_item in fm_songs:
1585                if not isinstance(fm_item, dict):
1586                    continue
1587                song_obj = fm_item.get("song") if isinstance(fm_item.get("song"), dict) else fm_item
1588                if not isinstance(song_obj, dict):
1589                    continue
1590                with suppress(InvalidDataError):
1591                    cached_tracks.append(self._parse_track(song_obj))
1592            return cached_tracks
1593
1594        # Fresh mode for dynamic playback: call endpoint in bounded batches and deduplicate tracks.
1595        result: list[Track] = []
1596        seen_ids: set[str] = set()
1597        attempts = max(2, min(max(target_count, 4), 8))
1598        no_new_rounds = 0
1599        for _ in range(attempts):
1600            fm_payload = await self._client.get(
1601                "/personal_fm",
1602                params={"timestamp": int(time.time() * 1000)},
1603                cookie=self._cookie,
1604            )
1605            fm_data = _extract_data(fm_payload)
1606            fm_songs = fm_data.get("data")
1607            if not isinstance(fm_songs, list) or not fm_songs:
1608                no_new_rounds += 1
1609                if no_new_rounds >= 2:
1610                    break
1611                continue
1612            before_count = len(result)
1613            for fm_item in fm_songs:
1614                if not isinstance(fm_item, dict):
1615                    continue
1616                song_obj = fm_item.get("song") if isinstance(fm_item.get("song"), dict) else fm_item
1617                if not isinstance(song_obj, dict):
1618                    continue
1619                with suppress(InvalidDataError):
1620                    track = self._parse_track(song_obj)
1621                    if track.item_id in seen_ids:
1622                        continue
1623                    seen_ids.add(track.item_id)
1624                    result.append(track)
1625            if len(result) == before_count:
1626                no_new_rounds += 1
1627            else:
1628                no_new_rounds = 0
1629            if len(result) >= target_count:
1630                break
1631            if no_new_rounds >= 2:
1632                break
1633        return filter_tracks(result)
1634
1635    async def _get_heart_mode_seed(self) -> tuple[str, str, str | None] | None:
1636        """Resolve heart mode seed ids as (seed_song_id, playlist_id, image_url)."""
1637        daily_payload = await self._get_recommend_payload_cached(
1638            "daily_songs",
1639            _RECOMMEND_DAILY_TTL,
1640            "/recommend/songs",
1641        )
1642        daily_data = _extract_data(daily_payload)
1643        daily_songs = daily_data.get("dailySongs")
1644        if not isinstance(daily_songs, list) or not daily_songs:
1645            return None
1646        seed_song = next(
1647            (item for item in daily_songs if isinstance(item, dict) and item.get("id")),
1648            None,
1649        )
1650        if not isinstance(seed_song, dict):
1651            return None
1652        seed_song_id = str(seed_song.get("id") or "").strip()
1653        if not seed_song_id.isdigit():
1654            return None
1655        seed_song_image = _extract_song_image_url(seed_song)
1656
1657        playlist_payload = await self._get_recommend_payload_cached(
1658            "heart_mode_playlist",
1659            _RECOMMEND_HEART_MODE_TTL,
1660            "/user/playlist",
1661            {"uid": self._uid, "limit": 1, "offset": 0},
1662        )
1663        playlist_data = _extract_data(playlist_payload)
1664        playlist_rows = playlist_data.get("playlist")
1665        if not isinstance(playlist_rows, list) or not playlist_rows:
1666            return None
1667        first_playlist = playlist_rows[0]
1668        if not isinstance(first_playlist, dict):
1669            return None
1670        playlist_id = str(first_playlist.get("id") or "").strip()
1671        if not playlist_id.isdigit():
1672            return None
1673        playlist_cover = first_playlist.get("coverImgUrl") or first_playlist.get("picUrl")
1674        if not seed_song_image and isinstance(playlist_cover, str) and playlist_cover.strip():
1675            seed_song_image = playlist_cover.strip()
1676
1677        return seed_song_id, playlist_id, seed_song_image
1678
1679    async def _build_heart_mode_dynamic_playlist(self) -> Playlist | None:
1680        """Build heart mode dynamic playlist item."""
1681        heart_parts = await self._get_heart_mode_seed()
1682        if heart_parts is None:
1683            return None
1684        seed_song_id, playlist_id, image_url = heart_parts
1685        return self._build_dynamic_playlist(
1686            f"{_PLAYLIST_HEART_MODE_PREFIX}:{seed_song_id}:{playlist_id}",
1687            "Heart Mode",
1688            translation_key="heart_mode",
1689            image_url=image_url,
1690        )
1691
1692    async def _pick_heart_mode_tracks(
1693        self, seed_song_id: str, playlist_id: str, count: int = 20
1694    ) -> list[Track]:
1695        """Fetch heart mode recommendation tracks."""
1696        payload = await self._client.get(
1697            "/playmode/intelligence/list",
1698            params={
1699                "id": seed_song_id,
1700                "pid": playlist_id,
1701                "sid": seed_song_id,
1702                "count": count,
1703                "cookie": self._cookie,
1704            },
1705            cookie=self._cookie,
1706        )
1707        data = _extract_data(payload)
1708        rows = data.get("data")
1709        if not isinstance(rows, list) or not rows:
1710            return []
1711        result: list[Track] = []
1712        for row in rows:
1713            if not isinstance(row, dict):
1714                continue
1715            song_obj = None
1716            for key in ("songInfo", "song", "songData", "trackData"):
1717                if isinstance(row.get(key), dict):
1718                    song_obj = row[key]
1719                    break
1720            if song_obj is None and isinstance(row.get("id"), (int, str)):
1721                song_obj = row
1722            if not isinstance(song_obj, dict):
1723                continue
1724            with suppress(InvalidDataError):
1725                result.append(self._parse_track(song_obj))
1726        return filter_tracks(result)
1727
1728    def _quality_candidates(self) -> list[str]:
1729        """Return ordered quality levels based on config."""
1730        raw_quality = str(self.config.get_value(CONF_QUALITY) or QUALITY_EXHIGH).lower()
1731        explicit_levels = {
1732            QUALITY_STANDARD,
1733            QUALITY_HIGHER,
1734            QUALITY_EXHIGH,
1735            QUALITY_LOSSLESS,
1736            QUALITY_HIRES,
1737            QUALITY_JYEFFECT,
1738            QUALITY_JYMASTER,
1739        }
1740        quality = (
1741            raw_quality
1742            if raw_quality in explicit_levels
1743            else (self._normalize_level_name(raw_quality) or raw_quality)
1744        )
1745        if quality == QUALITY_JYMASTER:
1746            return [
1747                QUALITY_JYMASTER,
1748                QUALITY_JYEFFECT,
1749                QUALITY_HIRES,
1750                QUALITY_LOSSLESS,
1751                QUALITY_EXHIGH,
1752                QUALITY_HIGHER,
1753                QUALITY_STANDARD,
1754            ]
1755        if quality == QUALITY_JYEFFECT:
1756            return [
1757                QUALITY_JYEFFECT,
1758                QUALITY_HIRES,
1759                QUALITY_LOSSLESS,
1760                QUALITY_EXHIGH,
1761                QUALITY_HIGHER,
1762                QUALITY_STANDARD,
1763            ]
1764        if quality == QUALITY_HIRES:
1765            return [
1766                QUALITY_JYMASTER,
1767                QUALITY_JYEFFECT,
1768                QUALITY_HIRES,
1769                QUALITY_LOSSLESS,
1770                QUALITY_EXHIGH,
1771                QUALITY_HIGHER,
1772                QUALITY_STANDARD,
1773            ]
1774        if quality == QUALITY_LOSSLESS:
1775            return [QUALITY_LOSSLESS, QUALITY_EXHIGH, QUALITY_HIGHER, QUALITY_STANDARD]
1776        if quality == QUALITY_EXHIGH:
1777            return [QUALITY_EXHIGH, QUALITY_HIGHER, QUALITY_STANDARD]
1778        if quality == QUALITY_HIGHER:
1779            return [QUALITY_HIGHER, QUALITY_STANDARD]
1780        if quality == QUALITY_STANDARD:
1781            return [QUALITY_STANDARD]
1782        return [QUALITY_EXHIGH, QUALITY_HIGHER, QUALITY_STANDARD]
1783
1784    def _parse_content_type(self, stream_type: str | None, url: str) -> ContentType:
1785        """Map stream type/ext to MA content type."""
1786        if stream_type:
1787            lowered = stream_type.lower()
1788            if lowered == "flac":
1789                return ContentType.FLAC
1790            if lowered in ("mp3", "mpeg"):
1791                return ContentType.MP3
1792            if lowered in ("aac", "m4a"):
1793                return ContentType.AAC
1794        path = urlparse(url).path.lower()
1795        if path.endswith(".flac"):
1796            return ContentType.FLAC
1797        if path.endswith(".mp3"):
1798            return ContentType.MP3
1799        if path.endswith((".m4a", ".aac")):
1800            return ContentType.AAC
1801        return ContentType.UNKNOWN
1802
1803    def _is_preview_stream(
1804        self,
1805        stream_info: dict[str, Any],
1806        track_duration_ms: int | None = None,
1807    ) -> bool:
1808        """Return True when stream payload indicates a trial/preview clip."""
1809        # 1) Strong signal: explicit free trial fragment window.
1810        free_trial_info = stream_info.get("freeTrialInfo")
1811        if isinstance(free_trial_info, dict):
1812            trial_start = _to_positive_int(free_trial_info.get("start"))
1813            trial_end = _to_positive_int(free_trial_info.get("end"))
1814            if trial_end > trial_start:
1815                return True
1816
1817        # 2) Fallback signal: returned stream duration is much shorter than track duration.
1818        # Some responses include `freeTrialPrivilege` even for playable tracks, so we do not
1819        # treat its mere presence as preview.
1820        stream_time_ms = _to_positive_int(stream_info.get("time"))
1821        if track_duration_ms and stream_time_ms:
1822            # Keep a small tolerance to avoid false positives on rounding differences.
1823            if (stream_time_ms + 5000) < track_duration_ms:
1824                return True
1825
1826        return False
1827
1828    async def _get_track_stream_details(  # noqa: PLR0915
1829        self,
1830        track_id: str,
1831        *,
1832        stream_item_id: str,
1833        media_type: MediaType,
1834        allow_seek: bool,
1835    ) -> StreamDetails:
1836        """Resolve stream details for a concrete track id."""
1837        detail_obj: dict[str, Any] | None = None
1838        track_duration_ms: int | None = None
1839        track_duration_seconds: int | None = None
1840        with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1841            detail_rows = await self._get_song_detail(track_id)
1842            if detail_rows:
1843                detail_obj = detail_rows[0]
1844                if isinstance(detail_obj, dict):
1845                    track_duration_ms = _to_positive_int(detail_obj.get("dt")) or None
1846                    parsed_duration = _parse_track_duration_seconds(detail_obj)
1847                    if parsed_duration > 0:
1848                        track_duration_seconds = parsed_duration
1849        preview_fallback: StreamDetails | None = None
1850        for requested_level in self._quality_candidates():
1851            payload = await self._client.get(
1852                "/song/url/v1",
1853                params={"id": track_id, "level": requested_level},
1854                cookie=_with_pc_os_cookie(self._cookie),
1855            )
1856            data = _extract_data(payload)
1857            stream_rows = data.get("data")
1858            if not isinstance(stream_rows, list) or not stream_rows:
1859                continue
1860            stream_info = stream_rows[0] if isinstance(stream_rows[0], dict) else {}
1861            if not stream_info:
1862                continue
1863            stream_url = str(stream_info.get("url") or "").strip()
1864            if not stream_url:
1865                continue
1866            resolved_level = str(stream_info.get("level") or requested_level).lower()
1867            normalized_level = self._normalize_level_name(resolved_level) or resolved_level
1868            preview = self._is_preview_stream(stream_info, track_duration_ms)
1869            level_quality_obj = (
1870                self._get_quality_obj(detail_obj, normalized_level)
1871                if isinstance(detail_obj, dict)
1872                else None
1873            )
1874            inferred_format, _ = self._infer_audio_format_from_level(
1875                normalized_level, level_quality_obj
1876            )
1877            detected_content_type = self._parse_content_type(
1878                str(stream_info.get("type") or stream_info.get("encodeType") or ""),
1879                stream_url,
1880            )
1881            audio_format = AudioFormat(
1882                content_type=(
1883                    inferred_format.content_type
1884                    if detected_content_type == ContentType.UNKNOWN
1885                    else detected_content_type
1886                ),
1887                sample_rate=inferred_format.sample_rate,
1888                bit_depth=inferred_format.bit_depth,
1889                bit_rate=inferred_format.bit_rate,
1890            )
1891            bitrate = stream_info.get("br")
1892            if isinstance(bitrate, int) and bitrate > 0:
1893                audio_format.bit_rate = bitrate
1894            stream_sr = _to_positive_int(stream_info.get("sr"))
1895            if stream_sr > 0 and (
1896                level_quality_obj is None or audio_format.sample_rate in (0, 44100)
1897            ):
1898                audio_format.sample_rate = stream_sr
1899            stream_time_ms = _to_positive_int(stream_info.get("time"))
1900            stream_duration_seconds = int(stream_time_ms / 1000) if stream_time_ms > 0 else None
1901            expiration = 3600
1902            with suppress(TypeError, ValueError):
1903                parsed = parse_qs(urlparse(stream_url).query)
1904                if expire_raw := parsed.get("expire", [None])[0]:
1905                    expiration = max(60, int(expire_raw) - int(time.time()))
1906            details = StreamDetails(
1907                provider=self.instance_id,
1908                item_id=stream_item_id,
1909                media_type=media_type,
1910                audio_format=audio_format,
1911                stream_type=StreamType.HTTP,
1912                path=stream_url,
1913                can_seek=allow_seek,
1914                allow_seek=allow_seek,
1915                duration=track_duration_seconds or stream_duration_seconds,
1916                expiration=expiration,
1917                data={
1918                    "preview": preview,
1919                    "requested_level": requested_level,
1920                    "resolved_level": resolved_level,
1921                    "track_id": track_id,
1922                },
1923            )
1924            self.logger.debug(
1925                "NCM stream selected item=%s track=%s requested=%s resolved=%s preview=%s format=%s/%s bitrate=%s",
1926                stream_item_id,
1927                track_id,
1928                requested_level,
1929                resolved_level,
1930                preview,
1931                audio_format.sample_rate,
1932                audio_format.bit_depth,
1933                audio_format.bit_rate,
1934            )
1935            if not preview:
1936                return details
1937            if preview_fallback is None:
1938                preview_fallback = details
1939        # If account/song entitlement only allows trial playback, return preview stream.
1940        if preview_fallback is not None:
1941            return preview_fallback
1942        raise UnplayableMediaError(f"No playable stream URL returned for track {track_id}")
1943
1944    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
1945        """Return streamdetails for track."""
1946        if media_type == MediaType.TRACK:
1947            return await self._get_track_stream_details(
1948                item_id,
1949                stream_item_id=item_id,
1950                media_type=MediaType.TRACK,
1951                allow_seek=True,
1952            )
1953        raise UnsupportedFeaturedException(f"Unsupported media type {media_type}")
1954
1955    async def _build_radio_items(self) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
1956        """Build the dynamic radio playlist items for the recommended_radios row."""
1957        items: UniqueList[MediaItemType | ItemMapping | BrowseFolder] = UniqueList()
1958        personal_fm_image_url: str | None = None
1959        with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1960            fm_payload = await self._get_recommend_payload_cached(
1961                "personal_fm",
1962                _RECOMMEND_PERSONAL_FM_TTL,
1963                "/personal_fm",
1964            )
1965            fm_data = _extract_data(fm_payload)
1966            fm_rows = fm_data.get("data")
1967            if isinstance(fm_rows, list) and fm_rows and isinstance(fm_rows[0], dict):
1968                fm_item = fm_rows[0]
1969                song_obj = fm_item.get("song") if isinstance(fm_item.get("song"), dict) else fm_item
1970                if isinstance(song_obj, dict):
1971                    personal_fm_image_url = _extract_song_image_url(song_obj)
1972        if not personal_fm_image_url:
1973            with suppress(InvalidDataError, ResourceTemporarilyUnavailable):
1974                daily_payload = await self._get_recommend_payload_cached(
1975                    "daily_songs",
1976                    _RECOMMEND_DAILY_TTL,
1977                    "/recommend/songs",
1978                )
1979                daily_data = _extract_data(daily_payload)
1980                daily_rows = daily_data.get("dailySongs")
1981                if isinstance(daily_rows, list) and daily_rows and isinstance(daily_rows[0], dict):
1982                    personal_fm_image_url = _extract_song_image_url(daily_rows[0])
1983        items.append(
1984            self._build_dynamic_playlist(
1985                _PLAYLIST_PERSONAL_FM_ID,
1986                "Personal FM",
1987                translation_key="personal_fm",
1988                image_url=personal_fm_image_url,
1989            )
1990        )
1991        if heart_playlist := await self._build_heart_mode_dynamic_playlist():
1992            items.append(heart_playlist)
1993        return items
1994