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    return mass_playlist
104
105
106def parse_podcast(
107    *,
108    abs_podcast: AbsLibraryItemExpandedPodcast
109    | AbsLibraryItemMinifiedPodcast
110    | AbsLibraryItemPodcast,
111    instance_id: str,
112    domain: str,
113    token: str | None,
114    base_url: str,
115) -> MassPodcast:
116    """Translate ABSPodcast to MassPodcast."""
117    title = abs_podcast.media.metadata.title
118    # Per API doc title may be None.
119    if title is None:
120        title = "UNKNOWN"
121    mass_podcast = MassPodcast(
122        item_id=abs_podcast.id_,
123        name=title,
124        publisher=abs_podcast.media.metadata.author,
125        provider=instance_id,
126        provider_mappings={
127            ProviderMapping(
128                item_id=abs_podcast.id_,
129                provider_domain=domain,
130                provider_instance=instance_id,
131            )
132        },
133    )
134    mass_podcast.metadata.description = abs_podcast.media.metadata.description
135    if token is not None and abs_podcast.media.cover_path is not None:
136        image_url = _build_cover_url(
137            base_url=base_url,
138            item_id=abs_podcast.id_,
139            token=token,
140            version=abs_podcast.updated_at,
141        )
142        mass_podcast.metadata.images = UniqueList(
143            [MediaItemImage(type=ImageType.THUMB, path=image_url, provider=instance_id)]
144        )
145    mass_podcast.metadata.explicit = abs_podcast.media.metadata.explicit
146    if abs_podcast.media.metadata.language is not None:
147        mass_podcast.metadata.languages = UniqueList([abs_podcast.media.metadata.language])
148    if abs_podcast.media.metadata.genres is not None:
149        mass_podcast.metadata.genres = set(abs_podcast.media.metadata.genres)
150
151    # podcast object has no published_at int, but an iso string
152    if abs_podcast.media.metadata.release_date is not None:
153        with suppress(ValueError):
154            mass_podcast.metadata.release_date = datetime.fromisoformat(
155                abs_podcast.media.metadata.release_date
156            )
157
158    if isinstance(abs_podcast, AbsLibraryItemExpandedPodcast | AbsLibraryItemPodcast):
159        mass_podcast.total_episodes = len(abs_podcast.media.episodes)
160    elif isinstance(abs_podcast, AbsLibraryItemMinifiedPodcast):
161        mass_podcast.total_episodes = abs_podcast.media.num_episodes
162
163    return mass_podcast
164
165
166def parse_podcast_episode(
167    *,
168    episode: AbsPodcastEpisode | AbsPodcastEpisodeExpanded,
169    prov_podcast_id: str,
170    prov_podcast_name: str | None,
171    fallback_episode_cnt: int | None = None,
172    instance_id: str,
173    domain: str,
174    token: str | None,
175    base_url: str,
176    media_progress: AbsMediaProgress | None = None,
177    cover_path: str | None = None,
178    cover_version: int | None = None,
179) -> MassPodcastEpisode:
180    """
181    Translate ABSPodcastEpisode to MassPodcastEpisode.
182
183    For an episode the id is set to f"{podcast_id} {episode_id}".
184    ABS ids have no spaces, so we can split at a space to retrieve both
185    in other functions.
186
187    NOTE: We should always use a PodcastEpisodeExpanded when possible.
188    A PodcastEpisode has only limited information, and is currently only used
189    within the recommendations.
190    """
191    # ruff: noqa: PLR0913 (too many arguments)
192    episode_id = f"{prov_podcast_id} {episode.id_}"
193
194    if isinstance(episode, AbsPodcastEpisodeExpanded):
195        url = f"{base_url}{episode.audio_track.content_url}"
196        duration = int(episode.duration)
197        provider_mappings = {
198            ProviderMapping(
199                item_id=episode_id,
200                provider_domain=domain,
201                provider_instance=instance_id,
202                audio_format=AudioFormat(
203                    content_type=ContentType.UNKNOWN,
204                ),
205                url=url,
206            )
207        }
208    else:
209        # PodcastEpisode
210        duration = 0  # mass default
211        provider_mappings = {
212            ProviderMapping(
213                item_id=episode_id,
214                provider_domain=domain,
215                provider_instance=instance_id,
216            )
217        }
218
219    release_date: datetime | None = None
220    if episode.published_at is not None:
221        position = -episode.published_at
222        # abs published_at is ms epoch; leave the date unset if it is out of range
223        with suppress(ValueError, OverflowError, OSError):
224            release_date = from_utc_timestamp(episode.published_at / 1000)
225    else:
226        position = 0
227        if fallback_episode_cnt is not None:
228            position = fallback_episode_cnt
229    mass_episode = MassPodcastEpisode(
230        item_id=episode_id,
231        provider=instance_id,
232        name=episode.title,
233        duration=duration,
234        position=position,
235        podcast=ItemMapping(
236            item_id=prov_podcast_id,
237            provider=instance_id,
238            name=prov_podcast_name or episode.title,
239            media_type=MediaType.PODCAST,
240        ),
241        provider_mappings=provider_mappings,
242    )
243
244    mass_episode.metadata.release_date = release_date
245
246    # cover image
247    if token is not None and cover_path:
248        url_cover = _build_cover_url(
249            base_url=base_url,
250            item_id=prov_podcast_id,
251            token=token,
252            version=cover_version,
253        )
254        mass_episode.metadata.images = UniqueList(
255            [MediaItemImage(type=ImageType.THUMB, path=url_cover, provider=instance_id)]
256        )
257
258    if media_progress is not None and media_progress.current_time is not None:
259        mass_episode.resume_position_ms = int(media_progress.current_time * 1000)
260        mass_episode.fully_played = media_progress.is_finished
261
262    if episode.chapters:
263        mass_episode.metadata.chapters = [
264            MediaItemChapter(
265                position=position,
266                name=chapter.title,
267                start=chapter.start,
268                end=chapter.end,
269            )
270            for position, chapter in enumerate(episode.chapters, 1)
271        ]
272
273    return mass_episode
274
275
276def parse_audiobook(
277    *,
278    abs_audiobook: AbsLibraryItemExpandedBook,
279    instance_id: str,
280    domain: str,
281    token: str | None,
282    base_url: str,
283    audiobook_narrators: set[AbsNarrator] | set[NarratorHelper],
284    media_progress: AbsMediaProgress | None = None,
285) -> MassAudiobook:
286    """Translate AbsBook to Mass Book."""
287    title = abs_audiobook.media.metadata.title
288    # Per API doc title may be None.
289    if title is None:
290        title = "UNKNOWN TITLE"
291    subtitle = abs_audiobook.media.metadata.subtitle
292    if subtitle is not None or subtitle:
293        title += f" | {subtitle}"
294    mass_audiobook = MassAudiobook(
295        item_id=abs_audiobook.id_,
296        provider=instance_id,
297        name=title,
298        duration=int(abs_audiobook.media.duration),
299        provider_mappings={
300            ProviderMapping(
301                item_id=abs_audiobook.id_,
302                provider_domain=domain,
303                provider_instance=instance_id,
304            )
305        },
306        publisher=abs_audiobook.media.metadata.publisher,
307    )
308    mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
309    if abs_audiobook.media.metadata.language is not None:
310        mass_audiobook.metadata.languages = UniqueList([abs_audiobook.media.metadata.language])
311
312    if abs_audiobook.media.metadata.published_date is not None:
313        with suppress(ValueError):
314            mass_audiobook.metadata.release_date = datetime.fromisoformat(
315                abs_audiobook.media.metadata.published_date
316            )
317    elif abs_audiobook.media.metadata.published_year is not None:
318        with suppress(ValueError):
319            # ruff: noqa: DTZ001 # ignore tzinfo, this is a fallback attempt
320            mass_audiobook.metadata.release_date = datetime(
321                year=int(abs_audiobook.media.metadata.published_year), month=1, day=1
322            )
323
324    book_series: list[MediaItemCollection] = []
325    for abs_series_sequence in abs_audiobook.media.metadata.series:
326        book_series.append(
327            MediaItemCollection(
328                title=abs_series_sequence.name, sequence=abs_series_sequence.sequence
329            )
330        )
331
332    if book_series:
333        mass_audiobook.metadata.collections = UniqueList(book_series)
334
335    if abs_audiobook.media.metadata.genres is not None:
336        mass_audiobook.metadata.genres = set(abs_audiobook.media.metadata.genres)
337
338    mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
339
340    # cover
341    if token is not None and abs_audiobook.media.cover_path is not None:
342        cover_url = _build_cover_url(
343            base_url=base_url,
344            item_id=abs_audiobook.id_,
345            token=token,
346            version=abs_audiobook.updated_at,
347        )
348        mass_audiobook.metadata.images = UniqueList(
349            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
350        )
351
352    # expanded version
353    mass_audiobook.authors.set(
354        [
355            parse_author(
356                abs_author=author,
357                instance_id=instance_id,
358                domain=domain,
359                token=token,
360                base_url=base_url,
361            )
362            for author in abs_audiobook.media.metadata.authors
363        ]
364    )
365
366    mass_audiobook.narrators.set(
367        [
368            parse_narrator(abs_narrator=narrator, instance_id=instance_id, domain=domain)
369            for narrator in audiobook_narrators
370        ]
371    )
372
373    chapters = []
374    for idx, chapter in enumerate(abs_audiobook.media.chapters, 1):
375        chapters.append(
376            MediaItemChapter(
377                position=idx,
378                name=chapter.title,
379                start=chapter.start,
380                end=chapter.end,
381            )
382        )
383    mass_audiobook.metadata.chapters = chapters
384
385    if media_progress is not None and media_progress.current_time is not None:
386        mass_audiobook.resume_position_ms = int(media_progress.current_time * 1000)
387        mass_audiobook.fully_played = media_progress.is_finished
388
389    mass_audiobook.date_added = from_utc_timestamp(abs_audiobook.added_at / 1000)
390
391    return mass_audiobook
392
393
394def parse_author(
395    *,
396    abs_author: AbsAuthorExpanded | AbsAuthorMinified,
397    instance_id: str,
398    domain: str,
399    token: str | None,
400    base_url: str,
401) -> MassArtist:
402    """Translate AbsAuthor to MassArtist."""
403    mass_artist = MassArtist(
404        item_id=abs_author.id_,
405        provider=instance_id,
406        name=abs_author.name,
407        sort_name=abs_author.name,
408        provider_mappings={
409            ProviderMapping(
410                item_id=abs_author.id_, provider_domain=domain, provider_instance=instance_id
411            )
412        },
413        artist_type=ArtistType.AUTHOR,
414    )
415    # cover
416    if (
417        isinstance(abs_author, AbsAuthorExpanded)
418        and abs_author.image_path is not None
419        and token is not None
420    ):
421        api_url = f"/api/authors/{abs_author.id_}/image?token={token}"
422        cover_url = f"{base_url}{api_url}"
423        mass_artist.metadata.images = UniqueList(
424            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
425        )
426    return mass_artist
427
428
429def parse_narrator(
430    *,
431    abs_narrator: AbsNarrator | NarratorHelper,
432    instance_id: str,
433    domain: str,
434) -> MassArtist:
435    """Translate AbsNarrator to MassArtist."""
436    return MassArtist(
437        item_id=abs_narrator.id_,
438        provider=instance_id,
439        name=abs_narrator.name,
440        sort_name=abs_narrator.name,
441        provider_mappings={
442            ProviderMapping(
443                item_id=abs_narrator.id_, provider_domain=domain, provider_instance=instance_id
444            )
445        },
446        artist_type=ArtistType.NARRATOR,
447    )
448