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