/
/
/
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:
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) -> MassPodcastEpisode:
106 """Translate ABSPodcastEpisode to MassPodcastEpisode.
107
108 For an episode the id is set to f"{podcast_id} {episode_id}".
109 ABS ids have no spaces, so we can split at a space to retrieve both
110 in other functions.
111
112 NOTE: We should always use a PodcastEpisodeExpanded when possible.
113 A PodcastEpisode has only limited information, and is currently only used
114 within the recommendations.
115 """
116 episode_id = f"{prov_podcast_id} {episode.id_}"
117
118 if isinstance(episode, AbsPodcastEpisodeExpanded):
119 url = f"{base_url}{episode.audio_track.content_url}"
120 duration = int(episode.duration)
121 provider_mappings = {
122 ProviderMapping(
123 item_id=episode_id,
124 provider_domain=domain,
125 provider_instance=instance_id,
126 audio_format=AudioFormat(
127 content_type=ContentType.UNKNOWN,
128 ),
129 url=url,
130 )
131 }
132 else:
133 # PodcastEpisode
134 duration = 0 # mass default
135 provider_mappings = {
136 ProviderMapping(
137 item_id=episode_id,
138 provider_domain=domain,
139 provider_instance=instance_id,
140 )
141 }
142
143 release_date: datetime | None = None
144 if episode.published_at is not None:
145 position = -episode.published_at
146 # abs published_at is ms epoch
147 release_date = datetime.fromtimestamp(episode.published_at / 1000)
148 else:
149 position = 0
150 if fallback_episode_cnt is not None:
151 position = fallback_episode_cnt
152 mass_episode = MassPodcastEpisode(
153 item_id=episode_id,
154 provider=instance_id,
155 name=episode.title,
156 duration=duration,
157 position=position,
158 podcast=ItemMapping(
159 item_id=prov_podcast_id,
160 provider=instance_id,
161 name=episode.title,
162 media_type=MediaType.PODCAST,
163 ),
164 provider_mappings=provider_mappings,
165 )
166
167 mass_episode.metadata.release_date = release_date
168
169 # cover image
170 if token is not None:
171 url_api = f"/api/items/{prov_podcast_id}/cover?token={token}"
172 url_cover = f"{base_url}{url_api}"
173 mass_episode.metadata.images = UniqueList(
174 [MediaItemImage(type=ImageType.THUMB, path=url_cover, provider=instance_id)]
175 )
176
177 if media_progress is not None and media_progress.current_time is not None:
178 mass_episode.resume_position_ms = int(media_progress.current_time * 1000)
179 mass_episode.fully_played = media_progress.is_finished
180
181 return mass_episode
182
183
184def parse_audiobook(
185 *,
186 abs_audiobook: AbsLibraryItemExpandedBook | AbsLibraryItemMinifiedBook,
187 instance_id: str,
188 domain: str,
189 token: str | None,
190 base_url: str,
191 media_progress: AbsMediaProgress | None = None,
192) -> MassAudiobook:
193 """Translate AbsBook to Mass Book."""
194 title = abs_audiobook.media.metadata.title
195 # Per API doc title may be None.
196 if title is None:
197 title = "UNKNOWN TITLE"
198 subtitle = abs_audiobook.media.metadata.subtitle
199 if subtitle is not None or subtitle:
200 title += f" | {subtitle}"
201 mass_audiobook = MassAudiobook(
202 item_id=abs_audiobook.id_,
203 provider=instance_id,
204 name=title,
205 duration=int(abs_audiobook.media.duration),
206 provider_mappings={
207 ProviderMapping(
208 item_id=abs_audiobook.id_,
209 provider_domain=domain,
210 provider_instance=instance_id,
211 )
212 },
213 publisher=abs_audiobook.media.metadata.publisher,
214 )
215 mass_audiobook.metadata.description = abs_audiobook.media.metadata.description
216 if abs_audiobook.media.metadata.language is not None:
217 mass_audiobook.metadata.languages = UniqueList([abs_audiobook.media.metadata.language])
218
219 if abs_audiobook.media.metadata.published_date is not None:
220 with suppress(ValueError):
221 mass_audiobook.metadata.release_date = datetime.fromisoformat(
222 abs_audiobook.media.metadata.published_date
223 )
224 elif abs_audiobook.media.metadata.published_year is not None:
225 with suppress(ValueError):
226 # ruff: noqa: DTZ001 # ignore tzinfo, this is a fallback attempt
227 mass_audiobook.metadata.release_date = datetime(
228 year=int(abs_audiobook.media.metadata.published_year), month=1, day=1
229 )
230
231 if abs_audiobook.media.metadata.genres is not None:
232 mass_audiobook.metadata.genres = set(abs_audiobook.media.metadata.genres)
233
234 mass_audiobook.metadata.explicit = abs_audiobook.media.metadata.explicit
235
236 # cover
237 if token is not None:
238 api_url = f"/api/items/{abs_audiobook.id_}/cover?token={token}"
239 cover_url = f"{base_url}{api_url}"
240 mass_audiobook.metadata.images = UniqueList(
241 [MediaItemImage(type=ImageType.THUMB, path=cover_url, provider=instance_id)]
242 )
243
244 # expanded version
245 if isinstance(abs_audiobook, AbsLibraryItemExpandedBook):
246 mass_audiobook.authors.set([x.name for x in abs_audiobook.media.metadata.authors])
247 mass_audiobook.narrators.set(abs_audiobook.media.metadata.narrators)
248 chapters = []
249 for idx, chapter in enumerate(abs_audiobook.media.chapters, 1):
250 chapters.append(
251 MediaItemChapter(
252 position=idx,
253 name=chapter.title,
254 start=chapter.start,
255 end=chapter.end,
256 )
257 )
258 mass_audiobook.metadata.chapters = chapters
259
260 elif isinstance(abs_audiobook, AbsLibraryItemMinifiedBook):
261 mass_audiobook.authors.set([abs_audiobook.media.metadata.author_name])
262 mass_audiobook.narrators.set([abs_audiobook.media.metadata.narrator_name])
263
264 if media_progress is not None and media_progress.current_time is not None:
265 mass_audiobook.resume_position_ms = int(media_progress.current_time * 1000)
266 mass_audiobook.fully_played = media_progress.is_finished
267
268 mass_audiobook.date_added = datetime.fromtimestamp(abs_audiobook.added_at / 1000)
269
270 return mass_audiobook
271