music-assistant-server

1"""QQ Music provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8import time
9from asyncio import Semaphore
10from collections.abc import AsyncGenerator, Awaitable, Callable
11from contextlib import suppress
12from typing import TYPE_CHECKING, Any
13from urllib.parse import parse_qs, urlparse
14
15from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
16from music_assistant_models.enums import (
17    ConfigEntryType,
18    ContentType,
19    MediaType,
20    ProviderFeature,
21    StreamType,
22)
23from music_assistant_models.errors import (
24    InvalidDataError,
25    LoginFailed,
26    MediaNotFoundError,
27    ResourceTemporarilyUnavailable,
28    UnplayableMediaError,
29)
30from music_assistant_models.media_items import (
31    Album,
32    Artist,
33    AudioFormat,
34    BrowseFolder,
35    ItemMapping,
36    MediaItemType,
37    Playlist,
38    RecommendationFolder,
39    SearchResults,
40    Track,
41    UniqueList,
42)
43from music_assistant_models.streamdetails import StreamDetails
44from qqmusic_api import (
45    CgiApiException,
46    Credential,
47    CredentialExpiredError,
48    CredentialRefreshError,
49    LoginError,
50)
51from qqmusic_api import (
52    Client as QQClient,
53)
54from qqmusic_api.algorithms import qrc_decrypt
55from qqmusic_api.modules.search import SearchType
56from qqmusic_api.modules.singer import TabType
57from qqmusic_api.modules.song import SongFileInfo, SongFileType, SpecialSongFileType
58
59from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
60from music_assistant.controllers.cache import use_cache
61from music_assistant.models.music_provider import MusicProvider
62
63from .constants import (
64    CONF_CREDENTIAL_JSON,
65    CONF_LOGIN_TYPE,
66    CONF_MUSICID,
67    CONF_MUSICKEY,
68    CONF_QUALITY,
69    CONF_UIN,
70    QUALITY_FLAC,
71    QUALITY_HI_RES,
72    QUALITY_MP3_128,
73    QUALITY_MP3_320,
74)
75from .helpers import extract_first_text, normalize_qq_lyric_text, qrc_to_lrc
76from .parsers import (
77    build_playlist_id,
78    extract_guess_recommend_tracks,
79    extract_items,
80    extract_newsong_tracks,
81    extract_radar_recommend_tracks,
82    extract_recommend_songlists,
83    extract_song_id,
84    get_artist_mapping,
85    parse_album,
86    parse_artist,
87    parse_playlist,
88    parse_playlist_id,
89    parse_track,
90)
91
92if TYPE_CHECKING:
93    from music_assistant_models.config_entries import ProviderConfig
94    from music_assistant_models.provider import ProviderManifest
95
96    from music_assistant.mass import MusicAssistant
97    from music_assistant.models import ProviderInstanceType
98
99SUPPORTED_FEATURES = {
100    ProviderFeature.LIBRARY_ARTISTS,
101    ProviderFeature.LIBRARY_ALBUMS,
102    ProviderFeature.LIBRARY_TRACKS,
103    ProviderFeature.LIBRARY_PLAYLISTS,
104    ProviderFeature.RECOMMENDATIONS,
105    ProviderFeature.SEARCH,
106    ProviderFeature.ARTIST_ALBUMS,
107    ProviderFeature.ARTIST_TRACKS,
108    ProviderFeature.ARTIST_TOPTRACKS,
109    ProviderFeature.SIMILAR_TRACKS,
110    ProviderFeature.SIMILAR_ARTISTS,
111    ProviderFeature.PLAYLIST_CREATE,
112    ProviderFeature.PLAYLIST_TRACKS_EDIT,
113    ProviderFeature.LYRICS,
114}
115
116_LRC_TIMESTAMP_PATTERN = re.compile(r"\[\d{1,2}:\d{2}(?:\.\d{1,3})?\]")
117_HEX_LYRIC_PATTERN = re.compile(r"^[0-9A-Fa-f]{32,}$")
118_RECOMMEND_GUESS_TTL = 60 * 60
119_RECOMMEND_NEWSONG_TTL = 60 * 60 * 6
120_RECOMMEND_PLAYLIST_TTL = 60 * 60 * 6
121
122
123async def setup(
124    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
125) -> ProviderInstanceType:
126    """Initialize provider(instance) with given configuration."""
127    return QQMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
128
129
130def _store_credential(values: dict[str, ConfigValueType], credential: Any) -> None:
131    if not credential.musicid or not credential.musickey:
132        raise LoginFailed("QR login succeeded but credential is incomplete")
133    if callable(getattr(credential, "model_dump_json", None)):
134        credential_json = credential.model_dump_json(by_alias=True)
135    else:
136        fallback_credential = Credential.model_validate(
137            {
138                "musicid": int(credential.musicid),
139                "musickey": str(credential.musickey),
140                "loginType": int(getattr(credential, "login_type", 2) or 2),
141                "refresh_key": str(getattr(credential, "refresh_key", "") or ""),
142                "refresh_token": str(getattr(credential, "refresh_token", "") or ""),
143                "encryptUin": str(getattr(credential, "encrypt_uin", "") or ""),
144                "str_musicid": str(getattr(credential, "str_musicid", "") or ""),
145            }
146        )
147        credential_json = fallback_credential.model_dump_json(by_alias=True)
148    values[CONF_UIN] = str(credential.musicid)
149    values[CONF_MUSICID] = str(credential.musicid)
150    values[CONF_MUSICKEY] = str(credential.musickey)
151    values[CONF_LOGIN_TYPE] = str(credential.login_type or 2)
152    values[CONF_CREDENTIAL_JSON] = credential_json
153
154
155class QQMusicProvider(MusicProvider):
156    """QQ Music provider."""
157
158    _credential: Any = None
159    _qq_search: Any = None
160    _qq_song: Any = None
161    _qq_album: Any = None
162    _qq_singer: Any = None
163    _qq_client: Any = None
164    _qq_user: Any = None
165    _qq_songlist: Any = None
166    _qq_lyric: Any = None
167    _qq_recommend: Any = None
168    _api_semaphore: Semaphore
169    _credential_refresh_lock: asyncio.Lock
170    _last_credential_check_monotonic: float
171    _musicid: int = 0
172    _euin: str = ""
173    _recommend_payload_cache: dict[str, tuple[float, Any]]
174
175    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
176        """
177        Return the configuration (options) entries for the QQ Music provider.
178
179        Authentication runs in the interactive setup flow (see ``setup_flow.py``); the only
180        genuine option configured here is the preferred streaming quality.
181        """
182        return (
183            CONF_ENTRY_UNOFFICIAL_PROVIDER,
184            ConfigEntry(
185                key=CONF_QUALITY,
186                type=ConfigEntryType.STRING,
187                default_value=QUALITY_MP3_320,
188                options=[
189                    ConfigValueOption(QUALITY_MP3_128),
190                    ConfigValueOption(QUALITY_MP3_320),
191                    ConfigValueOption(QUALITY_FLAC),
192                    ConfigValueOption(QUALITY_HI_RES),
193                ],
194            ),
195        )
196
197    async def handle_async_init(self) -> None:
198        """Validate auth and initialize qqmusic api adapters."""
199        credential: Credential | None = None
200        if credential_json := str(self.get_setup_value(CONF_CREDENTIAL_JSON) or "").strip():
201            try:
202                credential = Credential.model_validate_json(credential_json)
203            except Exception as err:
204                self.logger.warning(
205                    "Failed to parse persisted QQ credential_json, fallback to legacy fields: %s",
206                    err,
207                )
208
209        if not credential or not credential.musicid or not credential.musickey:
210            config_musicid = self.get_setup_value(CONF_MUSICID) or self.get_setup_value(CONF_UIN)
211            config_musickey = self.get_setup_value(CONF_MUSICKEY)
212            config_login_type = self.get_setup_value(CONF_LOGIN_TYPE)
213            if not (config_musicid and config_musickey):
214                raise LoginFailed("No QQ Music authentication configured, please login by QR code")
215            login_type_raw = str(config_login_type or "2")
216            login_type = int(login_type_raw) if login_type_raw.isdigit() else 2
217            credential = Credential.model_validate(
218                {
219                    "musicid": int(str(config_musicid).strip()),
220                    "musickey": str(config_musickey),
221                    "loginType": login_type,
222                }
223            )
224        if not credential.encrypt_uin:
225            raise LoginFailed(
226                "QQ Music credential is missing encryptUin, please re-authenticate by QR code"
227            )
228
229        self._qq_client = QQClient(credential=credential)
230        self._qq_search = self._qq_client.search
231        self._qq_song = self._qq_client.song
232        self._qq_album = self._qq_client.album
233        self._qq_singer = self._qq_client.singer
234        self._qq_user = self._qq_client.user
235        self._qq_songlist = self._qq_client.songlist
236        self._qq_lyric = self._qq_client.lyric
237        self._qq_recommend = self._qq_client.recommend
238        # Keep qqmusic_api internal logs in sync with MA log level.
239        logging.getLogger("qqmusicapi").setLevel(self.logger.level + 10)
240        self._credential = credential
241        self._api_semaphore = Semaphore(4)
242        self._credential_refresh_lock = asyncio.Lock()
243        self._last_credential_check_monotonic = 0.0
244        self._musicid = int(self._credential.musicid)
245        self._recommend_payload_cache = {}
246        self.logger.info("QQ Music authenticated for uin %s", self._musicid)
247        # Persist complete credential once on init so legacy configs gain refresh fields.
248        self._persist_credential()
249
250    async def get_recommendations(self) -> list[RecommendationFolder]:
251        """Get the available QQ Music recommendation rows, without items."""
252        return [
253            RecommendationFolder(
254                item_id="guess_recommend",
255                provider=self.instance_id,
256                name="Recommended tracks",
257                translation_key="recommended_tracks",
258                icon="mdi-lightbulb-on-outline",
259            ),
260            RecommendationFolder(
261                item_id="new_songs",
262                provider=self.instance_id,
263                name="Recommended new tracks",
264                translation_key="recommended_new_tracks",
265                icon="mdi-music-note-plus",
266            ),
267            RecommendationFolder(
268                item_id="recommended_playlists",
269                provider=self.instance_id,
270                name="Recommended playlists",
271                translation_key="recommended_playlists",
272                icon="mdi-playlist-music",
273            ),
274        ]
275
276    async def get_recommendation_items(
277        self, item_id: str
278    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
279        """
280        Get the items for a single QQ Music recommendation row.
281
282        :param item_id: The item_id of the row, as returned by get_recommendations.
283        """
284        items: UniqueList[MediaItemType | ItemMapping | BrowseFolder] = UniqueList()
285        if item_id == "guess_recommend":
286            guess_response = await self._get_recommend_payload_cached(
287                "guess_recommend",
288                _RECOMMEND_GUESS_TTL,
289                lambda: self._qq_recommend.get_guess_recommend(credential=self._credential),
290            )
291            for item in extract_guess_recommend_tracks(self._to_dict(guess_response)):
292                with suppress(InvalidDataError, TypeError, ValueError):
293                    items.append(self._parse_track(item))
294            if not items:
295                # Fall back to radar recommendations when the personalised
296                # guess endpoint yields no usable tracks.
297                radar_response = await self._get_recommend_payload_cached(
298                    "guess_recommend_radar",
299                    _RECOMMEND_GUESS_TTL,
300                    self._qq_recommend.get_radar_recommend,
301                )
302                for item in extract_radar_recommend_tracks(self._to_dict(radar_response)):
303                    with suppress(InvalidDataError, TypeError, ValueError):
304                        items.append(self._parse_track(item))
305        elif item_id == "new_songs":
306            new_song_response = await self._get_recommend_payload_cached(
307                "new_songs",
308                _RECOMMEND_NEWSONG_TTL,
309                self._qq_recommend.get_recommend_newsong,
310            )
311            for item in extract_newsong_tracks(self._to_dict(new_song_response)):
312                with suppress(InvalidDataError, TypeError, ValueError):
313                    items.append(self._parse_track(item))
314        elif item_id == "recommended_playlists":
315            playlist_response = await self._get_recommend_payload_cached(
316                "recommended_playlists",
317                _RECOMMEND_PLAYLIST_TTL,
318                self._qq_recommend.get_recommend_songlist,
319            )
320            for item in extract_recommend_songlists(self._to_dict(playlist_response)):
321                with suppress(InvalidDataError, TypeError, ValueError):
322                    items.append(self._parse_playlist(item))
323        return items
324
325    def _persist_credential(self) -> None:
326        """Persist the current credential into this provider's setup data."""
327        if not self._credential:
328            return
329        self._update_setup_data(CONF_UIN, str(self._credential.musicid))
330        self._update_setup_data(CONF_MUSICID, str(self._credential.musicid))
331        self._update_setup_data(CONF_MUSICKEY, str(self._credential.musickey))
332        self._update_setup_data(CONF_LOGIN_TYPE, str(self._credential.login_type or 2))
333        self._update_setup_data(
334            CONF_CREDENTIAL_JSON, self._credential.model_dump_json(by_alias=True)
335        )
336
337    async def _ensure_valid_credential(self) -> None:
338        """Refresh credential when expired and persistence data allows refresh."""
339        if not self._credential:
340            raise LoginFailed("QQ Music credential is not initialized")
341        now = time.monotonic()
342        # Avoid checking expiry on every single API call.
343        if (now - self._last_credential_check_monotonic) < 300:
344            return
345        async with self._credential_refresh_lock:
346            now = time.monotonic()
347            if (now - self._last_credential_check_monotonic) < 300:
348                return
349            self._last_credential_check_monotonic = now
350            if not self._qq_client:
351                raise LoginFailed("QQ Music client is not initialized")
352            if not await self._qq_client.login.check_expired(self._credential):
353                return
354            try:
355                self._credential = await self._qq_client.login.refresh_credential(self._credential)
356            except CredentialRefreshError as err:
357                raise LoginFailed(
358                    "QQ Music credential refresh failed, please re-authenticate"
359                ) from err
360            self._qq_client.credential = self._credential
361            self._persist_credential()
362            self.logger.info("QQ Music credential refreshed and persisted")
363
364    async def _run_with_session(self, coro: Awaitable[Any]) -> Any:
365        """Run qqmusic_api call with the provider-bound Client."""
366        try:
367            await self._ensure_valid_credential()
368            async with self._api_semaphore:
369                return await coro
370        except Exception as err:
371            raise self._translate_qq_exception(err) from err
372
373    def _translate_qq_exception(self, err: Exception) -> Exception:
374        """Translate qqmusic_api/http exceptions to MA domain exceptions."""
375        if isinstance(err, CredentialExpiredError):
376            return LoginFailed("QQ Music credential expired, please re-authenticate")
377        if isinstance(err, LoginError):
378            return LoginFailed(f"QQ Music login failed: {err}")
379        if isinstance(err, CgiApiException):
380            code = getattr(err, "code", None)
381            if code in (1000, 2000):
382                return LoginFailed(f"QQ Music API auth/sign failure (code={code})")
383            if code == 404:
384                return MediaNotFoundError("QQ Music item not found (code=404)")
385            if code == 10007:
386                return MediaNotFoundError(
387                    "QQ Music item not found or invalid provider id (code=10007)"
388                )
389            return ResourceTemporarilyUnavailable(
390                f"QQ Music API error (code={code})",
391                backoff_time=30,
392            )
393        err_str = str(err).lower()
394        if "timeout" in err_str or "temporarily" in err_str or "connection" in err_str:
395            return ResourceTemporarilyUnavailable(
396                "QQ Music network temporarily unavailable",
397                backoff_time=20,
398            )
399        return err
400
401    async def unload(self, is_removed: bool = False) -> None:
402        """Handle unload/close of provider."""
403        if self._qq_client:
404            await self._qq_client.close()
405        self._qq_client = None
406        self._recommend_payload_cache = {}
407        await super().unload(is_removed)
408
409    async def _get_recommend_payload_cached(
410        self, key: str, ttl: int, fetcher: Callable[[], Awaitable[Any]]
411    ) -> Any:
412        """Return recommendation payload from in-memory TTL cache or fetch fresh."""
413        if cached := self._recommend_payload_cache.get(key):
414            timestamp, payload = cached
415            if (time.time() - timestamp) < ttl:
416                self.logger.debug("QQ recommendations %s payload cache hit", key)
417                return payload
418        payload = await self._run_with_session(fetcher())
419        self._recommend_payload_cache[key] = (time.time(), payload)
420        return payload
421
422    def _to_dict(self, data: Any) -> dict[str, Any]:
423        """Normalize qqmusic-api response models to dictionaries."""
424        if isinstance(data, dict):
425            return data
426        if callable(dump := getattr(data, "model_dump", None)):
427            dumped = dump(by_alias=True)
428            return dumped if isinstance(dumped, dict) else {}
429        return {}
430
431    def _decode_lyric_response(self, response: Any) -> dict[str, Any]:
432        """Normalize and decrypt QQ Music lyric responses."""
433        if callable(decrypt := getattr(response, "decrypt", None)):
434            response = decrypt()
435        lyric_obj = dict(self._to_dict(response))
436        if str(lyric_obj.get("crypt") or "0") != "1":
437            return lyric_obj
438        for key in ("lyric", "trans", "roma"):
439            value = str(lyric_obj.get(key) or "").strip()
440            if value and _HEX_LYRIC_PATTERN.fullmatch(value):
441                try:
442                    lyric_obj[key] = qrc_decrypt(value)
443                except (TypeError, ValueError) as err:
444                    self.logger.debug("Failed to decrypt QQ Music %s lyric payload: %s", key, err)
445        return lyric_obj
446
447    def _response_items(self, data: Any, keys: tuple[str, ...]) -> list[dict[str, Any]]:
448        """Extract dict items from list, dict, or qqmusic-api response model."""
449        if isinstance(data, list):
450            return [item for item in data if isinstance(item, dict)]
451        return extract_items(self._to_dict(data), keys)
452
453    def _get_candidate_file_types(self) -> list[Any]:
454        """Return ordered quality candidates based on provider config."""
455        quality = str(self.config.get_value(CONF_QUALITY) or QUALITY_MP3_320)
456        if quality == QUALITY_HI_RES:
457            return [
458                SongFileType.MASTER,
459                SongFileType.FLAC,
460                SongFileType.MP3_320,
461                SongFileType.MP3_128,
462            ]
463        if quality == QUALITY_FLAC:
464            return [
465                SongFileType.FLAC,
466                SongFileType.MP3_320,
467                SongFileType.MP3_128,
468            ]
469        if quality == QUALITY_MP3_320:
470            return [
471                SongFileType.MP3_320,
472                SongFileType.MP3_128,
473            ]
474        return [SongFileType.MP3_128]
475
476    async def _resolve_stream_url(
477        self, item_id: str, track_obj: dict[str, Any]
478    ) -> tuple[str, Any | None, bool, int | None]:
479        """Resolve stream URL with full-stream and preview fallback."""
480        stream_url = ""
481        selected_file_type = None
482        is_preview_stream = False
483        preview_duration = None
484        file_obj = track_obj.get("file")
485        if not isinstance(file_obj, dict):
486            file_obj = {}
487        media_mid = str(file_obj.get("media_mid") or file_obj.get("mediaMid") or "")
488        song_mid = str(
489            track_obj.get("mid") or track_obj.get("songMid") or track_obj.get("songmid") or item_id
490        )
491        song_type = self._to_positive_int(track_obj.get("type") or track_obj.get("songtype"))
492        file_info = [SongFileInfo(song_mid, song_type=song_type, media_mid=media_mid or None)]
493
494        for file_type in self._get_candidate_file_types():
495            url_response = await self._run_with_session(
496                self._qq_song.get_song_urls(
497                    file_info,
498                    file_type=file_type,
499                    credential=self._credential,
500                )
501            )
502            url = self._extract_stream_url(url_response, song_mid)
503            if url.startswith("http"):
504                return (url, file_type, False, None)
505
506        # Fallback to 30s preview URL when full stream URL is unavailable.
507        vs_list = track_obj.get("vs")
508        if isinstance(vs_list, list):
509            first_vs = next((vs for vs in vs_list if isinstance(vs, str) and vs), None)
510            if first_vs:
511                try_file_info = [
512                    SongFileInfo(
513                        song_mid,
514                        song_type=song_type,
515                        media_mid=first_vs,
516                    )
517                ]
518                try_response = await self._run_with_session(
519                    self._qq_song.get_song_urls(
520                        try_file_info,
521                        file_type=SpecialSongFileType.TRY,
522                        credential=self._credential,
523                    )
524                )
525                if try_url := self._extract_stream_url(try_response, song_mid):
526                    stream_url = try_url
527                    selected_file_type = SpecialSongFileType.TRY
528                    is_preview_stream = True
529                    try_begin = track_obj.get("file", {}).get("try_begin")
530                    try_end = track_obj.get("file", {}).get("try_end")
531                    if (
532                        isinstance(try_begin, int)
533                        and isinstance(try_end, int)
534                        and try_end > try_begin
535                    ):
536                        preview_duration = int((try_end - try_begin) / 1000)
537                    self.logger.info(
538                        "QQ Music full stream unavailable for %s, using preview stream fallback",
539                        item_id,
540                    )
541
542        return (stream_url, selected_file_type, is_preview_stream, preview_duration)
543
544    def _extract_stream_url(self, url_response: Any, item_id: str) -> str:
545        """Extract absolute stream URL from qqmusic-api 0.6 or legacy URL payload."""
546        if isinstance(url_response, dict) and isinstance(url_response.get(item_id), str):
547            return str(url_response[item_id])
548        response = self._to_dict(url_response)
549        url_items = response.get("midurlinfo") or response.get("data") or []
550        if not isinstance(url_items, list):
551            return ""
552        cdn_base = getattr(self._qq_song, "_SONG_URL_FALLBACK_DOMAIN", None)
553        if not cdn_base:
554            self.logger.debug("QQ Music API did not expose stream URL fallback domain")
555            cdn_base = "https://isure.stream.qqmusic.qq.com/"
556        cdn_base = str(cdn_base)
557        for item in url_items:
558            if not isinstance(item, dict):
559                continue
560            if str(item.get("songmid") or item.get("mid") or "") != item_id:
561                continue
562            purl = str(item.get("purl") or "")
563            if purl.startswith("http"):
564                return purl
565            if purl:
566                return f"{cdn_base.rstrip('/')}/{purl.lstrip('/')}"
567        return ""
568
569    def _get_stream_expiration(self, stream_url: str) -> int:
570        """Derive expiration from stream URL query string."""
571        expiration = 3600
572        if parsed_qs := parse_qs(urlparse(stream_url).query):
573            for param in ("Expires", "expire"):
574                if expire_ts := parsed_qs.get(param, [None])[0]:
575                    expiration = max(30, int(expire_ts) - int(time.time()) - 10)
576                    break
577        return expiration
578
579    def _get_content_type(self, selected_file_type: Any | None) -> ContentType:
580        """Map qqmusic file type enum to MA content type."""
581        if not selected_file_type:
582            return ContentType.UNKNOWN
583        if selected_file_type in (SongFileType.FLAC, SongFileType.MASTER):
584            return ContentType.FLAC
585        if selected_file_type in (
586            SongFileType.ACC_48,
587            SongFileType.ACC_96,
588            SongFileType.ACC_192,
589        ):
590            return ContentType.M4A
591        return ContentType.MPEG
592
593    @staticmethod
594    def _to_positive_int(value: Any) -> int:
595        """Convert value to positive int, fallback to 0."""
596        with suppress(TypeError, ValueError):
597            parsed = int(value)
598            if parsed > 0:
599                return parsed
600        return 0
601
602    def _file_size(self, file_obj: dict[str, Any], *keys: str) -> int:
603        """Read first positive file size from multiple key variants."""
604        for key in keys:
605            if size := self._to_positive_int(file_obj.get(key)):
606                return size
607        return 0
608
609    def _get_max_supported_audio_format(
610        self, track_obj: dict[str, Any]
611    ) -> tuple[AudioFormat, str | None]:
612        """Infer max supported audio quality from QQ track file metadata."""
613        file_obj = track_obj.get("file")
614        if not isinstance(file_obj, dict):
615            return (AudioFormat(content_type=ContentType.UNKNOWN), None)
616
617        size_new = file_obj.get("size_new")
618        size_new_list = size_new if isinstance(size_new, list) else []
619
620        def _size_new_at(index: int) -> int:
621            if index >= len(size_new_list):
622                return 0
623            return self._to_positive_int(size_new_list[index])
624
625        # QQMusicApi docs: size_new[0] is "master" (24bit/192kHz).
626        if _size_new_at(0):
627            return (
628                AudioFormat(content_type=ContentType.FLAC, sample_rate=192000, bit_depth=24),
629                "Hi-Res",
630            )
631        if self._file_size(file_obj, "size_flac", "sizeFlac") or _size_new_at(5):
632            return (
633                AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16),
634                None,
635            )
636        if self._file_size(file_obj, "size_320mp3", "size320mp3") or _size_new_at(3):
637            return (
638                AudioFormat(content_type=ContentType.MPEG, bit_rate=320000),
639                None,
640            )
641        if self._file_size(file_obj, "size_192ogg", "size192ogg"):
642            return (
643                AudioFormat(content_type=ContentType.OGG, bit_rate=192000),
644                None,
645            )
646        if self._file_size(file_obj, "size_192aac", "size192aac"):
647            return (
648                AudioFormat(content_type=ContentType.M4A, bit_rate=192000),
649                None,
650            )
651        if self._file_size(file_obj, "size_128mp3", "size128mp3"):
652            return (
653                AudioFormat(content_type=ContentType.MPEG, bit_rate=128000),
654                None,
655            )
656        if self._file_size(file_obj, "size_96ogg", "size96ogg"):
657            return (
658                AudioFormat(content_type=ContentType.OGG, bit_rate=96000),
659                None,
660            )
661        if self._file_size(file_obj, "size_96aac", "size96aac"):
662            return (
663                AudioFormat(content_type=ContentType.M4A, bit_rate=96000),
664                None,
665            )
666        if self._file_size(file_obj, "size_48aac", "size48aac"):
667            return (
668                AudioFormat(content_type=ContentType.M4A, bit_rate=48000),
669                None,
670            )
671        if self._file_size(file_obj, "size_try", "sizeTry"):
672            return (
673                AudioFormat(content_type=ContentType.MPEG),
674                None,
675            )
676        return (AudioFormat(content_type=ContentType.UNKNOWN), None)
677
678    def _get_stream_audio_format(self, selected_file_type: Any | None) -> AudioFormat:
679        """Build stream audio format for currently selected file type."""
680        if not selected_file_type:
681            return AudioFormat(content_type=ContentType.UNKNOWN)
682        if selected_file_type == SongFileType.FLAC:
683            return AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16)
684        if selected_file_type == SongFileType.MASTER:
685            return AudioFormat(content_type=ContentType.FLAC, sample_rate=192000, bit_depth=24)
686        if selected_file_type == SongFileType.MP3_320:
687            return AudioFormat(content_type=ContentType.MPEG, bit_rate=320000)
688        if selected_file_type in (SongFileType.MP3_128, SpecialSongFileType.TRY):
689            return AudioFormat(content_type=ContentType.MPEG, bit_rate=128000)
690        if selected_file_type == SongFileType.ACC_192:
691            return AudioFormat(content_type=ContentType.M4A, bit_rate=192000)
692        if selected_file_type == SongFileType.ACC_96:
693            return AudioFormat(content_type=ContentType.M4A, bit_rate=96000)
694        if selected_file_type == SongFileType.ACC_48:
695            return AudioFormat(content_type=ContentType.M4A, bit_rate=48000)
696        return AudioFormat(content_type=self._get_content_type(selected_file_type))
697
698    def _get_artist_mapping(self, artist_obj: dict[str, Any] | str) -> ItemMapping | None:
699        return get_artist_mapping(artist_obj, self.instance_id)
700
701    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
702        return parse_artist(artist_obj, self.domain, self.instance_id)
703
704    def _parse_album(self, album_obj: dict[str, Any]) -> Album:
705        return parse_album(album_obj, self.domain, self.instance_id)
706
707    def _parse_track(self, track_obj: dict[str, Any]) -> Track:
708        return parse_track(
709            track_obj=track_obj,
710            provider_domain=self.domain,
711            provider_instance_id=self.instance_id,
712            get_max_supported_audio_format=self._get_max_supported_audio_format,
713        )
714
715    async def _resolve_song_id(self, prov_track_id: str) -> int:
716        """Resolve provider track id (mid/id) to numeric song id."""
717        song_id, _song_type = await self._resolve_song_info(prov_track_id)
718        return song_id
719
720    async def _resolve_song_info(self, prov_track_id: str) -> tuple[int, int]:
721        """Resolve provider track id to numeric song id and QQ song type."""
722        if prov_track_id.isdigit():
723            return (int(prov_track_id), 0)
724        response = await self._run_with_session(self._qq_song.get_detail(prov_track_id))
725        response_obj = self._to_dict(response)
726        track_obj = response_obj.get("track_info") or response_obj.get("track") or {}
727        if song_id := extract_song_id(track_obj):
728            song_type = self._to_positive_int(track_obj.get("type") or track_obj.get("songtype"))
729            return (song_id, song_type)
730        raise MediaNotFoundError(f"Unable to resolve numeric song info for track {prov_track_id}")
731
732    # Compatibility wrappers for existing tests/extensions.
733    def _extract_song_id(self, track_obj: dict[str, Any]) -> int | None:
734        """Backward-compatible wrapper for song id extraction helper."""
735        return extract_song_id(track_obj)
736
737    def _extract_items(
738        self, data: dict[str, Any], candidate_keys: tuple[str, ...]
739    ) -> list[dict[str, Any]]:
740        """Backward-compatible wrapper for list extraction helper."""
741        return extract_items(data, candidate_keys)
742
743    async def _ensure_user_euin(self) -> str:
744        """Resolve and cache current user's encrypted uin."""
745        if self._euin:
746            return self._euin
747        euin = str(getattr(self._credential, "encrypt_uin", "") or "")
748        if not euin:
749            raise LoginFailed("Failed to resolve QQ Music user profile (euin)")
750        self._euin = euin
751        return self._euin
752
753    def _build_playlist_id(self, dissid: int | str, dirid: int | str) -> str:
754        return build_playlist_id(dissid, dirid)
755
756    def _parse_playlist_id(self, prov_playlist_id: str) -> tuple[int, int]:
757        return parse_playlist_id(prov_playlist_id)
758
759    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
760        return parse_playlist(playlist_obj, self.domain, self.instance_id)
761
762    @use_cache(3600 * 3)
763    async def search(
764        self,
765        search_query: str,
766        media_types: list[MediaType],
767        limit: int = 5,
768    ) -> SearchResults:
769        """Perform search on QQ Music."""
770        result = SearchResults()
771        if MediaType.TRACK in media_types:
772            raw_tracks = await self._run_with_session(
773                self._qq_search.search_by_type(
774                    search_query,
775                    SearchType.SONG,
776                    num=limit,
777                )
778            )
779            result.tracks = []
780            for item in self._response_items(raw_tracks, ("song", "songlist", "list")):
781                with suppress(InvalidDataError, TypeError, ValueError):
782                    result.tracks.append(self._parse_track(item))
783
784        if MediaType.ALBUM in media_types:
785            raw_albums = await self._run_with_session(
786                self._qq_search.search_by_type(
787                    search_query,
788                    SearchType.ALBUM,
789                    num=limit,
790                )
791            )
792            result.albums = []
793            for item in self._response_items(raw_albums, ("album", "album_list", "list")):
794                with suppress(InvalidDataError, TypeError, ValueError):
795                    result.albums.append(self._parse_album(item))
796
797        if MediaType.ARTIST in media_types:
798            raw_artists = await self._run_with_session(
799                self._qq_search.search_by_type(
800                    search_query,
801                    SearchType.SINGER,
802                    num=limit,
803                )
804            )
805            result.artists = []
806            for item in self._response_items(raw_artists, ("singer", "singer_list", "list")):
807                with suppress(InvalidDataError, TypeError, ValueError):
808                    result.artists.append(self._parse_artist(item))
809
810        if MediaType.PLAYLIST in media_types:
811            raw_playlists = await self._run_with_session(
812                self._qq_search.search_by_type(
813                    search_query,
814                    SearchType.SONGLIST,
815                    num=limit,
816                )
817            )
818            result.playlists = []
819            for item in self._response_items(raw_playlists, ("songlist", "playlists", "list")):
820                with suppress(InvalidDataError, TypeError, ValueError):
821                    result.playlists.append(self._parse_playlist(item))
822        return result
823
824    @use_cache(3600 * 24 * 7)
825    async def get_artist(self, prov_artist_id: str) -> Artist:
826        """Get full artist details by id."""
827        if prov_artist_id.isdigit():
828            raise MediaNotFoundError(
829                f"Artist id {prov_artist_id} is not a QQ singer mid, cannot fetch artist details"
830            )
831        response = await self._run_with_session(self._qq_singer.get_info(prov_artist_id))
832        artist_obj: dict[str, Any] | None = None
833        response_obj = self._to_dict(response)
834        info = response_obj.get("Info") or response_obj.get("info") or {}
835        if isinstance(info, dict):
836            singer_obj = info.get("Singer")
837            base_info = info.get("BaseInfo")
838            if isinstance(singer_obj, dict):
839                artist_obj = dict(singer_obj)
840            if isinstance(base_info, dict):
841                if artist_obj is None:
842                    artist_obj = dict(base_info)
843                else:
844                    if not extract_first_text(artist_obj, ("name", "Name", "singerName"), ""):
845                        artist_obj["Name"] = base_info.get("Name") or base_info.get("name")
846                    if not artist_obj.get("Avatar"):
847                        artist_obj["Avatar"] = base_info.get("Avatar") or base_info.get("avatar")
848            if artist_obj is None:
849                artist_obj = info
850        if artist_obj is None:
851            singer_list = response_obj.get("singer_list")
852            if isinstance(singer_list, list) and singer_list:
853                singer_item = singer_list[0]
854                if isinstance(singer_item, dict):
855                    artist_obj = singer_item.get("basic_info") or singer_item
856        if not artist_obj:
857            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
858        return self._parse_artist(artist_obj)
859
860    @use_cache(3600 * 12, allow_expired_cache=True)
861    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
862        """Get all albums for artist."""
863        if prov_artist_id.isdigit():
864            raise MediaNotFoundError(
865                f"Artist id {prov_artist_id} is not a QQ singer mid, cannot fetch albums"
866            )
867        raw_albums: list[dict[str, Any]] = []
868        try:
869            tab_albums = await self._run_with_session(
870                self._qq_singer.get_tab_detail(
871                    prov_artist_id,
872                    TabType.ALBUM,
873                    page=1,
874                    num=100,
875                )
876            )
877            raw_albums = self._response_items(
878                tab_albums,
879                ("album_tab", "albumList", "album_list", "list"),
880            )
881        except MediaNotFoundError, InvalidDataError, TypeError, ValueError:
882            raw_albums = []
883
884        if not raw_albums:
885            response = await self._run_with_session(
886                self._qq_singer.get_album_list(
887                    prov_artist_id,
888                    num=100,
889                    page=1,
890                )
891            )
892            raw_albums = self._response_items(
893                response,
894                ("albumList", "album_list", "list"),
895            )
896
897        albums: list[Album] = []
898        for item in raw_albums:
899            with suppress(InvalidDataError, TypeError, ValueError):
900                albums.append(self._parse_album(item))
901        return albums
902
903    async def _get_artist_song_list(self, prov_artist_id: str) -> list[Track]:
904        """Get parsed tracks from QQ Music singer song list."""
905        response = await self._run_with_session(
906            self._qq_singer.get_songs_list(
907                prov_artist_id,
908                num=100,
909                page=1,
910            )
911        )
912        response_obj = self._to_dict(response)
913        songs: list[dict[str, Any]] = []
914        for item in response_obj.get("songList", []):
915            if isinstance(item, dict) and isinstance(song_info := item.get("songInfo"), dict):
916                songs.append(song_info)
917        if not songs:
918            songs = extract_items(response_obj, ("song_list", "songs", "list"))
919        return [self._parse_track(item) for item in songs if item.get("mid")]
920
921    @use_cache(3600 * 6, allow_expired_cache=True)
922    async def get_artist_tracks(self, prov_artist_id: str) -> list[Track]:
923        """Get tracks for artist."""
924        if prov_artist_id.isdigit():
925            raise MediaNotFoundError(
926                f"Artist id {prov_artist_id} is not a QQ singer mid, cannot fetch tracks"
927            )
928        return await self._get_artist_song_list(prov_artist_id)
929
930    @use_cache(3600 * 6, allow_expired_cache=True)
931    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
932        """Get top tracks for artist."""
933        if prov_artist_id.isdigit():
934            raise MediaNotFoundError(
935                f"Artist id {prov_artist_id} is not a QQ singer mid, cannot fetch top tracks"
936            )
937        return await self._get_artist_song_list(prov_artist_id)
938
939    @use_cache(3600 * 24 * 7)
940    async def get_album(self, prov_album_id: str) -> Album:
941        """Get full album details by id."""
942        album_value: str | int = int(prov_album_id) if prov_album_id.isdigit() else prov_album_id
943        response = await self._run_with_session(self._qq_album.get_detail(album_value))
944        if not response:
945            raise MediaNotFoundError(f"Album {prov_album_id} not found")
946        album_obj: dict[str, Any] | None = None
947        response_obj = self._to_dict(response)
948        basic_info = response_obj.get("basicInfo") or response_obj.get("album")
949        if isinstance(basic_info, dict):
950            album_obj = dict(basic_info)
951            if "singer" not in album_obj:
952                singer_list = response_obj.get("singer", {}).get("singerList")
953                if not isinstance(singer_list, list):
954                    singer_list = response_obj.get("singers")
955                if isinstance(singer_list, list):
956                    album_obj["singer"] = singer_list
957        else:
958            album_obj = response_obj
959        if not isinstance(album_obj, dict):
960            raise MediaNotFoundError(f"Album {prov_album_id} returned unexpected payload")
961        return self._parse_album(album_obj)
962
963    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
964    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
965        """Get album tracks for album id."""
966        album_value: str | int = int(prov_album_id) if prov_album_id.isdigit() else prov_album_id
967        response = await self._run_with_session(
968            self._qq_album.get_song(album_value, num=300, page=1)
969        )
970        response_obj = self._to_dict(response)
971        songs: list[dict[str, Any]] = []
972        for item in response_obj.get("songList", []):
973            if isinstance(item, dict) and isinstance(song_info := item.get("songInfo"), dict):
974                songs.append(song_info)
975        if not songs:
976            songs = self._response_items(response, ("song_list", "songs", "list"))
977        return [self._parse_track(item) for item in songs if item.get("mid")]
978
979    @use_cache(3600 * 24 * 7, cache_checksum="qqmusic_lyrics_v2")
980    async def get_track(self, prov_track_id: str) -> Track:
981        """Get full track details by id."""
982        track_value: str | int = int(prov_track_id) if prov_track_id.isdigit() else prov_track_id
983        response = await self._run_with_session(self._qq_song.get_detail(track_value))
984        response_obj = self._to_dict(response)
985        track_obj = response_obj.get("track_info") or response_obj.get("track")
986        if not track_obj:
987            raise MediaNotFoundError(f"Track {prov_track_id} not found")
988        track = self._parse_track(track_obj)
989        try:
990            # Prefer normal lyric first: this is typically LRC and works best for MA synced scroll.
991            lyric_response = await self._run_with_session(
992                self._qq_lyric.get_lyric(prov_track_id, qrc=False, trans=True)
993            )
994            lyric_text = ""
995            trans_text = ""
996            lyric_obj = self._decode_lyric_response(lyric_response)
997            lyric_text = str(lyric_obj.get("lyric") or "").strip()
998            trans_text = str(lyric_obj.get("trans") or "").strip()
999            # Fallback to QRC when standard lyric is empty/unavailable.
1000            if not lyric_text:
1001                lyric_response = await self._run_with_session(
1002                    self._qq_lyric.get_lyric(prov_track_id, qrc=True, trans=True)
1003                )
1004                lyric_obj = self._decode_lyric_response(lyric_response)
1005            lyric_text = str(lyric_obj.get("lyric") or lyric_text).strip()
1006            trans_text = str(lyric_obj.get("trans") or trans_text).strip()
1007            if lyric_text:
1008                if _LRC_TIMESTAMP_PATTERN.search(lyric_text):
1009                    track.metadata.lrc_lyrics = normalize_qq_lyric_text(lyric_text)
1010                else:
1011                    # QRC (e.g. [36438,1880]当(36438,161)...) -> LRC for synced display.
1012                    qrc_lrc = qrc_to_lrc(lyric_text)
1013                    if qrc_lrc:
1014                        track.metadata.lrc_lyrics = qrc_lrc
1015                track.metadata.lyrics = normalize_qq_lyric_text(lyric_text)
1016            if trans_text:
1017                trans_text = normalize_qq_lyric_text(trans_text)
1018                if track.metadata.lyrics:
1019                    track.metadata.lyrics = f"{track.metadata.lyrics}\n\n{trans_text}".strip()
1020                else:
1021                    track.metadata.lyrics = trans_text
1022        except Exception as err:
1023            self.logger.debug("Failed to load QQ Music lyrics for %s: %s", prov_track_id, err)
1024        return track
1025
1026    async def get_library_artists(self) -> AsyncGenerator[Artist]:
1027        """Retrieve followed artists from QQ Music."""
1028        euin = await self._ensure_user_euin()
1029        page = 1
1030        num = 100
1031        total_yielded = 0
1032        while True:
1033            response = await self._run_with_session(
1034                self._qq_user.get_follow_singers(
1035                    euin,
1036                    page=page,
1037                    num=num,
1038                    credential=self._credential,
1039                )
1040            )
1041            artists = self._response_items(
1042                response,
1043                ("List", "Users", "list", "v_list", "users", "singer_list", "singers"),
1044            )
1045            if not artists:
1046                break
1047            for artist_obj in artists:
1048                try:
1049                    yield self._parse_artist(artist_obj)
1050                    total_yielded += 1
1051                except InvalidDataError, TypeError, ValueError:
1052                    continue
1053            if len(artists) < num:
1054                break
1055            page += 1
1056        self.logger.info("QQ library artists sync yielded %s artist(s)", total_yielded)
1057
1058    async def get_library_tracks(self) -> AsyncGenerator[Track]:
1059        """Retrieve library tracks from QQ Music."""
1060        euin = await self._ensure_user_euin()
1061        page = 1
1062        num = 100
1063        yielded = 0
1064        total = None
1065        while True:
1066            response = await self._run_with_session(
1067                self._qq_user.get_fav_song(euin, page=page, num=num, credential=self._credential)
1068            )
1069            response_obj = self._to_dict(response)
1070            songs = self._response_items(response, ("songlist", "song_list", "songs", "list"))
1071            if total is None:
1072                total = int(response_obj.get("total_song_num") or response_obj.get("total") or 0)
1073            if not songs:
1074                break
1075            for song in songs:
1076                try:
1077                    yield self._parse_track(song)
1078                    yielded += 1
1079                except InvalidDataError, TypeError, ValueError:
1080                    continue
1081            if total and yielded >= total:
1082                break
1083            page += 1
1084
1085    async def get_library_albums(self) -> AsyncGenerator[Album]:
1086        """Retrieve library albums from QQ Music."""
1087        euin = await self._ensure_user_euin()
1088        page = 1
1089        num = 100
1090        total_yielded = 0
1091        while True:
1092            response = await self._run_with_session(
1093                self._qq_user.get_fav_album(euin, page=page, num=num, credential=self._credential)
1094            )
1095            albums = self._response_items(
1096                response,
1097                (
1098                    "albums",
1099                    "album_list",
1100                    "albumList",
1101                    "v_list",
1102                    "list",
1103                    "v_album",
1104                    "favAlbumList",
1105                ),
1106            )
1107            if not albums:
1108                break
1109            for album_obj in albums:
1110                try:
1111                    yield self._parse_album(album_obj)
1112                    total_yielded += 1
1113                except InvalidDataError, TypeError, ValueError:
1114                    continue
1115            if len(albums) < num:
1116                break
1117            page += 1
1118        self.logger.info("QQ library albums sync yielded %s album(s)", total_yielded)
1119
1120    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
1121        """Retrieve user playlists from QQ Music."""
1122        euin = await self._ensure_user_euin()
1123        created = await self._run_with_session(
1124            self._qq_user.get_created_songlist(self._musicid, credential=self._credential)
1125        )
1126        for playlist_obj in self._response_items(
1127            created,
1128            ("playlists", "v_playlist", "list", "playlist"),
1129        ):
1130            try:
1131                yield self._parse_playlist(playlist_obj)
1132            except InvalidDataError, TypeError, ValueError:
1133                continue
1134
1135        page = 1
1136        num = 100
1137        while True:
1138            response = await self._run_with_session(
1139                self._qq_user.get_fav_songlist(
1140                    euin, page=page, num=num, credential=self._credential
1141                )
1142            )
1143            fav_playlists = self._response_items(
1144                response,
1145                ("playlists", "list", "v_list", "playlist", "vec_kept_playlist"),
1146            )
1147            if not fav_playlists:
1148                break
1149            for playlist_obj in fav_playlists:
1150                try:
1151                    yield self._parse_playlist(playlist_obj)
1152                except InvalidDataError, TypeError, ValueError:
1153                    continue
1154            if len(fav_playlists) < num:
1155                break
1156            page += 1
1157
1158    @use_cache(3600 * 3)
1159    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
1160        """Get full playlist details by id."""
1161        dissid, dirid = self._parse_playlist_id(prov_playlist_id)
1162        response = await self._run_with_session(
1163            self._qq_songlist.get_detail(
1164                songlist_id=dissid,
1165                dirid=dirid,
1166                num=1,
1167                page=1,
1168                onlysong=False,
1169            )
1170        )
1171        response_obj = self._to_dict(response)
1172        playlist_obj = response_obj.get("dirinfo") or response_obj.get("info")
1173        if not isinstance(playlist_obj, dict):
1174            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
1175        # Ensure parsed playlist keeps composite id including dirid.
1176        playlist_obj = {**playlist_obj, "dissid": dissid, "dirid": dirid}
1177        return self._parse_playlist(playlist_obj)
1178
1179    @use_cache(3600, allow_expired_cache=True)
1180    async def get_playlist_tracks(
1181        self,
1182        prov_playlist_id: str,
1183        page: int = 0,
1184    ) -> list[Track]:
1185        """Get playlist tracks for given playlist id."""
1186        dissid, dirid = self._parse_playlist_id(prov_playlist_id)
1187        response = await self._run_with_session(
1188            self._qq_songlist.get_detail(
1189                songlist_id=dissid,
1190                dirid=dirid,
1191                num=200,
1192                page=page + 1,
1193                onlysong=True,
1194            )
1195        )
1196        songs = self._response_items(response, ("songlist", "songs", "song_list", "list"))
1197        results: list[Track] = []
1198        for index, song in enumerate(songs, start=1 + page * 200):
1199            try:
1200                track = self._parse_track(song)
1201                track.position = index
1202                results.append(track)
1203            except InvalidDataError, TypeError, ValueError:
1204                continue
1205        return results
1206
1207    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
1208        """Create a new playlist on provider with given name."""
1209        created = await self._run_with_session(
1210            self._qq_songlist.create(dirname=name, credential=self._credential)
1211        )
1212        created_obj = self._to_dict(created)
1213        if not created_obj:
1214            raise InvalidDataError("QQ Music create playlist returned invalid response")
1215        dirid_raw = created_obj.get("dirid") or created_obj.get("dirId") or created_obj.get("id")
1216        if dirid_raw is None:
1217            raise InvalidDataError("QQ Music create playlist response missing dirid")
1218        dirid = int(dirid_raw)
1219        dissid = int(created_obj.get("tid") or created_obj.get("dissid") or dirid)
1220        return await self.get_playlist(self._build_playlist_id(dissid, dirid))
1221
1222    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
1223        """Add track(s) to playlist."""
1224        dissid, dirid = self._parse_playlist_id(prov_playlist_id)
1225        target_dirid = dirid or dissid
1226        if target_dirid <= 0:
1227            raise InvalidDataError("QQ Music playlist id is invalid for playlist edit")
1228        song_info: list[tuple[int, int]] = []
1229        for track_id in prov_track_ids:
1230            try:
1231                song_info.append(await self._resolve_song_info(track_id))
1232            except (MediaNotFoundError, InvalidDataError, ResourceTemporarilyUnavailable) as err:
1233                self.logger.warning("Skipping track %s while adding to playlist: %s", track_id, err)
1234        if not song_info:
1235            raise InvalidDataError("No valid QQ Music tracks to add")
1236        await self._run_with_session(
1237            self._qq_songlist.add_songs(
1238                dirid=target_dirid,
1239                song_info=song_info,
1240                tid=dissid,
1241                credential=self._credential,
1242            )
1243        )
1244
1245    async def remove_playlist_tracks(
1246        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
1247    ) -> None:
1248        """Remove track(s) from playlist."""
1249        dissid, dirid = self._parse_playlist_id(prov_playlist_id)
1250        target_dirid = dirid or dissid
1251        if target_dirid <= 0:
1252            raise InvalidDataError("QQ Music playlist id is invalid for playlist edit")
1253        playlist_tracks = await self.get_playlist_tracks(prov_playlist_id, page=0)
1254        song_info: list[tuple[int, int]] = []
1255        target_positions = set(positions_to_remove)
1256        for track in playlist_tracks:
1257            if track.position not in target_positions:
1258                continue
1259            try:
1260                song_info.append(await self._resolve_song_info(track.item_id))
1261            except (MediaNotFoundError, InvalidDataError, ResourceTemporarilyUnavailable) as err:
1262                self.logger.warning(
1263                    "Skipping track %s while removing from playlist: %s", track.item_id, err
1264                )
1265        if not song_info:
1266            return
1267        await self._run_with_session(
1268            self._qq_songlist.del_songs(
1269                dirid=target_dirid,
1270                song_info=song_info,
1271                tid=dissid,
1272                credential=self._credential,
1273            )
1274        )
1275
1276    @use_cache(3600 * 24, allow_expired_cache=True)
1277    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
1278        """Retrieve a dynamic list of similar artists based on the provided artist."""
1279        if prov_artist_id.isdigit():
1280            raise MediaNotFoundError(
1281                f"Artist id {prov_artist_id} is not a QQ singer mid, cannot fetch similar artists"
1282            )
1283        response = await self._run_with_session(
1284            self._qq_singer.get_similar(prov_artist_id, number=limit)
1285        )
1286        artists: list[Artist] = []
1287        for item in self._response_items(
1288            response, ("singerlist", "singer_list", "singers", "list")
1289        ):
1290            if len(artists) >= limit:
1291                break
1292            with suppress(InvalidDataError, TypeError, ValueError):
1293                artists.append(self._parse_artist(item))
1294        return artists
1295
1296    @use_cache(3600 * 24, allow_expired_cache=True)
1297    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
1298        """Retrieve a dynamic list of similar tracks based on the provided track."""
1299        song_id = await self._resolve_song_id(prov_track_id)
1300        response = await self._run_with_session(self._qq_song.get_similar_song(song_id))
1301        response_obj = self._to_dict(response)
1302        response_items: Any = response if isinstance(response, list) else []
1303        if not response_items:
1304            response_items = response_obj.get("song") or response_obj.get("songlist") or []
1305        if not isinstance(response_items, list):
1306            return []
1307        tracks: list[Track] = []
1308        for item in response_items:
1309            if len(tracks) >= limit:
1310                break
1311            if not isinstance(item, dict):
1312                continue
1313            grouped_songs = item.get("song")
1314            candidates = grouped_songs if isinstance(grouped_songs, list) else [item]
1315            for candidate in candidates:
1316                if len(tracks) >= limit:
1317                    break
1318                if not isinstance(candidate, dict):
1319                    continue
1320                with suppress(InvalidDataError, TypeError, ValueError):
1321                    tracks.append(self._parse_track(candidate))
1322        return tracks
1323
1324    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
1325        """Return streamdetails for given track id."""
1326        if media_type != MediaType.TRACK:
1327            raise MediaNotFoundError(f"Unsupported media type {media_type}")
1328        track_response = await self._run_with_session(self._qq_song.get_detail(item_id))
1329        track_response_obj = self._to_dict(track_response)
1330        track_obj = track_response_obj.get("track_info") or track_response_obj.get("track") or {}
1331        if not track_obj:
1332            raise MediaNotFoundError(f"Track {item_id} not found")
1333        (
1334            stream_url,
1335            selected_file_type,
1336            is_preview_stream,
1337            preview_duration,
1338        ) = await self._resolve_stream_url(item_id, track_obj)
1339
1340        if not stream_url:
1341            pay_info = track_obj.get("pay", {})
1342            pay_play = pay_info.get("pay_play", "unknown")
1343            pay_status = pay_info.get("pay_status", "unknown")
1344            raise UnplayableMediaError(
1345                f"No playable stream URL returned for track {item_id} "
1346                f"(pay_play={pay_play}, pay_status={pay_status})"
1347            )
1348
1349        expiration = self._get_stream_expiration(stream_url)
1350        return StreamDetails(
1351            provider=self.instance_id,
1352            item_id=item_id,
1353            audio_format=self._get_stream_audio_format(selected_file_type),
1354            stream_type=StreamType.HTTP,
1355            path=stream_url,
1356            duration=preview_duration if is_preview_stream else None,
1357            data={"preview": is_preview_stream},
1358            can_seek=True,
1359            allow_seek=True,
1360            expiration=expiration,
1361        )
1362