music-assistant-server

1"""Parser for ABS -> MASS."""
2
3from contextlib import suppress
4from datetime import datetime
5
6from aioaudiobookshelf.schema.library import (
7    LibraryItemExpandedBook as AbsLibraryItemExpandedBook,
8)
9from aioaudiobookshelf.schema.library import (
10    LibraryItemExpandedPodcast as AbsLibraryItemExpandedPodcast,
11)
12from aioaudiobookshelf.schema.library import (
13    LibraryItemMinifiedBook as AbsLibraryItemMinifiedBook,
14)
15from aioaudiobookshelf.schema.library import (
16    LibraryItemMinifiedPodcast as AbsLibraryItemMinifiedPodcast,
17)
18from aioaudiobookshelf.schema.library import (
19    LibraryItemPodcast as AbsLibraryItemPodcast,
20)
21from aioaudiobookshelf.schema.media_progress import MediaProgress as AbsMediaProgress
22from aioaudiobookshelf.schema.podcast import PodcastEpisode as AbsPodcastEpisode
23from aioaudiobookshelf.schema.podcast import (
24    PodcastEpisodeExpanded as AbsPodcastEpisodeExpanded,
25)
26from music_assistant_models.enums import ContentType, ImageType, MediaType
27from music_assistant_models.media_items import Audiobook as MassAudiobook
28from music_assistant_models.media_items import (
29    AudioFormat,
30    ItemMapping,
31    MediaItemChapter,
32    MediaItemImage,
33    ProviderMapping,
34    UniqueList,
35)
36from music_assistant_models.media_items import Podcast as MassPodcast
37from music_assistant_models.media_items import PodcastEpisode as MassPodcastEpisode
38
39
40def parse_podcast(
41    *,
42    abs_podcast: AbsLibraryItemExpandedPodcast
43    | AbsLibraryItemMinifiedPodcast
44    | AbsLibraryItemPodcast,
45    instance_id: str,
46    domain: str,
47    token: str | None,
48    base_url: str,
49) -> MassPodcast:
50    """Translate ABSPodcast to MassPodcast."""
51    title = abs_podcast.media.metadata.title
52    # Per API doc title may be None.
53    if title is None:
54        title = "UNKNOWN"
55    mass_podcast = MassPodcast(
56        item_id=abs_podcast.id_,
57        name=title,
58        publisher=abs_podcast.media.metadata.author,
59        provider=instance_id,
60        provider_mappings={
61            ProviderMapping(
62                item_id=abs_podcast.id_,
63                provider_domain=domain,
64                provider_instance=instance_id,
65            )
66        },
67    )
68    mass_podcast.metadata.description = abs_podcast.media.metadata.description
69    if token is not None and abs_podcast.media.cover_path is not None:
70        image_url = f"{base_url}/api/items/{abs_podcast.id_}/cover?token={token}"
71        mass_podcast.metadata.images = UniqueList(
72            [MediaItemImage(type=ImageType.THUMB, path=image_url, provider=instance_id)]
73        )
74    mass_podcast.metadata.explicit = abs_podcast.media.metadata.explicit
75    if abs_podcast.media.metadata.language is not None:
76        mass_podcast.metadata.languages = UniqueList([abs_podcast.media.metadata.language])
77    if abs_podcast.media.metadata.genres is not None:
78        mass_podcast.metadata.genres = set(abs_podcast.media.metadata.genres)
79
80    # podcast object has no published_at int, but an iso string
81    if abs_podcast.media.metadata.release_date is not None:
82        with suppress(ValueError):
83            mass_podcast.metadata.release_date = datetime.fromisoformat(
84                abs_podcast.media.metadata.release_date
85            )
86
87    if isinstance(abs_podcast, AbsLibraryItemExpandedPodcast | AbsLibraryItemPodcast):
88        mass_podcast.total_episodes = len(abs_podcast.media.episodes)
89    elif isinstance(abs_podcast, AbsLibraryItemMinifiedPodcast):
90        mass_podcast.total_episodes = abs_podcast.media.num_episodes
91
92    return mass_podcast
93
94
95def parse_podcast_episode(
96    *,
97    episode: AbsPodcastEpisode | AbsPodcastEpisodeExpanded,
98    prov_podcast_id: str,
99    fallback_episode_cnt: int | None = None,
100    instance_id: str,
101    domain: str,
102    token: str | None,
103    base_url: str,
104    media_progress: AbsMediaProgress | None = None,
105    add_cover: bool = False,
106) -> MassPodcastEpisode:
107    """Translate ABSPodcastEpisode to MassPodcastEpisode.
108
109    For an episode the id is set to f"{podcast_id} {episode_id}".
110    ABS ids have no spaces, so we can split at a space to retrieve both
111    in other functions.
112
113    NOTE: We should always use a PodcastEpisodeExpanded when possible.
114    A PodcastEpisode has only limited information, and is currently only used
115    within the recommendations.
116    """
117    episode_id = f"{prov_podcast_id} {episode.id_}"
118
119    if isinstance(episode, AbsPodcastEpisodeExpanded):
120        url = f"{base_url}{episode.audio_track.content_url}"
121        duration = int(episode.duration)
122        provider_mappings = {
123            ProviderMapping(
124                item_id=episode_id,
125                provider_domain=domain,
126                provider_instance=instance_id,
127                audio_format=AudioFormat(
128                    content_type=ContentType.UNKNOWN,
129                ),
130                url=url,
131            )
132        }
133    else:
134        # PodcastEpisode
135        duration = 0  # mass default
136        provider_mappings = {
137            ProviderMapping(
138                item_id=episode_id,
139                provider_domain=domain,
140                provider_instance=instance_id,
141            )
142        }
143
144    release_date: datetime | None = None
145    if episode.published_at is not None:
146        position = -episode.published_at
147        # abs published_at is ms epoch
148        release_date = datetime.fromtimestamp(episode.published_at / 1000)
149    else:
150        position = 0
151        if fallback_episode_cnt is not None:
152            position = fallback_episode_cnt
153    mass_episode = MassPodcastEpisode(
154        item_id=episode_id,
155        provider=instance_id,
156        name=episode.title,
157        duration=duration,
158        position=position,
159        podcast=ItemMapping(
160            item_id=prov_podcast_id,
161            provider=instance_id,
162            name=episode.title,
163            media_type=MediaType.PODCAST,
164        ),
165        provider_mappings=provider_mappings,
166    )
167
168    mass_episode.metadata.release_date = release_date
169
170    # cover image
171    if token is not None and add_cover:
172        url_api = f"/api/items/{prov_podcast_id}/cover?token={token}"
173        url_cover = f"{base_url}{url_api}"
174        mass_episode.metadata.images = UniqueList(
175            [MediaItemImage(type=ImageType.THUMB, path=url_cover, provider=instance_id)]
176        )
177
178    if media_progress is not None and media_progress.current_time is not None:
179        mass_episode.resume_position_ms = int(media_progress.current_time * 1000)
180        mass_episode.fully_played = media_progress.is_finished
181
182    return mass_episode
183
184
185def parse_audiobook(
186    *,
187    abs_audiobook: AbsLibraryItemExpandedBook | AbsLibraryItemMinifiedBook,
188    instance_id: str,
189    domain: str,
190    token: str | None,
191    base_url: str,
192    media_progress: AbsMediaProgress | None = None,
193) -> MassAudiobook:
194    """Translate AbsBook to Mass Book."""
195    title = abs_audiobook.media.metadata.title
196    # Per API doc title may be None.
197    if title is None:
198        title = "UNKNOWN TITLE"
199    subtitle = abs_audiobook.media.metadata.subtitle
200    if subtitle is not None or subtitle:
201        title += f" | {subtitle}"
202    mass_audiobook = MassAudiobook(
203        item_id=abs_audiobook.id_,
204        provider=instance_id,
205        name=title,
206        duration=int(abs_audiobook.media.duration),
207        provider_mappings={
208            ProviderMapping(
209                item_id=abs_audiobook.id_,
210                provider_domain=domain,
211                provider_instance=instance_id,
212            )
213        },
214        publisher=abs_audiobook.media.metadata.publisher,
215    )
216    mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
217    if abs_audiobook.media.metadata.language is not None:
218        mass_audiobook.metadata.languages = UniqueList([abs_audiobook.media.metadata.language])
219
220    if abs_audiobook.media.metadata.published_date is not None:
221        with suppress(ValueError):
222            mass_audiobook.metadata.release_date = datetime.fromisoformat(
223                abs_audiobook.media.metadata.published_date
224            )
225    elif abs_audiobook.media.metadata.published_year is not None:
226        with suppress(ValueError):
227            # ruff: noqa: DTZ001 # ignore tzinfo, this is a fallback attempt
228            mass_audiobook.metadata.release_date = datetime(
229                year=int(abs_audiobook.media.metadata.published_year), month=1, day=1
230            )
231
232    if abs_audiobook.media.metadata.genres is not None:
233        mass_audiobook.metadata.genres = set(abs_audiobook.media.metadata.genres)
234
235    mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
236
237    # cover
238    if token is not None and abs_audiobook.media.cover_path is not None:
239        api_url = f"/api/items/{abs_audiobook.id_}/cover?token={token}"
240        cover_url = f"{base_url}{api_url}"
241        mass_audiobook.metadata.images = UniqueList(
242            [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
243        )
244
245    # expanded version
246    if isinstance(abs_audiobook, AbsLibraryItemExpandedBook):
247        mass_audiobook.authors.set([x.name for x in abs_audiobook.media.metadata.authors])
248        mass_audiobook.narrators.set(abs_audiobook.media.metadata.narrators)
249        chapters = []
250        for idx, chapter in enumerate(abs_audiobook.media.chapters, 1):
251            chapters.append(
252                MediaItemChapter(
253                    position=idx,
254                    name=chapter.title,
255                    start=chapter.start,
256                    end=chapter.end,
257                )
258            )
259        mass_audiobook.metadata.chapters = chapters
260
261    elif isinstance(abs_audiobook, AbsLibraryItemMinifiedBook):
262        mass_audiobook.authors.set([abs_audiobook.media.metadata.author_name])
263        mass_audiobook.narrators.set([abs_audiobook.media.metadata.narrator_name])
264
265    if media_progress is not None and media_progress.current_time is not None:
266        mass_audiobook.resume_position_ms = int(media_progress.current_time * 1000)
267        mass_audiobook.fully_played = media_progress.is_finished
268
269    mass_audiobook.date_added = datetime.fromtimestamp(abs_audiobook.added_at / 1000)
270
271    return mass_audiobook
272