music-assistant-server

1"""Parser for ABS -> MASS."""
2
3from contextlib import suppress
4from datetime import datetime
5from typing import TYPE_CHECKING
6
7from aioaudiobookshelf.schema.author import AuthorExpanded as AbsAuthorExpanded
8from aioaudiobookshelf.schema.author import AuthorMinified as AbsAuthorMinified
9from aioaudiobookshelf.schema.author import Narrator as AbsNarrator
10from aioaudiobookshelf.schema.library import (
11    LibraryItemExpandedBook as AbsLibraryItemExpandedBook,
12)
13from aioaudiobookshelf.schema.library import (
14    LibraryItemExpandedPodcast as AbsLibraryItemExpandedPodcast,
15)
16from aioaudiobookshelf.schema.library import (
17    LibraryItemMinifiedPodcast as AbsLibraryItemMinifiedPodcast,
18)
19from aioaudiobookshelf.schema.library import (
20    LibraryItemPodcast as AbsLibraryItemPodcast,
21)
22from aioaudiobookshelf.schema.podcast import PodcastEpisode as AbsPodcastEpisode
23from aioaudiobookshelf.schema.podcast import (
24    PodcastEpisodeExpanded as AbsPodcastEpisodeExpanded,
25)
26from music_assistant_models.enums import ArtistType, ContentType, ImageType, MediaType
27from music_assistant_models.media_items import Artist as MassArtist
28from music_assistant_models.media_items import Audiobook as MassAudiobook
29from music_assistant_models.media_items import (
30    AudioFormat,
31    ItemMapping,
32    MediaItemChapter,
33    MediaItemCollection,
34    MediaItemImage,
35    ProviderMapping,
36    UniqueList,
37)
38from music_assistant_models.media_items import Playlist as MassPlaylist
39from music_assistant_models.media_items import Podcast as MassPodcast
40from music_assistant_models.media_items import PodcastEpisode as MassPodcastEpisode
41
42from music_assistant.helpers.datetime import from_utc_timestamp
43from music_assistant.providers.audiobookshelf.helpers import NarratorHelper
44
45if TYPE_CHECKING:
46    from aioaudiobookshelf.schema.media_progress import MediaProgress as AbsMediaProgress
47    from aioaudiobookshelf.schema.playlist import PlaylistExpanded as AbsPlaylistExpanded
48
49
50def _build_cover_url(*, base_url: str, item_id: str, token: str, version: int | None = None) -> str:
51    """
52    Build the cover url for an Audiobookshelf library item.
53
54    :param base_url: Base url of the Audiobookshelf server.
55    :param item_id: Id of the library item to fetch the cover for.
56    :param token: Access token for the Audiobookshelf api.
57    :param version: Last-updated timestamp of the item.
58    """
59    # the cover endpoint is static and keeps its filename when artwork is
60    # replaced in place, so append the last-updated timestamp to bust caches
61    cover_url = f"{base_url}/api/items/{item_id}/cover?token={token}"
62    if version is not None:
63        cover_url += f"&ts={version}"
64    return cover_url
65
66
67def parse_playlist(
68    *,
69    abs_playlist: AbsPlaylistExpanded,
70    instance_id: str,
71    domain: str,
72    token: str,
73    base_url: str,
74    owner: str,
75    media_type: MediaType,
76) -> MassPlaylist:
77    """Translate AbsPlaylist to MassPlaylist."""
78    mass_playlist = MassPlaylist(
79        item_id=abs_playlist.id_,
80        provider=instance_id,
81        name=abs_playlist.name,
82        sort_name=abs_playlist.name,
83        provider_mappings={
84            ProviderMapping(
85                item_id=abs_playlist.id_, provider_domain=domain, provider_instance=instance_id
86            )
87        },
88        supported_mediatypes={media_type},
89        is_editable=True,
90        owner=owner,
91    )
92    # cover
93    if abs_playlist.cover_path is not None:
94        cover_url = _build_cover_url(
95            base_url=base_url,
96            item_id=abs_playlist.id_,
97            token=token,
98            version=abs_playlist.last_update,
99        )
100        mass_playlist.metadata.images = UniqueList(
101            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
102        )
103    else:
104        mass_playlist.metadata.images = UniqueList([])
105    return mass_playlist
106
107
108def parse_podcast(
109    *,
110    abs_podcast: AbsLibraryItemExpandedPodcast
111    | AbsLibraryItemMinifiedPodcast
112    | AbsLibraryItemPodcast,
113    instance_id: str,
114    domain: str,
115    token: str | None,
116    base_url: str,
117) -> MassPodcast:
118    """Translate ABSPodcast to MassPodcast."""
119    title = abs_podcast.media.metadata.title
120    # Per API doc title may be None.
121    if title is None:
122        title = "UNKNOWN"
123    mass_podcast = MassPodcast(
124        item_id=abs_podcast.id_,
125        name=title,
126        publisher=abs_podcast.media.metadata.author,
127        provider=instance_id,
128        provider_mappings={
129            ProviderMapping(
130                item_id=abs_podcast.id_,
131                provider_domain=domain,
132                provider_instance=instance_id,
133            )
134        },
135    )
136    mass_podcast.metadata.description = abs_podcast.media.metadata.description
137    if token is not None and abs_podcast.media.cover_path is not None:
138        image_url = _build_cover_url(
139            base_url=base_url,
140            item_id=abs_podcast.id_,
141            token=token,
142            version=abs_podcast.updated_at,
143        )
144        mass_podcast.metadata.images = UniqueList(
145            [MediaItemImage(type=ImageType.THUMB, path=image_url, provider=instance_id)]
146        )
147    elif abs_podcast.media.cover_path is None:
148        mass_podcast.metadata.images = UniqueList([])
149
150    mass_podcast.metadata.explicit = abs_podcast.media.metadata.explicit
151    mass_podcast.metadata.languages = (
152        UniqueList([abs_podcast.media.metadata.language])
153        if abs_podcast.media.metadata.language is not None
154        else UniqueList([])
155    )
156    mass_podcast.metadata.genres = (
157        set(abs_podcast.media.metadata.genres)
158        if abs_podcast.media.metadata.genres is not None
159        else set()
160    )
161
162    # podcast object has no published_at int, but an iso string
163    if abs_podcast.media.metadata.release_date is not None:
164        with suppress(ValueError):
165            mass_podcast.metadata.release_date = datetime.fromisoformat(
166                abs_podcast.media.metadata.release_date
167            )
168
169    if isinstance(abs_podcast, AbsLibraryItemExpandedPodcast | AbsLibraryItemPodcast):
170        mass_podcast.total_episodes = len(abs_podcast.media.episodes)
171    elif isinstance(abs_podcast, AbsLibraryItemMinifiedPodcast):
172        mass_podcast.total_episodes = abs_podcast.media.num_episodes
173
174    return mass_podcast
175
176
177def parse_podcast_episode(
178    *,
179    episode: AbsPodcastEpisode | AbsPodcastEpisodeExpanded,
180    prov_podcast_id: str,
181    prov_podcast_name: str | None,
182    fallback_episode_cnt: int | None = None,
183    instance_id: str,
184    domain: str,
185    token: str | None,
186    base_url: str,
187    media_progress: AbsMediaProgress | None = None,
188    cover_path: str | None = None,
189    cover_version: int | None = None,
190) -> MassPodcastEpisode:
191    """
192    Translate ABSPodcastEpisode to MassPodcastEpisode.
193
194    For an episode the id is set to f"{podcast_id} {episode_id}".
195    ABS ids have no spaces, so we can split at a space to retrieve both
196    in other functions.
197
198    NOTE: We should always use a PodcastEpisodeExpanded when possible.
199    A PodcastEpisode has only limited information, and is currently only used
200    within the recommendations.
201    """
202    # ruff: noqa: PLR0913 (too many arguments)
203    episode_id = f"{prov_podcast_id} {episode.id_}"
204
205    if isinstance(episode, AbsPodcastEpisodeExpanded):
206        url = f"{base_url}{episode.audio_track.content_url}"
207        duration = int(episode.duration)
208        provider_mappings = {
209            ProviderMapping(
210                item_id=episode_id,
211                provider_domain=domain,
212                provider_instance=instance_id,
213                audio_format=AudioFormat(
214                    content_type=ContentType.UNKNOWN,
215                ),
216                url=url,
217            )
218        }
219    else:
220        # PodcastEpisode
221        duration = 0  # mass default
222        provider_mappings = {
223            ProviderMapping(
224                item_id=episode_id,
225                provider_domain=domain,
226                provider_instance=instance_id,
227            )
228        }
229
230    release_date: datetime | None = None
231    if episode.published_at is not None:
232        position = episode.published_at
233        # abs published_at is ms epoch; leave the date unset if it is out of range
234        with suppress(ValueError, OverflowError, OSError):
235            release_date = from_utc_timestamp(episode.published_at / 1000)
236    else:
237        position = 0
238        if fallback_episode_cnt is not None:
239            position = fallback_episode_cnt
240    mass_episode = MassPodcastEpisode(
241        item_id=episode_id,
242        provider=instance_id,
243        name=episode.title,
244        duration=duration,
245        position=position,
246        podcast=ItemMapping(
247            item_id=prov_podcast_id,
248            provider=instance_id,
249            name=prov_podcast_name or episode.title,
250            media_type=MediaType.PODCAST,
251        ),
252        provider_mappings=provider_mappings,
253    )
254
255    mass_episode.metadata.release_date = release_date
256    if episode.description:
257        mass_episode.metadata.description = episode.description
258
259    # cover image
260    if token is not None and cover_path:
261        url_cover = _build_cover_url(
262            base_url=base_url,
263            item_id=prov_podcast_id,
264            token=token,
265            version=cover_version,
266        )
267        mass_episode.metadata.images = UniqueList(
268            [MediaItemImage(type=ImageType.THUMB, path=url_cover, provider=instance_id)]
269        )
270
271    if media_progress is not None and media_progress.current_time is not None:
272        mass_episode.resume_position_ms = int(media_progress.current_time * 1000)
273        mass_episode.fully_played = media_progress.is_finished
274
275    mass_episode.metadata.chapters = [
276        MediaItemChapter(
277            position=position,
278            name=chapter.title,
279            start=chapter.start,
280            end=chapter.end,
281        )
282        for position, chapter in enumerate(episode.chapters, 1)
283    ]
284
285    return mass_episode
286
287
288def parse_audiobook(
289    *,
290    abs_audiobook: AbsLibraryItemExpandedBook,
291    instance_id: str,
292    domain: str,
293    token: str | None,
294    base_url: str,
295    audiobook_narrators: set[AbsNarrator] | set[NarratorHelper],
296    media_progress: AbsMediaProgress | None = None,
297) -> MassAudiobook:
298    """Translate AbsBook to Mass Book."""
299    title = abs_audiobook.media.metadata.title
300    # Per API doc title may be None.
301    if title is None:
302        title = "UNKNOWN TITLE"
303    subtitle = abs_audiobook.media.metadata.subtitle
304    if subtitle is not None or subtitle:
305        title += f" | {subtitle}"
306    mass_audiobook = MassAudiobook(
307        item_id=abs_audiobook.id_,
308        provider=instance_id,
309        name=title,
310        duration=int(abs_audiobook.media.duration),
311        provider_mappings={
312            ProviderMapping(
313                item_id=abs_audiobook.id_,
314                provider_domain=domain,
315                provider_instance=instance_id,
316            )
317        },
318        publisher=abs_audiobook.media.metadata.publisher,
319    )
320    mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
321    mass_audiobook.metadata.languages = (
322        UniqueList([abs_audiobook.media.metadata.language])
323        if abs_audiobook.media.metadata.language is not None
324        else UniqueList([])
325    )
326
327    if abs_audiobook.media.metadata.published_date is not None:
328        with suppress(ValueError):
329            mass_audiobook.metadata.release_date = datetime.fromisoformat(
330                abs_audiobook.media.metadata.published_date
331            )
332    elif abs_audiobook.media.metadata.published_year is not None:
333        with suppress(ValueError):
334            # ruff: noqa: DTZ001 # ignore tzinfo, this is a fallback attempt
335            mass_audiobook.metadata.release_date = datetime(
336                year=int(abs_audiobook.media.metadata.published_year), month=1, day=1
337            )
338
339    book_series: list[MediaItemCollection] = []
340    for abs_series_sequence in abs_audiobook.media.metadata.series:
341        book_series.append(
342            MediaItemCollection(
343                title=abs_series_sequence.name, sequence=abs_series_sequence.sequence
344            )
345        )
346    mass_audiobook.metadata.collections = UniqueList(book_series)
347
348    mass_audiobook.metadata.genres = (
349        set(abs_audiobook.media.metadata.genres)
350        if abs_audiobook.media.metadata.genres is not None
351        else set()
352    )
353
354    mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
355
356    # cover
357    if token is not None and abs_audiobook.media.cover_path is not None:
358        cover_url = _build_cover_url(
359            base_url=base_url,
360            item_id=abs_audiobook.id_,
361            token=token,
362            version=abs_audiobook.updated_at,
363        )
364        mass_audiobook.metadata.images = UniqueList(
365            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
366        )
367    elif abs_audiobook.media.cover_path is None:
368        mass_audiobook.metadata.images = UniqueList([])
369
370    # expanded version
371    mass_audiobook.authors.set(
372        [
373            parse_author(
374                abs_author=author,
375                instance_id=instance_id,
376                domain=domain,
377                token=token,
378                base_url=base_url,
379            )
380            for author in abs_audiobook.media.metadata.authors
381        ]
382    )
383
384    mass_audiobook.narrators.set(
385        [
386            parse_narrator(abs_narrator=narrator, instance_id=instance_id, domain=domain)
387            for narrator in audiobook_narrators
388        ]
389    )
390
391    chapters = []
392    for idx, chapter in enumerate(abs_audiobook.media.chapters, 1):
393        chapters.append(
394            MediaItemChapter(
395                position=idx,
396                name=chapter.title,
397                start=chapter.start,
398                end=chapter.end,
399            )
400        )
401    mass_audiobook.metadata.chapters = chapters
402
403    if media_progress is not None and media_progress.current_time is not None:
404        mass_audiobook.resume_position_ms = int(media_progress.current_time * 1000)
405        mass_audiobook.fully_played = media_progress.is_finished
406
407    mass_audiobook.date_added = from_utc_timestamp(abs_audiobook.added_at / 1000)
408
409    return mass_audiobook
410
411
412def parse_author(
413    *,
414    abs_author: AbsAuthorExpanded | AbsAuthorMinified,
415    instance_id: str,
416    domain: str,
417    token: str | None,
418    base_url: str,
419) -> MassArtist:
420    """Translate AbsAuthor to MassArtist."""
421    mass_artist = MassArtist(
422        item_id=abs_author.id_,
423        provider=instance_id,
424        name=abs_author.name,
425        sort_name=abs_author.name,
426        provider_mappings={
427            ProviderMapping(
428                item_id=abs_author.id_, provider_domain=domain, provider_instance=instance_id
429            )
430        },
431        artist_type=ArtistType.AUTHOR,
432    )
433    # cover
434    if (
435        isinstance(abs_author, AbsAuthorExpanded)
436        and abs_author.image_path is not None
437        and token is not None
438    ):
439        api_url = f"/api/authors/{abs_author.id_}/image?token={token}"
440        cover_url = f"{base_url}{api_url}"
441        mass_artist.metadata.images = UniqueList(
442            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
443        )
444    elif isinstance(abs_author, AbsAuthorExpanded) and abs_author.image_path is None:
445        mass_artist.metadata.images = UniqueList([])
446
447    return mass_artist
448
449
450def parse_narrator(
451    *,
452    abs_narrator: AbsNarrator | NarratorHelper,
453    instance_id: str,
454    domain: str,
455) -> MassArtist:
456    """Translate AbsNarrator to MassArtist."""
457    return MassArtist(
458        item_id=abs_narrator.id_,
459        provider=instance_id,
460        name=abs_narrator.name,
461        sort_name=abs_narrator.name,
462        provider_mappings={
463            ProviderMapping(
464                item_id=abs_narrator.id_, provider_domain=domain, provider_instance=instance_id
465            )
466        },
467        artist_type=ArtistType.NARRATOR,
468    )
469