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