music-assistant-server

1"""Parsers for KION Music API responses."""
2
3from __future__ import annotations
4
5from contextlib import suppress
6from datetime import datetime
7from typing import TYPE_CHECKING
8
9from music_assistant_models.enums import (
10    AlbumType,
11    ContentType,
12    ImageType,
13)
14from music_assistant_models.errors import InvalidDataError
15from music_assistant_models.media_items import (
16    Album,
17    Artist,
18    AudioFormat,
19    MediaItemImage,
20    Playlist,
21    ProviderMapping,
22    Track,
23    UniqueList,
24)
25
26from music_assistant.helpers.util import parse_title_and_version
27
28from .constants import (
29    IMAGE_SIZE_LARGE,
30    KION_SYSTEM_OWNER_NAMES,
31    PROVIDER_DISPLAY_NAME_EN,
32    PROVIDER_DISPLAY_NAME_RU,
33    WEB_BASE_URL,
34)
35
36if TYPE_CHECKING:
37    from yandex_music import Album as KionAlbum
38    from yandex_music import Artist as KionArtist
39    from yandex_music import Playlist as KionPlaylist
40    from yandex_music import Track as KionTrack
41
42    from .provider import KionMusicProvider
43
44
45def get_canonical_provider_name(provider: KionMusicProvider) -> str:
46    """
47    Return the locale-aware canonical display name for the KION Music system account.
48
49    :param provider: The KION Music provider instance.
50    :return: Localized provider display name.
51    """
52    with suppress(Exception):
53        locale = (provider.mass.metadata.locale or "en_US").lower()
54        if locale.startswith("ru"):
55            return PROVIDER_DISPLAY_NAME_RU
56    return PROVIDER_DISPLAY_NAME_EN
57
58
59def _get_image_url(cover_uri: str | None, size: str = IMAGE_SIZE_LARGE) -> str | None:
60    """
61    Convert Kion cover URI to full URL.
62
63    :param cover_uri: Kion cover URI template.
64    :param size: Image size (e.g., '1000x1000').
65    :return: Full image URL or None.
66    """
67    if not cover_uri:
68        return None
69    # Cover URIs come in format "avatars.kion.net/get-music-content/xxx/yyy/%%"
70    # Replace %% with the desired size
71    return f"https://{cover_uri.replace('%%', size)}"
72
73
74def parse_artist(
75    provider: KionMusicProvider,
76    artist_obj: KionArtist,
77    *,
78    about: object | None = None,
79) -> Artist:
80    """
81    Parse Kion artist object to MA Artist model.
82
83    :param provider: The KION Music provider instance.
84    :param artist_obj: Kion artist object.
85    :param about: Optional ArtistAbout enrichment (description + listener stats).
86    :return: Music Assistant Artist model.
87    """
88    if artist_obj.id is None:
89        raise InvalidDataError("Kion artist missing id")
90    artist_id = str(artist_obj.id)
91    artist = Artist(
92        item_id=artist_id,
93        provider=provider.instance_id,
94        name=artist_obj.name or "Unknown Artist",
95        provider_mappings={
96            ProviderMapping(
97                item_id=artist_id,
98                provider_domain=provider.domain,
99                provider_instance=provider.instance_id,
100                url=f"{WEB_BASE_URL}/artist/{artist_id}",
101            )
102        },
103    )
104
105    # Add image if available
106    if artist_obj.cover:
107        image_url = _get_image_url(artist_obj.cover.uri)
108        if image_url:
109            artist.metadata.images = UniqueList(
110                [
111                    MediaItemImage(
112                        type=ImageType.THUMB,
113                        path=image_url,
114                        provider=provider.instance_id,
115                        remotely_accessible=True,
116                    )
117                ]
118            )
119    elif artist_obj.og_image:
120        image_url = _get_image_url(artist_obj.og_image)
121        if image_url:
122            artist.metadata.images = UniqueList(
123                [
124                    MediaItemImage(
125                        type=ImageType.THUMB,
126                        path=image_url,
127                        provider=provider.instance_id,
128                        remotely_accessible=True,
129                    )
130                ]
131            )
132
133    if about is not None:
134        description = getattr(about, "description", None)
135        if description:
136            artist.metadata.description = description
137        stats = getattr(about, "stats", None)
138        monthly = getattr(stats, "last_month_listeners", None) if stats else None
139        if monthly:
140            artist.metadata.popularity = max(0, min(100, monthly // 10000))
141
142    return artist
143
144
145def parse_album(provider: KionMusicProvider, album_obj: KionAlbum) -> Album:
146    """
147    Parse Kion album object to MA Album model.
148
149    :param provider: The KION Music provider instance.
150    :param album_obj: Kion album object.
151    :return: Music Assistant Album model.
152    """
153    if album_obj.id is None:
154        raise InvalidDataError("Kion album missing id")
155    name, version = parse_title_and_version(
156        album_obj.title or "Unknown Album",
157        album_obj.version or None,
158    )
159    album_id = str(album_obj.id)
160
161    # Determine availability
162    available = album_obj.available or False
163
164    album = Album(
165        item_id=album_id,
166        provider=provider.instance_id,
167        name=name,
168        version=version,
169        provider_mappings={
170            ProviderMapping(
171                item_id=album_id,
172                provider_domain=provider.domain,
173                provider_instance=provider.instance_id,
174                audio_format=AudioFormat(
175                    content_type=ContentType.UNKNOWN,
176                ),
177                url=f"{WEB_BASE_URL}/album/{album_id}",
178                available=available,
179            )
180        },
181    )
182
183    # Parse artists
184    various_artist_album = False
185    if album_obj.artists:
186        for artist in album_obj.artists:
187            if artist.name and artist.name.lower() in ("various artists", "сборник"):
188                various_artist_album = True
189            album.artists.append(parse_artist(provider, artist))
190
191    # Determine album type
192    album_type_str = album_obj.type or "album"
193    if album_type_str == "compilation" or various_artist_album:
194        album.album_type = AlbumType.COMPILATION
195    elif album_type_str == "single":
196        album.album_type = AlbumType.SINGLE
197    else:
198        album.album_type = AlbumType.ALBUM
199
200    # Parse year
201    if album_obj.year:
202        album.year = album_obj.year
203    if album_obj.release_date:
204        with suppress(ValueError):
205            album.metadata.release_date = datetime.fromisoformat(album_obj.release_date)
206
207    # Parse metadata
208    if album_obj.genre:
209        album.metadata.genres = {album_obj.genre}
210
211    # Add cover image
212    if album_obj.cover_uri:
213        image_url = _get_image_url(album_obj.cover_uri)
214        if image_url:
215            album.metadata.images = UniqueList(
216                [
217                    MediaItemImage(
218                        type=ImageType.THUMB,
219                        path=image_url,
220                        provider=provider.instance_id,
221                        remotely_accessible=True,
222                    )
223                ]
224            )
225    elif album_obj.og_image:
226        image_url = _get_image_url(album_obj.og_image)
227        if image_url:
228            album.metadata.images = UniqueList(
229                [
230                    MediaItemImage(
231                        type=ImageType.THUMB,
232                        path=image_url,
233                        provider=provider.instance_id,
234                        remotely_accessible=True,
235                    )
236                ]
237            )
238
239    return album
240
241
242def parse_track(
243    provider: KionMusicProvider,
244    track_obj: KionTrack,
245    lyrics: str | None = None,
246    lyrics_synced: bool = False,
247) -> Track:
248    """
249    Parse Kion track object to MA Track model.
250
251    :param provider: The KION Music provider instance.
252    :param track_obj: Kion track object.
253    :param lyrics: Optional lyrics text.
254    :param lyrics_synced: Whether lyrics are in synced LRC format.
255    :return: Music Assistant Track model.
256    """
257    if track_obj.id is None:
258        raise InvalidDataError("Kion track missing id")
259    name, version = parse_title_and_version(
260        track_obj.title or "Unknown Track",
261        track_obj.version or None,
262    )
263    track_id = str(track_obj.id)
264
265    # Determine availability
266    available = track_obj.available or False
267
268    # Duration is in milliseconds in Kion API
269    duration = (track_obj.duration_ms or 0) // 1000
270
271    track = Track(
272        item_id=track_id,
273        provider=provider.instance_id,
274        name=name,
275        version=version,
276        duration=duration,
277        provider_mappings={
278            ProviderMapping(
279                item_id=track_id,
280                provider_domain=provider.domain,
281                provider_instance=provider.instance_id,
282                audio_format=AudioFormat(
283                    content_type=ContentType.UNKNOWN,
284                ),
285                url=f"{WEB_BASE_URL}/track/{track_id}",
286                available=available,
287            )
288        },
289    )
290
291    # Parse artists
292    if track_obj.artists:
293        track.artists = UniqueList()
294        for artist in track_obj.artists:
295            track.artists.append(parse_artist(provider, artist))
296
297    # Parse album (full data so album gets cover art in the library)
298    if track_obj.albums and len(track_obj.albums) > 0:
299        album_obj = track_obj.albums[0]
300        track.album = parse_album(provider, album_obj)
301        # Also set track image from album cover if available
302        if album_obj.cover_uri:
303            image_url = _get_image_url(album_obj.cover_uri)
304            if image_url:
305                track.metadata.images = UniqueList(
306                    [
307                        MediaItemImage(
308                            type=ImageType.THUMB,
309                            path=image_url,
310                            provider=provider.instance_id,
311                            remotely_accessible=True,
312                        )
313                    ]
314                )
315
316    # Parse external IDs
317    if track_obj.real_id:
318        # real_id can be used as an identifier
319        pass
320
321    # Metadata
322    if track_obj.content_warning:
323        track.metadata.explicit = track_obj.content_warning == "explicit"
324
325    # Lyrics
326    # Core metadata controller checks `metadata.lyrics` first when picking
327    # the provider's own lyrics, so always populate `.lyrics` — otherwise
328    # synced-only LRC would be skipped and replaced by fallback metadata
329    # providers (lrclib/genius).
330    if lyrics:
331        track.metadata.lyrics = lyrics
332        if lyrics_synced:
333            track.metadata.lrc_lyrics = lyrics
334
335    return track
336
337
338def parse_playlist(
339    provider: KionMusicProvider, playlist_obj: KionPlaylist, owner_name: str | None = None
340) -> Playlist:
341    """
342    Parse Kion playlist object to MA Playlist model.
343
344    :param provider: The KION Music provider instance.
345    :param playlist_obj: Kion playlist object.
346    :param owner_name: Optional owner name override.
347    :return: Music Assistant Playlist model.
348    """
349    # Playlist ID in Kion is a combination of owner uid and playlist kind
350    owner_id = str(playlist_obj.owner.uid) if playlist_obj.owner else str(provider.client.user_id)
351    playlist_kind = str(playlist_obj.kind)
352    playlist_id = f"{owner_id}:{playlist_kind}"
353
354    # Determine if editable (user owns the playlist)
355    is_editable = owner_id == str(provider.client.user_id)
356
357    # Get owner name
358    if owner_name is None:
359        if playlist_obj.owner and playlist_obj.owner.name:
360            owner_name = playlist_obj.owner.name
361        elif is_editable:
362            owner_name = "Me"
363        else:
364            owner_name = get_canonical_provider_name(provider)
365
366    # Normalize all known system account name variants to locale-aware canonical form
367    if owner_name and owner_name.lower() in KION_SYSTEM_OWNER_NAMES:
368        owner_name = get_canonical_provider_name(provider)
369
370    playlist = Playlist(
371        item_id=playlist_id,
372        provider=provider.instance_id,
373        name=playlist_obj.title or "Unknown Playlist",
374        owner=owner_name,
375        provider_mappings={
376            ProviderMapping(
377                item_id=playlist_id,
378                provider_domain=provider.domain,
379                provider_instance=provider.instance_id,
380                url=f"{WEB_BASE_URL}/users/{owner_id}/playlists/{playlist_kind}",
381                is_unique=is_editable,
382            )
383        },
384        is_editable=is_editable,
385    )
386
387    # Metadata
388    if playlist_obj.description:
389        playlist.metadata.description = playlist_obj.description
390
391    # Add cover image
392    if playlist_obj.cover:
393        # Cover can be CoverImage or a string
394        cover = playlist_obj.cover
395        if hasattr(cover, "uri") and cover.uri:
396            image_url = _get_image_url(cover.uri)
397            if image_url:
398                playlist.metadata.images = UniqueList(
399                    [
400                        MediaItemImage(
401                            type=ImageType.THUMB,
402                            path=image_url,
403                            provider=provider.instance_id,
404                            remotely_accessible=True,
405                        )
406                    ]
407                )
408    elif playlist_obj.og_image:
409        image_url = _get_image_url(playlist_obj.og_image)
410        if image_url:
411            playlist.metadata.images = UniqueList(
412                [
413                    MediaItemImage(
414                        type=ImageType.THUMB,
415                        path=image_url,
416                        provider=provider.instance_id,
417                        remotely_accessible=True,
418                    )
419                ]
420            )
421
422    return playlist
423