music-assistant-server

7.8 KBPY
parsers.py
7.8 KB250 lines • python
1"""Parsers for YouSee Musik API responses."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import AlbumType, ContentType, ExternalID, ImageType
8from music_assistant_models.media_items import (
9    Album,
10    Artist,
11    AudioFormat,
12    MediaItemImage,
13    Playlist,
14    ProviderMapping,
15    Track,
16)
17
18from music_assistant.constants import (
19    VARIOUS_ARTISTS_MBID,
20    VARIOUS_ARTISTS_NAME,
21)
22from music_assistant.helpers.util import infer_album_type, parse_title_and_version, try_parse_int
23from music_assistant.providers.yousee.constants import (
24    CONF_QUALITY,
25    VARIOUS_ARTISTS_ID,
26)
27
28if TYPE_CHECKING:
29    from music_assistant.providers.yousee.api_client import JsonLike
30    from music_assistant.providers.yousee.provider import YouSeeMusikProvider
31
32
33async def parse_track(provider: YouSeeMusikProvider, track_obj: JsonLike) -> Track:
34    """Parse track data from YouSee API response."""
35    track = Track(
36        item_id=track_obj["id"],
37        provider=provider.instance_id,
38        name=track_obj["title"],
39        duration=track_obj.get("duration", 0),
40        provider_mappings={
41            ProviderMapping(
42                item_id=str(track_obj["id"]),
43                provider_domain=provider.domain,
44                provider_instance=provider.instance_id,
45                available=track_obj.get("availableToStream", True),
46                audio_format=AudioFormat(
47                    content_type=ContentType.MP4,
48                    bit_rate=try_parse_int(provider.config.get_value(CONF_QUALITY)),
49                ),
50                url=track_obj.get("share"),
51            )
52        },
53    )
54
55    if isrc := track_obj.get("isrc"):
56        track.external_ids.add((ExternalID.ISRC, isrc))
57
58    if "artist" in track_obj:
59        artist = parse_artist(provider, track_obj["artist"])
60        track.artists.append(artist)
61
62    for feat_artist_obj in track_obj.get("featuredArtists", {}).get("items", []):
63        feat_artist = parse_artist(provider, feat_artist_obj)
64        track.artists.append(feat_artist)
65
66    if "album" in track_obj:
67        album = await parse_album(provider, track_obj["album"])
68        track.album = album
69
70    if track_genre := track_obj.get("genre"):
71        track.metadata.genres = set(track_genre)
72
73    if track_label := track_obj.get("label"):
74        track.metadata.label = track_label
75
76    if track_obj.get("cover"):
77        track.metadata.add_image(
78            MediaItemImage(
79                type=ImageType.THUMB,
80                path=track_obj["cover"],
81                remotely_accessible=True,
82                provider=provider.instance_id,
83            )
84        )
85
86    return track
87
88
89def parse_artist(provider: YouSeeMusikProvider, artist_obj: JsonLike) -> Artist:
90    """Parse artist data from YouSee API response."""
91    artist = Artist(
92        item_id=artist_obj["id"],
93        provider=provider.instance_id,
94        name=artist_obj["title"],
95        uri=artist_obj.get("share"),
96        provider_mappings={
97            ProviderMapping(
98                item_id=str(artist_obj["id"]),
99                provider_domain=provider.domain,
100                provider_instance=provider.instance_id,
101            )
102        },
103    )
104
105    if artist.item_id == VARIOUS_ARTISTS_ID:
106        artist.mbid = VARIOUS_ARTISTS_MBID
107        artist.name = VARIOUS_ARTISTS_NAME
108
109    if artist_obj.get("cover"):
110        artist.metadata.add_image(
111            MediaItemImage(
112                type=ImageType.THUMB,
113                path=artist_obj["cover"],
114                remotely_accessible=True,
115                provider=provider.instance_id,
116            )
117        )
118
119    return artist
120
121
122async def parse_album(provider: YouSeeMusikProvider, album_obj: JsonLike) -> Album:
123    """Parse album data from YouSee API response."""
124    if "artist" not in album_obj:
125        return await provider.get_album(str(album_obj["id"]))
126
127    name, version = parse_title_and_version(album_obj["title"])
128    album = Album(
129        item_id=album_obj["id"],
130        provider=provider.instance_id,
131        name=name,
132        version=version,
133        provider_mappings={
134            ProviderMapping(
135                item_id=str(album_obj["id"]),
136                provider_domain=provider.domain,
137                provider_instance=provider.instance_id,
138                audio_format=AudioFormat(
139                    content_type=ContentType.MP4,
140                    bit_rate=try_parse_int(provider.config.get_value(CONF_QUALITY)),
141                ),
142                url=album_obj.get("share"),
143            )
144        },
145        is_playable=album_obj.get("available", True),
146    )
147
148    if album_upc := album_obj.get("upc"):
149        album.external_ids.add((ExternalID.BARCODE, album_upc))
150
151    album.artists.append(parse_artist(provider, album_obj["artist"]))
152
153    for feat_artist_obj in album_obj.get("featuredArtists", {}).get("items", []):
154        feat_artist = parse_artist(provider, feat_artist_obj)
155        album.artists.append(feat_artist)
156
157    if album_genre := album_obj.get("genre"):
158        album.metadata.genres = set(album_genre)
159
160    if album_obj.get("type") == "COMPILATION":
161        album.album_type = AlbumType.COMPILATION
162    elif album_obj.get("type") == "SINGLE":
163        album.album_type = AlbumType.SINGLE
164    elif album_obj.get("type") == "REGULAR":
165        album.album_type = AlbumType.ALBUM
166
167    inferred_type = infer_album_type(name, version)
168    if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
169        album.album_type = inferred_type
170
171    if album_obj.get("cover"):
172        album.metadata.add_image(
173            MediaItemImage(
174                type=ImageType.THUMB,
175                path=album_obj["cover"],
176                remotely_accessible=True,
177                provider=provider.instance_id,
178            )
179        )
180
181    if album_label := album_obj.get("label"):
182        album.metadata.label = album_label
183
184    if album_obj.get("releaseDate"):
185        album.year = try_parse_int(album_obj["releaseDate"][:4])
186
187    return album
188
189
190async def parse_playlist(provider: YouSeeMusikProvider, playlist_obj: JsonLike) -> Playlist:
191    """Parse playlist data from YouSee API response."""
192    playlist = Playlist(
193        item_id=str(playlist_obj["id"]),
194        provider=provider.instance_id,
195        name=playlist_obj["title"],
196        is_editable=playlist_obj["isOwned"],
197        provider_mappings={
198            ProviderMapping(
199                item_id=str(playlist_obj["id"]),
200                provider_domain=provider.domain,
201                provider_instance=provider.instance_id,
202                url=playlist_obj["share"],
203                is_unique=playlist_obj["isOwned"],
204            )
205        },
206    )
207
208    if playlist_obj.get("description"):
209        playlist.metadata.description = playlist_obj["description"]
210
211    if playlist_obj.get("cover"):
212        playlist.metadata.add_image(
213            MediaItemImage(
214                type=ImageType.THUMB,
215                path=playlist_obj["cover"],
216                remotely_accessible=True,
217                provider=provider.instance_id,
218            )
219        )
220
221    return playlist
222
223
224async def parse_lyrics(lyrics: list[JsonLike]) -> tuple[str | None, str | None]:
225    """Parse the YouSee lyrics payload and extract the lyric text in two formats if possible.
226
227    Returns:
228        Tuple[str | None, str | None]: lyrics (plain) and lyrics_lrc, if present.
229    """
230    if not lyrics:
231        return None, None
232
233    plain = ""
234    lrc = ""
235
236    for item in lyrics:
237        line = item.get("line", "")
238        if (start_ms := item.get("startInMs")) is not None:
239            minutes = start_ms // 60000
240            seconds = (start_ms % 60000) // 1000
241            milliseconds = start_ms % 1000
242            lrc += f"[{minutes:02}:{seconds:02}.{milliseconds:02}] {line}\n"
243
244        plain += line + "\n"
245
246    plain = plain.strip()
247    lrc = lrc.strip()
248
249    return plain if plain else None, lrc if lrc else None
250