/
/
/
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