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
257    # cover image
258    if token is not None and cover_path:
259        url_cover = _build_cover_url(
260            base_url=base_url,
261            item_id=prov_podcast_id,
262            token=token,
263            version=cover_version,
264        )
265        mass_episode.metadata.images = UniqueList(
266            [MediaItemImage(type=ImageType.THUMB, path=url_cover, provider=instance_id)]
267        )
268
269    if media_progress is not None and media_progress.current_time is not None:
270        mass_episode.resume_position_ms = int(media_progress.current_time * 1000)
271        mass_episode.fully_played = media_progress.is_finished
272
273    mass_episode.metadata.chapters = [
274        MediaItemChapter(
275            position=position,
276            name=chapter.title,
277            start=chapter.start,
278            end=chapter.end,
279        )
280        for position, chapter in enumerate(episode.chapters, 1)
281    ]
282
283    return mass_episode
284
285
286def parse_audiobook(
287    *,
288    abs_audiobook: AbsLibraryItemExpandedBook,
289    instance_id: str,
290    domain: str,
291    token: str | None,
292    base_url: str,
293    audiobook_narrators: set[AbsNarrator] | set[NarratorHelper],
294    media_progress: AbsMediaProgress | None = None,
295) -> MassAudiobook:
296    """Translate AbsBook to Mass Book."""
297    title = abs_audiobook.media.metadata.title
298    # Per API doc title may be None.
299    if title is None:
300        title = "UNKNOWN TITLE"
301    subtitle = abs_audiobook.media.metadata.subtitle
302    if subtitle is not None or subtitle:
303        title += f" | {subtitle}"
304    mass_audiobook = MassAudiobook(
305        item_id=abs_audiobook.id_,
306        provider=instance_id,
307        name=title,
308        duration=int(abs_audiobook.media.duration),
309        provider_mappings={
310            ProviderMapping(
311                item_id=abs_audiobook.id_,
312                provider_domain=domain,
313                provider_instance=instance_id,
314            )
315        },
316        publisher=abs_audiobook.media.metadata.publisher,
317    )
318    mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
319    mass_audiobook.metadata.languages = (
320        UniqueList([abs_audiobook.media.metadata.language])
321        if abs_audiobook.media.metadata.language is not None
322        else UniqueList([])
323    )
324
325    if abs_audiobook.media.metadata.published_date is not None:
326        with suppress(ValueError):
327            mass_audiobook.metadata.release_date = datetime.fromisoformat(
328                abs_audiobook.media.metadata.published_date
329            )
330    elif abs_audiobook.media.metadata.published_year is not None:
331        with suppress(ValueError):
332            # ruff: noqa: DTZ001 # ignore tzinfo, this is a fallback attempt
333            mass_audiobook.metadata.release_date = datetime(
334                year=int(abs_audiobook.media.metadata.published_year), month=1, day=1
335            )
336
337    book_series: list[MediaItemCollection] = []
338    for abs_series_sequence in abs_audiobook.media.metadata.series:
339        book_series.append(
340            MediaItemCollection(
341                title=abs_series_sequence.name, sequence=abs_series_sequence.sequence
342            )
343        )
344    mass_audiobook.metadata.collections = UniqueList(book_series)
345
346    mass_audiobook.metadata.genres = (
347        set(abs_audiobook.media.metadata.genres)
348        if abs_audiobook.media.metadata.genres is not None
349        else set()
350    )
351
352    mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
353
354    # cover
355    if token is not None and abs_audiobook.media.cover_path is not None:
356        cover_url = _build_cover_url(
357            base_url=base_url,
358            item_id=abs_audiobook.id_,
359            token=token,
360            version=abs_audiobook.updated_at,
361        )
362        mass_audiobook.metadata.images = UniqueList(
363            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
364        )
365    elif abs_audiobook.media.cover_path is None:
366        mass_audiobook.metadata.images = UniqueList([])
367
368    # expanded version
369    mass_audiobook.authors.set(
370        [
371            parse_author(
372                abs_author=author,
373                instance_id=instance_id,
374                domain=domain,
375                token=token,
376                base_url=base_url,
377            )
378            for author in abs_audiobook.media.metadata.authors
379        ]
380    )
381
382    mass_audiobook.narrators.set(
383        [
384            parse_narrator(abs_narrator=narrator, instance_id=instance_id, domain=domain)
385            for narrator in audiobook_narrators
386        ]
387    )
388
389    chapters = []
390    for idx, chapter in enumerate(abs_audiobook.media.chapters, 1):
391        chapters.append(
392            MediaItemChapter(
393                position=idx,
394                name=chapter.title,
395                start=chapter.start,
396                end=chapter.end,
397            )
398        )
399    mass_audiobook.metadata.chapters = chapters
400
401    if media_progress is not None and media_progress.current_time is not None:
402        mass_audiobook.resume_position_ms = int(media_progress.current_time * 1000)
403        mass_audiobook.fully_played = media_progress.is_finished
404
405    mass_audiobook.date_added = from_utc_timestamp(abs_audiobook.added_at / 1000)
406
407    return mass_audiobook
408
409
410def parse_author(
411    *,
412    abs_author: AbsAuthorExpanded | AbsAuthorMinified,
413    instance_id: str,
414    domain: str,
415    token: str | None,
416    base_url: str,
417) -> MassArtist:
418    """Translate AbsAuthor to MassArtist."""
419    mass_artist = MassArtist(
420        item_id=abs_author.id_,
421        provider=instance_id,
422        name=abs_author.name,
423        sort_name=abs_author.name,
424        provider_mappings={
425            ProviderMapping(
426                item_id=abs_author.id_, provider_domain=domain, provider_instance=instance_id
427            )
428        },
429        artist_type=ArtistType.AUTHOR,
430    )
431    # cover
432    if (
433        isinstance(abs_author, AbsAuthorExpanded)
434        and abs_author.image_path is not None
435        and token is not None
436    ):
437        api_url = f"/api/authors/{abs_author.id_}/image?token={token}"
438        cover_url = f"{base_url}{api_url}"
439        mass_artist.metadata.images = UniqueList(
440            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
441        )
442    elif isinstance(abs_author, AbsAuthorExpanded) and abs_author.image_path is None:
443        mass_artist.metadata.images = UniqueList([])
444
445    return mass_artist
446
447
448def parse_narrator(
449    *,
450    abs_narrator: AbsNarrator | NarratorHelper,
451    instance_id: str,
452    domain: str,
453) -> MassArtist:
454    """Translate AbsNarrator to MassArtist."""
455    return MassArtist(
456        item_id=abs_narrator.id_,
457        provider=instance_id,
458        name=abs_narrator.name,
459        sort_name=abs_narrator.name,
460        provider_mappings={
461            ProviderMapping(
462                item_id=abs_narrator.id_, provider_domain=domain, provider_instance=instance_id
463            )
464        },
465        artist_type=ArtistType.NARRATOR,
466    )
467