music-assistant-server

1"""Response parsing helpers for QQ Music provider."""
2
3from __future__ import annotations
4
5from collections.abc import Callable
6from contextlib import suppress
7from datetime import datetime
8from typing import Any
9
10from music_assistant_models.enums import ImageType, MediaType
11from music_assistant_models.errors import InvalidDataError
12from music_assistant_models.media_items import (
13    Album,
14    Artist,
15    AudioFormat,
16    ItemMapping,
17    MediaItemImage,
18    Playlist,
19    ProviderMapping,
20    Track,
21    UniqueList,
22)
23
24from music_assistant.helpers.util import parse_title_and_version
25
26from .helpers import clean_text, extract_artist_mid, extract_first_text, normalize_image_url
27
28
29def extract_song_id(track_obj: dict[str, Any]) -> int | None:
30    """Extract numeric QQ song id from a track-like payload."""
31    for key in ("id", "songid", "song_id", "songID"):
32        raw_val = track_obj.get(key)
33        if raw_val is None:
34            continue
35        try:
36            song_id = int(raw_val)
37        except TypeError, ValueError:
38            continue
39        if song_id > 0:
40            return song_id
41    return None
42
43
44def extract_items(data: dict[str, Any], candidate_keys: tuple[str, ...]) -> list[dict[str, Any]]:
45    """Extract list-like payload from possible response keys."""
46    for key in candidate_keys:
47        val = data.get(key)
48        if isinstance(val, list):
49            return [item for item in val if isinstance(item, dict)]
50        if isinstance(val, dict):
51            for nested_key in ("list", "v_playlist", "album_list", "songlist", "items"):
52                nested_val = val.get(nested_key)
53                if isinstance(nested_val, list):
54                    return [item for item in nested_val if isinstance(item, dict)]
55    return []
56
57
58def extract_guess_recommend_tracks(data: Any) -> list[dict[str, Any]]:
59    """Extract tracks from GuessRecommendResponse raw payload."""
60    if not isinstance(data, dict):
61        return []
62    tracks = data.get("songs") or data.get("Tracks")
63    if not isinstance(tracks, list):
64        return []
65    return [item for item in tracks if isinstance(item, dict)]
66
67
68def extract_radar_recommend_tracks(data: Any) -> list[dict[str, Any]]:
69    """Extract tracks from RadarRecommendResponse raw payload."""
70    if not isinstance(data, dict):
71        return []
72    songs = data.get("songs")
73    if isinstance(songs, list):
74        return [item for item in songs if isinstance(item, dict)]
75    vec_songs = data.get("VecSongs")
76    if not isinstance(vec_songs, list):
77        return []
78    result: list[dict[str, Any]] = []
79    for item in vec_songs:
80        if not isinstance(item, dict):
81            continue
82        track = item.get("Track")
83        if isinstance(track, dict):
84            result.append(track)
85    return result
86
87
88def extract_newsong_tracks(data: Any) -> list[dict[str, Any]]:
89    """Extract tracks from RecommendNewSongResponse raw payload."""
90    if not isinstance(data, dict):
91        return []
92    songs = data.get("songs") or data.get("songlist")
93    if not isinstance(songs, list):
94        return []
95    return [item for item in songs if isinstance(item, dict)]
96
97
98def extract_recommend_songlists(data: Any) -> list[dict[str, Any]]:
99    """Extract playlist basics from RecommendSonglistResponse raw payload."""
100    if not isinstance(data, dict):
101        return []
102    songlists = data.get("songlists")
103    if isinstance(songlists, list):
104        return [item for item in songlists if isinstance(item, dict)]
105    items = data.get("List")
106    if not isinstance(items, list):
107        return []
108    result: list[dict[str, Any]] = []
109    for item in items:
110        if not isinstance(item, dict):
111            continue
112        playlist_obj = item.get("Playlist")
113        if not isinstance(playlist_obj, dict):
114            continue
115        basic = playlist_obj.get("basic")
116        if not isinstance(basic, dict):
117            continue
118        merged = dict(basic)
119        cover_obj = playlist_obj.get("cover")
120        if isinstance(cover_obj, dict):
121            cover_url = normalize_image_url(
122                cover_obj.get("default_url")
123                or cover_obj.get("medium_url")
124                or cover_obj.get("big_url")
125                or cover_obj.get("small_url")
126            )
127            if cover_url and not merged.get("picurl"):
128                merged["picurl"] = cover_url
129        creator_obj = playlist_obj.get("creator")
130        if isinstance(creator_obj, dict) and not merged.get("creator"):
131            merged["creator"] = creator_obj
132        if playlist_obj.get("desc") and not merged.get("desc"):
133            merged["desc"] = playlist_obj.get("desc")
134        result.append(merged)
135    return result
136
137
138def get_artist_mapping(
139    artist_obj: dict[str, Any] | str, provider_instance_id: str
140) -> ItemMapping | None:
141    """Build artist item mapping from QQ artist payload."""
142    if isinstance(artist_obj, str):
143        return None
144    artist_id = extract_artist_mid(artist_obj)
145    if not artist_id:
146        return None
147    artist_name = extract_first_text(
148        artist_obj,
149        (
150            "name",
151            "title",
152            "singerName",
153            "singername",
154            "singer_name",
155            "title_main",
156            "search_title",
157            "singerName_hilight",
158            "name_hilight",
159            "title_hilight",
160        ),
161        "Unknown Artist",
162    )
163    return ItemMapping(
164        media_type=MediaType.ARTIST,
165        item_id=artist_id,
166        provider=provider_instance_id,
167        name=artist_name,
168    )
169
170
171def parse_artist(
172    artist_obj: dict[str, Any], provider_domain: str, provider_instance_id: str
173) -> Artist:
174    """Parse raw qqmusic artist object."""
175    artist_id = extract_artist_mid(artist_obj)
176    if not artist_id:
177        raise InvalidDataError("Artist object does not contain id/mid")
178    artist_name = extract_first_text(
179        artist_obj,
180        (
181            "name",
182            "Name",
183            "title",
184            "singerName",
185            "singername",
186            "singer_name",
187            "SingerName",
188            "title_main",
189            "search_title",
190            "singerName_hilight",
191            "name_hilight",
192            "title_hilight",
193            "singer_hilight",
194            "matchName",
195            "match_name",
196            "nickname",
197            "nick",
198        ),
199        "Unknown Artist",
200    )
201    artist = Artist(
202        item_id=artist_id,
203        provider=provider_instance_id,
204        name=artist_name,
205        provider_mappings={
206            ProviderMapping(
207                item_id=artist_id,
208                provider_domain=provider_domain,
209                provider_instance=provider_instance_id,
210                url=f"https://y.qq.com/n/ryqq/singer/{artist_id}",
211            )
212        },
213    )
214    subtitle = clean_text(
215        artist_obj.get("subtitle") or artist_obj.get("desc") or artist_obj.get("description") or "",
216        "",
217    )
218    if subtitle:
219        artist.metadata.description = subtitle
220        artist.metadata.description_language = "zh"
221    artist_image = (
222        normalize_image_url(
223            artist_obj.get("singerPic")
224            or artist_obj.get("pic")
225            or artist_obj.get("avatar")
226            or artist_obj.get("Avatar")
227            or artist_obj.get("AvatarUrl")
228            or artist_obj.get("singer_pic")
229        )
230        or f"https://y.qq.com/music/photo_new/T001R500x500M000{artist_id}.jpg"
231    )
232    artist.metadata.images = UniqueList(
233        [
234            MediaItemImage(
235                type=ImageType.THUMB,
236                path=artist_image,
237                provider=provider_instance_id,
238                remotely_accessible=True,
239            )
240        ]
241    )
242    return artist
243
244
245def parse_album(
246    album_obj: dict[str, Any],
247    provider_domain: str,
248    provider_instance_id: str,
249) -> Album:
250    """Parse raw qqmusic album object."""
251    album_id = str(
252        album_obj.get("mid")
253        or album_obj.get("albumMid")
254        or album_obj.get("albumMID")
255        or album_obj.get("album_mid")
256        or album_obj.get("albummid")
257        or album_obj.get("id")
258        or album_obj.get("albumID")
259        or ""
260    )
261    if not album_id:
262        raise InvalidDataError("Album object does not contain id/mid")
263    raw_album_name = str(
264        album_obj.get("title")
265        or album_obj.get("name")
266        or album_obj.get("albumName")
267        or "Unknown Album"
268    )
269    album_subtitle = clean_text(
270        album_obj.get("subtitle")
271        or album_obj.get("albumTranName")
272        or album_obj.get("title_extra")
273        or "",
274        "",
275    )
276    album_name, album_version = parse_title_and_version(raw_album_name, album_subtitle)
277    album_name = clean_text(album_name, "Unknown Album")
278    album = Album(
279        item_id=album_id,
280        provider=provider_instance_id,
281        name=album_name,
282        version=album_version,
283        provider_mappings={
284            ProviderMapping(
285                item_id=album_id,
286                provider_domain=provider_domain,
287                provider_instance=provider_instance_id,
288                url=f"https://y.qq.com/n/ryqq/albumDetail/{album_id}",
289            )
290        },
291    )
292    if release_str := album_obj.get("time_public") or album_obj.get("publishDate"):
293        with suppress(ValueError):
294            album.year = datetime.strptime(release_str, "%Y-%m-%d").year  # noqa: DTZ007
295    album_desc = clean_text(
296        album_obj.get("description")
297        or album_obj.get("description2")
298        or album_obj.get("desc")
299        or "",
300        "",
301    )
302    if album_desc:
303        album.metadata.description = album_desc
304        album.metadata.description_language = "zh"
305    artist_candidates = album_obj.get("singer")
306    if isinstance(artist_candidates, dict):
307        artist_iterable: list[dict[str, Any] | str] = [artist_candidates]
308    elif isinstance(artist_candidates, list):
309        artist_iterable = artist_candidates
310    elif isinstance(artist_candidates, str):
311        artist_iterable = [artist_candidates]
312    else:
313        singer_list = album_obj.get("singer_list")
314        artist_iterable = singer_list if isinstance(singer_list, list) else []
315    for artist_obj in artist_iterable:
316        if artist_mapping := get_artist_mapping(artist_obj, provider_instance_id):
317            album.artists.append(artist_mapping)
318    cover_path = normalize_image_url(
319        album_obj.get("pic")
320        or album_obj.get("picUrl")
321        or album_obj.get("picurl")
322        or album_obj.get("logo")
323        or album_obj.get("cover")
324    )
325    if cover_path:
326        album.metadata.images = UniqueList(
327            [
328                MediaItemImage(
329                    type=ImageType.THUMB,
330                    path=cover_path,
331                    provider=provider_instance_id,
332                    remotely_accessible=True,
333                )
334            ]
335        )
336    else:
337        album.metadata.images = UniqueList(
338            [
339                MediaItemImage(
340                    type=ImageType.THUMB,
341                    path=f"https://y.qq.com/music/photo_new/T002R500x500M000{album_id}.jpg",
342                    provider=provider_instance_id,
343                    remotely_accessible=True,
344                )
345            ]
346        )
347    return album
348
349
350def parse_track(  # noqa: PLR0915
351    track_obj: dict[str, Any],
352    provider_domain: str,
353    provider_instance_id: str,
354    get_max_supported_audio_format: Callable[[dict[str, Any]], tuple[AudioFormat, str | None]],
355) -> Track:
356    """Parse raw qqmusic track object."""
357    track_id = str(
358        track_obj.get("mid")
359        or track_obj.get("songMid")
360        or track_obj.get("songmid")
361        or track_obj.get("id")
362        or ""
363    )
364    if not track_id:
365        raise InvalidDataError("Track object does not contain id/mid")
366    raw_track_name = clean_text(
367        track_obj.get("title") or track_obj.get("name"),
368        "Unknown Track",
369    )
370    track_subtitle = clean_text(
371        track_obj.get("subtitle") or track_obj.get("title_extra") or "",
372        "",
373    )
374    track_name, track_version = parse_title_and_version(raw_track_name, track_subtitle)
375    duration = track_obj.get("interval")
376    max_audio_format, max_quality_label = get_max_supported_audio_format(track_obj)
377    track = Track(
378        item_id=track_id,
379        provider=provider_instance_id,
380        name=track_name,
381        version=track_version,
382        duration=int(duration) if duration else 0,
383        provider_mappings={
384            ProviderMapping(
385                item_id=track_id,
386                provider_domain=provider_domain,
387                provider_instance=provider_instance_id,
388                audio_format=max_audio_format,
389                url=f"https://y.qq.com/n/ryqq/songDetail/{track_id}",
390                details=max_quality_label,
391            )
392        },
393    )
394    track_desc = clean_text(
395        track_obj.get("desc") or track_obj.get("content") or "",
396        "",
397    )
398    if track_desc:
399        track.metadata.description = track_desc
400        track.metadata.description_language = "zh"
401    album_obj = track_obj.get("album")
402    album_id = ""
403    album_name = ""
404    if isinstance(album_obj, dict):
405        album_id = str(
406            album_obj.get("mid")
407            or album_obj.get("albumMid")
408            or album_obj.get("albumMID")
409            or album_obj.get("albummid")
410            or album_obj.get("albumID")
411            or album_obj.get("albumid")
412            or album_obj.get("id")
413            or ""
414        )
415        album_name = clean_text(
416            album_obj.get("name")
417            or album_obj.get("title")
418            or album_obj.get("albumName")
419            or album_obj.get("albumTitle")
420        )
421    if not album_id:
422        album_id = str(
423            track_obj.get("albummid")
424            or track_obj.get("albumMid")
425            or track_obj.get("albumMID")
426            or track_obj.get("albumid")
427            or track_obj.get("albumID")
428            or ""
429        )
430    if not album_name:
431        album_name = clean_text(
432            track_obj.get("albumname")
433            or track_obj.get("albumName")
434            or track_obj.get("albumTitle")
435            or track_obj.get("album_title")
436        )
437    if album_id and album_name:
438        track.album = Album(
439            item_id=album_id,
440            provider=provider_instance_id,
441            name=album_name,
442            provider_mappings={
443                ProviderMapping(
444                    item_id=album_id,
445                    provider_domain=provider_domain,
446                    provider_instance=provider_instance_id,
447                    url=f"https://y.qq.com/n/ryqq/albumDetail/{album_id}",
448                )
449            },
450        )
451
452    singer_list: list[dict[str, Any] | str]
453    raw_singer = track_obj.get("singer") or track_obj.get("singer_list") or track_obj.get("singers")
454    if isinstance(raw_singer, dict):
455        singer_list = [raw_singer]
456    elif isinstance(raw_singer, list):
457        singer_list = raw_singer
458    elif isinstance(raw_singer, str):
459        singer_list = [raw_singer]
460    else:
461        singer_list = []
462    for artist_obj in singer_list:
463        if artist_mapping := get_artist_mapping(artist_obj, provider_instance_id):
464            track.artists.append(artist_mapping)
465    if not track.artists:
466        fallback_artist_id = str(
467            track_obj.get("singerMid")
468            or track_obj.get("singerMID")
469            or track_obj.get("singer_mid")
470            or ""
471        )
472        fallback_artist_name = clean_text(
473            track_obj.get("singerName")
474            or track_obj.get("singername")
475            or track_obj.get("singer_name")
476            or ""
477        )
478        if fallback_artist_id and fallback_artist_name:
479            track.artists.append(
480                ItemMapping(
481                    media_type=MediaType.ARTIST,
482                    item_id=fallback_artist_id,
483                    provider=provider_instance_id,
484                    name=fallback_artist_name,
485                )
486            )
487
488    cover_id = ""
489    if isinstance(album_obj, dict):
490        cover_id = str(
491            album_obj.get("mid")
492            or album_obj.get("albumMid")
493            or album_obj.get("albumMID")
494            or album_obj.get("albummid")
495            or ""
496        )
497    if not cover_id:
498        cover_id = album_id
499    if cover_id:
500        if isinstance(track.album, Album):
501            track.album.metadata.images = UniqueList(
502                [
503                    MediaItemImage(
504                        type=ImageType.THUMB,
505                        path=f"https://y.qq.com/music/photo_new/T002R500x500M000{cover_id}.jpg",
506                        provider=provider_instance_id,
507                        remotely_accessible=True,
508                    )
509                ]
510            )
511        track.metadata.images = UniqueList(
512            [
513                MediaItemImage(
514                    type=ImageType.THUMB,
515                    path=f"https://y.qq.com/music/photo_new/T002R500x500M000{cover_id}.jpg",
516                    provider=provider_instance_id,
517                    remotely_accessible=True,
518                )
519            ]
520        )
521    return track
522
523
524def build_playlist_id(dissid: int | str, dirid: int | str) -> str:
525    """Build provider playlist id from dissid and dirid."""
526    return f"{dissid}:{dirid}"
527
528
529def parse_playlist_id(prov_playlist_id: str) -> tuple[int, int]:
530    """Parse provider playlist id into (dissid, dirid)."""
531    if ":" in prov_playlist_id:
532        dissid_raw, dirid_raw = prov_playlist_id.split(":", 1)
533        return (int(dissid_raw), int(dirid_raw))
534    return (int(prov_playlist_id), 0)
535
536
537def parse_playlist(
538    playlist_obj: dict[str, Any],
539    provider_domain: str,
540    provider_instance_id: str,
541) -> Playlist:
542    """Parse raw qqmusic playlist object."""
543    dissid = playlist_obj.get("tid") or playlist_obj.get("dissid") or playlist_obj.get("id") or 0
544    dirid = playlist_obj.get("dirid") or playlist_obj.get("dirId") or 0
545    if not dissid:
546        raise InvalidDataError("Playlist object missing dissid/tid")
547    playlist_id = build_playlist_id(dissid, dirid)
548    playlist_name = clean_text(
549        playlist_obj.get("dirName")
550        or playlist_obj.get("dirname")
551        or playlist_obj.get("name")
552        or playlist_obj.get("title")
553        or playlist_obj.get("search_title")
554        or playlist_obj.get("name_hilight")
555        or playlist_obj.get("title_hilight")
556        or playlist_obj.get("diss_name")
557        or playlist_obj.get("dissname")
558        or "QQ Music Playlist",
559        "QQ Music Playlist",
560    )
561    playlist = Playlist(
562        item_id=playlist_id,
563        provider=provider_instance_id,
564        name=playlist_name,
565        provider_mappings={
566            ProviderMapping(
567                item_id=playlist_id,
568                provider_domain=provider_domain,
569                provider_instance=provider_instance_id,
570                url=f"https://y.qq.com/n/ryqq/playlist/{dissid}",
571            )
572        },
573    )
574    owner_name = str(
575        playlist_obj.get("creator", {}).get("name")
576        or playlist_obj.get("creator", {}).get("nick")
577        or playlist_obj.get("nickname")
578        or playlist_obj.get("nick")
579        or ""
580    )
581    if owner_name:
582        playlist.owner = owner_name
583    description = clean_text(
584        playlist_obj.get("desc") or playlist_obj.get("description"),
585        "",
586    )
587    if description:
588        playlist.metadata.description = description
589        playlist.metadata.description_language = "zh"
590
591    cover_val = playlist_obj.get("cover")
592    cover = ""
593    if isinstance(cover_val, dict):
594        cover = normalize_image_url(
595            cover_val.get("default_url")
596            or cover_val.get("medium_url")
597            or cover_val.get("big_url")
598            or cover_val.get("small_url")
599        )
600    if not cover:
601        cover = normalize_image_url(
602            cover_val
603            or playlist_obj.get("logo")
604            or playlist_obj.get("picurl")
605            or playlist_obj.get("cover_url_big")
606            or playlist_obj.get("pic")
607            or playlist_obj.get("picUrl")
608        )
609    if cover:
610        playlist.metadata.images = UniqueList(
611            [
612                MediaItemImage(
613                    type=ImageType.THUMB,
614                    path=cover,
615                    provider=provider_instance_id,
616                    remotely_accessible=True,
617                )
618            ]
619        )
620    return playlist
621