music-assistant-server

1"""Parsers for Yandex Music API responses."""
2
3from __future__ import annotations
4
5from contextlib import suppress
6from datetime import datetime
7from typing import TYPE_CHECKING, Literal
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    Audiobook,
19    AudioFormat,
20    ItemMapping,
21    MediaItemImage,
22    Playlist,
23    Podcast,
24    PodcastEpisode,
25    ProviderMapping,
26    Track,
27    UniqueList,
28)
29
30from music_assistant.helpers.util import parse_title_and_version
31
32from .constants import (
33    IMAGE_SIZE_LARGE,
34    PROVIDER_DISPLAY_NAME_EN,
35    PROVIDER_DISPLAY_NAME_RU,
36    WEB_BASE_URL,
37    YANDEX_SYSTEM_OWNER_NAMES,
38)
39
40if TYPE_CHECKING:
41    from yandex_music import Album as YandexAlbum
42    from yandex_music import Artist as YandexArtist
43    from yandex_music import Playlist as YandexPlaylist
44    from yandex_music import Track as YandexTrack
45
46    from .provider import YandexMusicProvider
47
48
49AlbumKind = Literal["music", "podcast", "audiobook"]
50
51
52def classify_album(album_obj: YandexAlbum) -> AlbumKind:
53    """
54    Classify a Yandex album as music / podcast / audiobook.
55
56    Checks both ``meta_type`` and ``type`` for the substrings "audiobook" /
57    "podcast". The more specific "audiobook" signal wins over "podcast" on any
58    field because Yandex tags audiobooks with ``meta_type="podcast"`` *and*
59    ``type="audiobook"`` — empirically observed in production libraries.
60    Values are not documented in the yandex_music SDK.
61
62    :param album_obj: Yandex album object.
63    :return: One of "music", "podcast", "audiobook".
64    """
65    fields = [
66        (getattr(album_obj, "meta_type", None) or "").lower(),
67        (getattr(album_obj, "type", None) or "").lower(),
68    ]
69    if any("audiobook" in f for f in fields):
70        return "audiobook"
71    if any("podcast" in f for f in fields):
72        return "podcast"
73    return "music"
74
75
76def get_canonical_provider_name(provider: YandexMusicProvider) -> str:
77    """
78    Return the locale-aware canonical display name for the Yandex Music system account.
79
80    :param provider: The Yandex Music provider instance.
81    :return: Localized provider display name.
82    """
83    with suppress(Exception):
84        locale = (provider.mass.metadata.locale or "en_US").lower()
85        if locale.startswith("ru"):
86            return PROVIDER_DISPLAY_NAME_RU
87    return PROVIDER_DISPLAY_NAME_EN
88
89
90def _get_image_url(cover_uri: str | None, size: str = IMAGE_SIZE_LARGE) -> str | None:
91    """
92    Convert Yandex cover URI to full URL.
93
94    :param cover_uri: Yandex cover URI template.
95    :param size: Image size (e.g., '1000x1000').
96    :return: Full image URL or None.
97    """
98    if not cover_uri:
99        return None
100    # Cover URIs come in format "avatars.yandex.net/get-music-content/xxx/yyy/%%"
101    # Replace %% with the desired size
102    return f"https://{cover_uri.replace('%%', size)}"
103
104
105_NON_RUSSIAN_CYRILLIC_MARKERS = frozenset("їєґіўЇЄҐІЎ")
106
107
108def detect_description_language(text: str | None) -> Literal["ru"] | None:
109    """
110    Return ``"ru"`` for Russian-language text, ``None`` otherwise.
111
112    Yandex Music's API does not expose the language of artist / playlist /
113    podcast descriptions, so we infer it from script. A string classifies as
114    Russian when it (a) contains at least 8 Cyrillic characters that
115    (b) make up at least 50% of its length and (c) contains none of the
116    letters that mark another Slavic Cyrillic language (see
117    ``_NON_RUSSIAN_CYRILLIC_MARKERS`` — currently Ukrainian and Belarusian
118    discriminators). Everything else returns ``None`` so MA can fall back
119    to metadata providers for a user-localized bio.
120
121    :param text: The description string to classify.
122    :return: ``"ru"`` when the heuristic is confident, ``None`` otherwise.
123    """
124    if not text:
125        return None
126    text = text.strip()
127    if not text:
128        return None
129    if not _NON_RUSSIAN_CYRILLIC_MARKERS.isdisjoint(text):
130        return None
131    cyrillic = sum(1 for c in text if "Ѐ" <= c <= "ӿ")
132    # Floor + 50% share: a stray transliterated word in an English bio (e.g.
133    # an artist's Cyrillic name) must not flip the result to "ru".
134    if cyrillic >= 8 and cyrillic * 2 >= len(text):
135        return "ru"
136    return None
137
138
139def parse_artist(
140    provider: YandexMusicProvider,
141    artist_obj: YandexArtist,
142    *,
143    about: object | None = None,
144) -> Artist:
145    """
146    Parse Yandex artist object to MA Artist model.
147
148    :param provider: The Yandex Music provider instance.
149    :param artist_obj: Yandex artist object.
150    :param about: Optional ArtistAbout enrichment (description + listener stats).
151    :return: Music Assistant Artist model.
152    """
153    if artist_obj.id is None:
154        raise InvalidDataError("Yandex artist missing id")
155    artist_id = str(artist_obj.id)
156    artist = Artist(
157        item_id=artist_id,
158        provider=provider.instance_id,
159        name=artist_obj.name or "Unknown Artist",
160        provider_mappings={
161            ProviderMapping(
162                item_id=artist_id,
163                provider_domain=provider.domain,
164                provider_instance=provider.instance_id,
165                url=f"{WEB_BASE_URL}/artist/{artist_id}",
166            )
167        },
168    )
169
170    # Add image if available
171    if artist_obj.cover:
172        image_url = _get_image_url(artist_obj.cover.uri)
173        if image_url:
174            artist.metadata.images = UniqueList(
175                [
176                    MediaItemImage(
177                        type=ImageType.THUMB,
178                        path=image_url,
179                        provider=provider.instance_id,
180                        remotely_accessible=True,
181                    )
182                ]
183            )
184    elif artist_obj.og_image:
185        image_url = _get_image_url(artist_obj.og_image)
186        if image_url:
187            artist.metadata.images = UniqueList(
188                [
189                    MediaItemImage(
190                        type=ImageType.THUMB,
191                        path=image_url,
192                        provider=provider.instance_id,
193                        remotely_accessible=True,
194                    )
195                ]
196            )
197
198    if about is not None:
199        description = getattr(about, "description", None)
200        if description:
201            artist.metadata.description = description
202            artist.metadata.description_language = detect_description_language(description)
203        stats = getattr(about, "stats", None)
204        monthly = getattr(stats, "last_month_listeners", None) if stats else None
205        if monthly is not None:
206            artist.metadata.popularity = max(0, min(100, monthly // 10000))
207
208    return artist
209
210
211def _album_cover_images(
212    provider: YandexMusicProvider, album_obj: YandexAlbum
213) -> UniqueList[MediaItemImage]:
214    """
215    Build the UniqueList of images for an album-like object.
216
217    Prefers the templated ``cover_uri`` and falls back to ``og_image`` — matches
218    the selection rules used for podcasts and audiobooks so all album-like
219    parsers stay in sync.
220    """
221    images: UniqueList[MediaItemImage] = UniqueList()
222    image_url: str | None = None
223    if album_obj.cover_uri:
224        image_url = _get_image_url(album_obj.cover_uri)
225    elif album_obj.og_image:
226        image_url = _get_image_url(album_obj.og_image)
227    if image_url:
228        images.append(
229            MediaItemImage(
230                type=ImageType.THUMB,
231                path=image_url,
232                provider=provider.instance_id,
233                remotely_accessible=True,
234            )
235        )
236    return images
237
238
239def parse_album(provider: YandexMusicProvider, album_obj: YandexAlbum) -> Album:
240    """
241    Parse Yandex album object to MA Album model.
242
243    :param provider: The Yandex Music provider instance.
244    :param album_obj: Yandex album object.
245    :return: Music Assistant Album model.
246    """
247    if album_obj.id is None:
248        raise InvalidDataError("Yandex album missing id")
249    name, version = parse_title_and_version(
250        album_obj.title or "Unknown Album",
251        album_obj.version or None,
252    )
253    album_id = str(album_obj.id)
254
255    # Determine availability
256    available = album_obj.available or False
257
258    album = Album(
259        item_id=album_id,
260        provider=provider.instance_id,
261        name=name,
262        version=version,
263        provider_mappings={
264            ProviderMapping(
265                item_id=album_id,
266                provider_domain=provider.domain,
267                provider_instance=provider.instance_id,
268                audio_format=AudioFormat(
269                    content_type=ContentType.UNKNOWN,
270                ),
271                url=f"{WEB_BASE_URL}/album/{album_id}",
272                available=available,
273            )
274        },
275    )
276
277    # Parse artists
278    various_artist_album = False
279    if album_obj.artists:
280        for artist in album_obj.artists:
281            if artist.name and artist.name.lower() in ("various artists", "сборник"):
282                various_artist_album = True
283            album.artists.append(parse_artist(provider, artist))
284
285    # Determine album type
286    album_type_str = album_obj.type or "album"
287    if album_type_str == "compilation" or various_artist_album:
288        album.album_type = AlbumType.COMPILATION
289    elif album_type_str == "single":
290        album.album_type = AlbumType.SINGLE
291    else:
292        album.album_type = AlbumType.ALBUM
293
294    # Parse year
295    if album_obj.year:
296        album.year = album_obj.year
297    if album_obj.release_date:
298        with suppress(ValueError):
299            album.metadata.release_date = datetime.fromisoformat(album_obj.release_date)
300
301    # Parse metadata
302    if album_obj.genre:
303        album.metadata.genres = {album_obj.genre}
304
305    images = _album_cover_images(provider, album_obj)
306    if images:
307        album.metadata.images = images
308
309    return album
310
311
312def parse_track(
313    provider: YandexMusicProvider,
314    track_obj: YandexTrack,
315    lyrics: str | None = None,
316    lyrics_synced: bool = False,
317) -> Track:
318    """
319    Parse Yandex track object to MA Track model.
320
321    :param provider: The Yandex Music provider instance.
322    :param track_obj: Yandex track object.
323    :param lyrics: Optional lyrics text.
324    :param lyrics_synced: Whether lyrics are in synced LRC format.
325    :return: Music Assistant Track model.
326    """
327    if track_obj.id is None:
328        raise InvalidDataError("Yandex track missing id")
329    name, version = parse_title_and_version(
330        track_obj.title or "Unknown Track",
331        track_obj.version or None,
332    )
333    track_id = str(track_obj.id)
334
335    # Determine availability
336    available = track_obj.available or False
337
338    # Duration is in milliseconds in Yandex API
339    duration = (track_obj.duration_ms or 0) // 1000
340
341    track = Track(
342        item_id=track_id,
343        provider=provider.instance_id,
344        name=name,
345        version=version,
346        duration=duration,
347        provider_mappings={
348            ProviderMapping(
349                item_id=track_id,
350                provider_domain=provider.domain,
351                provider_instance=provider.instance_id,
352                audio_format=AudioFormat(
353                    content_type=ContentType.UNKNOWN,
354                ),
355                url=f"{WEB_BASE_URL}/track/{track_id}",
356                available=available,
357            )
358        },
359    )
360
361    # Parse artists
362    if track_obj.artists:
363        track.artists = UniqueList()
364        for artist in track_obj.artists:
365            track.artists.append(parse_artist(provider, artist))
366
367    # Parse album (full data so album gets cover art in the library)
368    if track_obj.albums and len(track_obj.albums) > 0:
369        album_obj = track_obj.albums[0]
370        track.album = parse_album(provider, album_obj)
371        # Also set track image from album cover if available
372        if album_obj.cover_uri:
373            image_url = _get_image_url(album_obj.cover_uri)
374            if image_url:
375                track.metadata.images = UniqueList(
376                    [
377                        MediaItemImage(
378                            type=ImageType.THUMB,
379                            path=image_url,
380                            provider=provider.instance_id,
381                            remotely_accessible=True,
382                        )
383                    ]
384                )
385
386    # Metadata
387    if track_obj.content_warning:
388        track.metadata.explicit = track_obj.content_warning == "explicit"
389
390    # Lyrics
391    if lyrics:
392        if lyrics_synced:
393            track.metadata.lrc_lyrics = lyrics
394        else:
395            track.metadata.lyrics = lyrics
396
397    return track
398
399
400def parse_playlist(
401    provider: YandexMusicProvider,
402    playlist_obj: YandexPlaylist,
403    owner_name: str | None = None,
404    *,
405    is_dynamic: bool = False,
406) -> Playlist:
407    """
408    Parse Yandex playlist object to MA Playlist model.
409
410    :param provider: The Yandex Music provider instance.
411    :param playlist_obj: Yandex playlist object.
412    :param owner_name: Optional owner name override.
413    :param is_dynamic: Mark the playlist as dynamic so Music Assistant does
414        not long-cache its content. Yandex regenerates "Playlist of the Day",
415        "DejaVu", "Premiere" etc. on a schedule, and those need a fresh read
416        on every browse so users actually see the updated selection.
417    :return: Music Assistant Playlist model.
418    """
419    # Playlist ID in Yandex is a combination of owner uid and playlist kind
420    owner_id = str(playlist_obj.owner.uid) if playlist_obj.owner else str(provider.client.user_id)
421    playlist_kind = str(playlist_obj.kind)
422    playlist_id = f"{owner_id}:{playlist_kind}"
423
424    # Determine if editable (user owns the playlist)
425    is_editable = owner_id == str(provider.client.user_id)
426
427    # Get owner name
428    if owner_name is None:
429        if playlist_obj.owner and playlist_obj.owner.name:
430            owner_name = playlist_obj.owner.name
431        elif is_editable:
432            owner_name = "Me"
433        else:
434            owner_name = get_canonical_provider_name(provider)
435
436    # Normalize all known system account name variants to locale-aware canonical form
437    if owner_name and owner_name.lower() in YANDEX_SYSTEM_OWNER_NAMES:
438        owner_name = get_canonical_provider_name(provider)
439
440    playlist = Playlist(
441        item_id=playlist_id,
442        provider=provider.instance_id,
443        name=playlist_obj.title or "Unknown Playlist",
444        owner=owner_name,
445        provider_mappings={
446            ProviderMapping(
447                item_id=playlist_id,
448                provider_domain=provider.domain,
449                provider_instance=provider.instance_id,
450                url=f"{WEB_BASE_URL}/users/{owner_id}/playlists/{playlist_kind}",
451                is_unique=is_editable,
452            )
453        },
454        is_editable=is_editable,
455        is_dynamic=is_dynamic,
456    )
457
458    # Metadata
459    if playlist_obj.description:
460        playlist.metadata.description = playlist_obj.description
461
462    # Add cover image
463    if playlist_obj.cover:
464        # Cover can be CoverImage or a string
465        cover = playlist_obj.cover
466        if hasattr(cover, "uri") and cover.uri:
467            image_url = _get_image_url(cover.uri)
468            if image_url:
469                playlist.metadata.images = UniqueList(
470                    [
471                        MediaItemImage(
472                            type=ImageType.THUMB,
473                            path=image_url,
474                            provider=provider.instance_id,
475                            remotely_accessible=True,
476                        )
477                    ]
478                )
479    elif playlist_obj.og_image:
480        image_url = _get_image_url(playlist_obj.og_image)
481        if image_url:
482            playlist.metadata.images = UniqueList(
483                [
484                    MediaItemImage(
485                        type=ImageType.THUMB,
486                        path=image_url,
487                        provider=provider.instance_id,
488                        remotely_accessible=True,
489                    )
490                ]
491            )
492
493    return playlist
494
495
496def parse_podcast(provider: YandexMusicProvider, album_obj: YandexAlbum) -> Podcast:
497    """
498    Parse Yandex album (meta_type=podcast) to MA Podcast model.
499
500    :param provider: The Yandex Music provider instance.
501    :param album_obj: Yandex album object classified as a podcast.
502    :return: Music Assistant Podcast model.
503    """
504    if album_obj.id is None:
505        raise InvalidDataError("Yandex podcast missing id")
506    name, _ = parse_title_and_version(
507        album_obj.title or "Unknown Podcast",
508        album_obj.version or None,
509    )
510    podcast_id = str(album_obj.id)
511    available = album_obj.available or False
512
513    # Publisher: prefer labels[0].name; fall back to first artist name
514    publisher: str | None = None
515    labels = getattr(album_obj, "labels", None)
516    if labels:
517        first = labels[0]
518        label_name = getattr(first, "name", None) if not isinstance(first, str) else first
519        if label_name:
520            publisher = label_name
521    if not publisher and album_obj.artists:
522        first_artist = album_obj.artists[0]
523        if first_artist.name:
524            publisher = first_artist.name
525
526    podcast = Podcast(
527        item_id=podcast_id,
528        provider=provider.instance_id,
529        name=name,
530        provider_mappings={
531            ProviderMapping(
532                item_id=podcast_id,
533                provider_domain=provider.domain,
534                provider_instance=provider.instance_id,
535                audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
536                url=f"{WEB_BASE_URL}/album/{podcast_id}",
537                available=available,
538            )
539        },
540        publisher=publisher,
541        total_episodes=album_obj.track_count,
542    )
543
544    description = album_obj.description or album_obj.short_description
545    if description:
546        podcast.metadata.description = description
547    if album_obj.content_warning:
548        podcast.metadata.explicit = album_obj.content_warning == "explicit"
549
550    images = _album_cover_images(provider, album_obj)
551    if images:
552        podcast.metadata.images = images
553
554    if album_obj.genre:
555        podcast.metadata.genres = {album_obj.genre}
556    else:
557        podcast.metadata.genres = {"Spoken Word"}
558
559    if album_obj.release_date:
560        with suppress(ValueError):
561            podcast.metadata.release_date = datetime.fromisoformat(album_obj.release_date)
562
563    return podcast
564
565
566def parse_podcast_episode(
567    provider: YandexMusicProvider,
568    track_obj: YandexTrack,
569    podcast: Podcast,
570    position: int = 0,
571) -> PodcastEpisode:
572    """
573    Parse Yandex track (episode of a podcast album) to MA PodcastEpisode.
574
575    :param provider: The Yandex Music provider instance.
576    :param track_obj: Yandex track object.
577    :param podcast: Parent Podcast object.
578    :param position: 1-based episode index (0 if unknown).
579    :return: Music Assistant PodcastEpisode model.
580    """
581    if track_obj.id is None:
582        raise InvalidDataError("Yandex podcast episode missing id")
583    episode_id = str(track_obj.id)
584    available = track_obj.available or False
585    duration = (track_obj.duration_ms or 0) // 1000
586
587    episode_name = track_obj.title or (f"Episode {position}" if position else "Unknown Episode")
588    episode = PodcastEpisode(
589        item_id=episode_id,
590        provider=provider.instance_id,
591        name=episode_name,
592        duration=duration,
593        podcast=podcast,
594        position=position,
595        provider_mappings={
596            ProviderMapping(
597                item_id=episode_id,
598                provider_domain=provider.domain,
599                provider_instance=provider.instance_id,
600                audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
601                url=f"{WEB_BASE_URL}/track/{episode_id}",
602                available=available,
603            )
604        },
605    )
606
607    if track_obj.short_description:
608        episode.metadata.description = track_obj.short_description
609    if track_obj.content_warning:
610        episode.metadata.explicit = track_obj.content_warning == "explicit"
611
612    # Track cover → fall back to podcast cover
613    if track_obj.cover_uri:
614        image_url = _get_image_url(track_obj.cover_uri)
615        if image_url:
616            episode.metadata.images = UniqueList(
617                [
618                    MediaItemImage(
619                        type=ImageType.THUMB,
620                        path=image_url,
621                        provider=provider.instance_id,
622                        remotely_accessible=True,
623                    )
624                ]
625            )
626    elif track_obj.og_image:
627        image_url = _get_image_url(track_obj.og_image)
628        if image_url:
629            episode.metadata.images = UniqueList(
630                [
631                    MediaItemImage(
632                        type=ImageType.THUMB,
633                        path=image_url,
634                        provider=provider.instance_id,
635                        remotely_accessible=True,
636                    )
637                ]
638            )
639    if not episode.metadata.images and podcast.metadata.images:
640        episode.metadata.images = UniqueList(podcast.metadata.images)
641
642    return episode
643
644
645def parse_audiobook(provider: YandexMusicProvider, album_obj: YandexAlbum) -> Audiobook:
646    """
647    Parse Yandex album (meta_type=audiobook) to MA Audiobook model.
648
649    :param provider: The Yandex Music provider instance.
650    :param album_obj: Yandex album object classified as an audiobook.
651    :return: Music Assistant Audiobook model. Chapters and duration are filled
652        by the provider's get_audiobook() method after loading album tracks.
653    """
654    if album_obj.id is None:
655        raise InvalidDataError("Yandex audiobook missing id")
656    name, _ = parse_title_and_version(
657        album_obj.title or "Unknown Audiobook",
658        album_obj.version or None,
659    )
660    audiobook_id = str(album_obj.id)
661    available = album_obj.available or False
662
663    # Publisher: prefer labels[0]; fall back to nothing (authors sit on artists)
664    publisher: str | None = None
665    labels = getattr(album_obj, "labels", None)
666    if labels:
667        first = labels[0]
668        label_name = getattr(first, "name", None) if not isinstance(first, str) else first
669        if label_name:
670            publisher = label_name
671
672    authors: UniqueList[str | Artist | ItemMapping] = UniqueList()
673    if album_obj.artists:
674        for artist in album_obj.artists:
675            if artist.name:
676                authors.append(artist.name)
677
678    audiobook = Audiobook(
679        item_id=audiobook_id,
680        provider=provider.instance_id,
681        name=name,
682        provider_mappings={
683            ProviderMapping(
684                item_id=audiobook_id,
685                provider_domain=provider.domain,
686                provider_instance=provider.instance_id,
687                audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
688                url=f"{WEB_BASE_URL}/album/{audiobook_id}",
689                available=available,
690            )
691        },
692        publisher=publisher,
693        authors=authors,
694        narrators=UniqueList(),
695        duration=0,
696    )
697
698    description = album_obj.description or album_obj.short_description
699    if description:
700        audiobook.metadata.description = description
701    if album_obj.content_warning:
702        audiobook.metadata.explicit = album_obj.content_warning == "explicit"
703
704    images = _album_cover_images(provider, album_obj)
705    if images:
706        audiobook.metadata.images = images
707
708    if album_obj.genre:
709        audiobook.metadata.genres = {album_obj.genre}
710    else:
711        audiobook.metadata.genres = {"Spoken Word"}
712
713    if album_obj.release_date:
714        with suppress(ValueError):
715            audiobook.metadata.release_date = datetime.fromisoformat(album_obj.release_date)
716
717    listening_finished = getattr(album_obj, "listening_finished", None)
718    if listening_finished is not None:
719        audiobook.fully_played = bool(listening_finished)
720
721    return audiobook
722