/
/
/
1"""Manage MediaItems of type Podcast."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING, Any, cast
7
8from music_assistant_models.auth import Scope
9from music_assistant_models.enums import MediaType, ProviderFeature
10from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
11from music_assistant_models.helpers import create_safe_string
12from music_assistant_models.media_items import (
13 Podcast,
14 PodcastEpisode,
15 PodcastSummary,
16 ProviderMapping,
17 UniqueList,
18)
19
20from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PODCASTS
21from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
22from music_assistant.helpers.audio import get_probed_duration
23from music_assistant.helpers.compare import (
24 compare_media_item,
25 compare_podcast,
26 loose_compare_strings,
27)
28from music_assistant.helpers.database import UNSET
29from music_assistant.helpers.json import serialize_to_json
30from music_assistant.models.music_provider import MusicProvider
31
32from .base import MediaControllerBase
33
34if TYPE_CHECKING:
35 from collections.abc import Mapping
36
37 from music_assistant_models.auth import User
38
39 from music_assistant import MusicAssistant
40
41
42class PodcastsController(MediaControllerBase[Podcast]):
43 """Controller managing MediaItems of type Podcast."""
44
45 db_table = DB_TABLE_PODCASTS
46 media_type = MediaType.PODCAST
47 item_cls = Podcast
48 summary_item_cls = PodcastSummary
49
50 def __init__(self, mass: MusicAssistant) -> None:
51 """Initialize class."""
52 super().__init__(mass)
53 # register (extra) api handlers
54 api_base = self.api_base
55 self.mass.register_api_command(
56 f"music/{api_base}/podcast_episodes", self.episodes, required_scope=Scope.LIBRARY_READ
57 )
58 self.mass.register_api_command(
59 f"music/{api_base}/podcast_episode", self.episode, required_scope=Scope.LIBRARY_READ
60 )
61 self.mass.register_api_command(
62 f"music/{api_base}/podcast_versions", self.versions, required_scope=Scope.LIBRARY_READ
63 )
64
65 @property
66 def summary_query(self) -> tuple[str, dict[str, Any]]:
67 """Return the slim SELECT query used for podcast summary listings."""
68 query = f"""
69 SELECT
70 {self._summary_base_columns()},
71 podcasts.version,
72 podcasts.publisher,
73 podcasts.total_episodes,
74 {self._provider_mappings_query()} AS provider_mappings
75 FROM podcasts"""
76 return query, {}
77
78 async def library_items(
79 self,
80 favorite: bool | None = None,
81 search: str | None = None,
82 limit: int = 500,
83 offset: int = 0,
84 order_by: str = "sort_name",
85 provider: str | list[str] | None = None,
86 genre: int | list[int] | None = None,
87 played_only: bool = False,
88 *,
89 summary: bool = True,
90 reachable_via: list[str] | None = None,
91 **kwargs: Any,
92 ) -> list[Podcast]:
93 """
94 Get in-database podcasts.
95
96 :param favorite: Filter by favorite status.
97 :param search: Filter by search query.
98 :param limit: Maximum number of items to return.
99 :param offset: Number of items to skip.
100 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
101 :param provider: Filter by provider instance ID (single string or list).
102 :param genre: Filter by genre id(s).
103 :param summary: When True (default), return slim summary items containing only the
104 fields needed for a list view. Set to False to get fully hydrated items.
105 :param reachable_via: Restrict results to items with a provider mapping reachable
106 through one of these provider instance ids (OR semantics). See
107 `MediaControllerBase.library_items` for the full semantics.
108 """
109 reachable_via = self._resolve_reachable_via(reachable_via)
110 if reachable_via is not None and not reachable_via:
111 return []
112 result = await self.get_library_items_by_query(
113 favorite=favorite,
114 search=search,
115 genre_ids=genre,
116 limit=limit,
117 offset=offset,
118 order_by=order_by,
119 provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
120 played_only=played_only,
121 in_library_only=True,
122 summary=summary,
123 reachable_via=reachable_via,
124 )
125 if search and len(result) < 25 and not offset:
126 # append publisher items to result
127 extra_query_parts: list[str] = [
128 "WHERE podcasts.publisher LIKE :search",
129 ]
130 extra_query_params: dict[str, Any] = {
131 "search": f"%{search}%",
132 }
133 return result + await self.get_library_items_by_query(
134 favorite=favorite,
135 search=None,
136 genre_ids=genre,
137 limit=limit,
138 order_by=order_by,
139 provider_filter=self._provider_filter_considering_reachability(
140 provider, reachable_via
141 ),
142 extra_query_parts=extra_query_parts,
143 extra_query_params=extra_query_params,
144 in_library_only=True,
145 summary=summary,
146 reachable_via=reachable_via,
147 )
148 return result
149
150 async def episodes(
151 self,
152 item_id: str,
153 provider_instance_id_or_domain: str,
154 ) -> AsyncGenerator[PodcastEpisode]:
155 """Return podcast episodes for the given provider podcast id."""
156 # always check if we have a library item for this podcast
157 if provider_instance_id_or_domain == "library":
158 library_podcast = await self.get_library_item(item_id)
159 if not library_podcast:
160 raise MediaNotFoundError(f"Podcast {item_id} not found in library")
161 provider_instance_id_or_domain, item_id = self._select_provider_id(library_podcast)
162 # podcast episodes are not stored in the db/library
163 # so we always need to fetch them from the provider
164 async for episode in self._get_provider_podcast_episodes(
165 item_id, provider_instance_id_or_domain
166 ):
167 yield episode
168
169 async def episode(
170 self,
171 item_id: str,
172 provider_instance_id_or_domain: str,
173 ) -> PodcastEpisode:
174 """Return single podcast episode by the given provider podcast id."""
175 prov = self.mass.get_provider(provider_instance_id_or_domain)
176 if not isinstance(prov, MusicProvider):
177 raise ProviderUnavailableError("Provider not found")
178 episode = await prov.get_podcast_episode(item_id)
179 await self._restore_resume_position(episode, prov.instance_id)
180 await self._restore_probed_duration(episode)
181 return episode
182
183 async def versions(
184 self,
185 item_id: str,
186 provider_instance_id_or_domain: str,
187 ) -> UniqueList[Podcast]:
188 """Return all versions of an podcast we can find on all providers."""
189 podcast = await self.get_provider_item(item_id, provider_instance_id_or_domain)
190 search_query = podcast.name
191 result: UniqueList[Podcast] = UniqueList()
192 for provider_id in self.mass.music.get_unique_providers():
193 provider = self.mass.get_provider(provider_id)
194 if not isinstance(provider, MusicProvider):
195 continue
196 if MediaType.PODCAST not in provider.supported_media_types:
197 continue
198 result.extend(
199 prov_item
200 for prov_item in await self.search(search_query, provider_id)
201 if loose_compare_strings(podcast.name, prov_item.name)
202 # make sure that the 'base' version is NOT included
203 and not podcast.provider_mappings.intersection(prov_item.provider_mappings)
204 )
205 return result
206
207 async def match_provider(
208 self, db_podcast: Podcast, provider: MusicProvider, strict: bool = True
209 ) -> list[ProviderMapping]:
210 """
211 Try to find match on (streaming) provider for the provided (database) podcast.
212
213 This is used to link objects of different providers/qualities together.
214 """
215 self.logger.debug(
216 "Trying to match podcast %s on provider %s",
217 db_podcast.name,
218 provider.name,
219 )
220 matches: list[ProviderMapping] = []
221 search_str = db_podcast.name
222 search_result = await self.search(search_str, provider.instance_id)
223 for search_result_item in search_result:
224 if not search_result_item.available:
225 continue
226 if not compare_media_item(db_podcast, search_result_item, strict=strict):
227 continue
228 # we must fetch the full podcast version, search results can be simplified objects
229 prov_podcast = await self.get_provider_item(
230 search_result_item.item_id,
231 search_result_item.provider,
232 fallback=search_result_item,
233 )
234 if compare_podcast(db_podcast, prov_podcast, strict=strict):
235 # 100% match
236 matches.extend(prov_podcast.provider_mappings)
237 if not matches:
238 self.logger.debug(
239 "Could not find match for Podcast %s on provider %s",
240 db_podcast.name,
241 provider.name,
242 )
243 return matches
244
245 async def match_providers(self, db_podcast: Podcast) -> None:
246 """
247 Try to find match on all (streaming) providers for the provided (database) podcast.
248
249 This is used to link objects of different providers/qualities together.
250 """
251 if db_podcast.provider != "library":
252 return # Matching only supported for database items
253
254 # try to find match on all providers
255 cur_provider_domains = {x.provider_domain for x in db_podcast.provider_mappings}
256 for provider in self.mass.music.providers:
257 if provider.domain in cur_provider_domains:
258 continue
259 if ProviderFeature.SEARCH not in provider.supported_features:
260 continue
261 if MediaType.PODCAST not in provider.supported_media_types:
262 continue
263 if not provider.is_streaming_provider:
264 # matching on unique providers is pointless as they push (all) their content to MA
265 continue
266 if match := await self.match_provider(db_podcast, provider):
267 # 100% match, we update the db with the additional provider mapping(s)
268 await self.add_provider_mappings(db_podcast.item_id, match)
269 cur_provider_domains.add(provider.domain)
270
271 async def _add_library_item(self, item: Podcast, overwrite_existing: bool = False) -> int:
272 """Add a new record to the database."""
273 db_id = await self.mass.music.database.insert(
274 self.db_table,
275 {
276 "name": item.name,
277 "sort_name": item.sort_name,
278 "version": item.version,
279 "favorite": item.favorite,
280 "metadata": serialize_to_json(item.metadata),
281 "publisher": item.publisher,
282 "total_episodes": item.total_episodes or 0,
283 "search_name": create_safe_string(item.name, True, True),
284 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
285 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
286 },
287 )
288 # update/set external id lookup table
289 await self.set_external_ids(db_id, item.external_ids)
290 # update/set provider_mappings table
291 await self.set_provider_mappings(db_id, item.provider_mappings)
292 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
293 return db_id
294
295 async def _update_library_item(
296 self, item_id: str | int, update: Podcast, overwrite: bool = False
297 ) -> None:
298 """Update existing record in the database."""
299 db_id = int(item_id) # ensure integer
300 cur_item = await self.get_library_item(db_id)
301 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
302 if not overwrite and update.metadata.images is not None:
303 # podcasts have no image picker, so keep the cover in sync with the
304 # provider instead of accumulating merged entries
305 metadata.images = update.metadata.images
306 cur_item.external_ids.update(update.external_ids)
307 name = update.name if overwrite else cur_item.name
308 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
309 await self.mass.music.database.update(
310 self.db_table,
311 {"item_id": db_id},
312 {
313 "name": name,
314 "sort_name": sort_name,
315 "version": update.version if overwrite else cur_item.version or update.version,
316 "metadata": serialize_to_json(metadata),
317 "publisher": cur_item.publisher or update.publisher,
318 "total_episodes": cur_item.total_episodes or update.total_episodes or 0,
319 "search_name": create_safe_string(name, True, True),
320 "search_sort_name": create_safe_string(sort_name or "", True, True),
321 "timestamp_added": int(update.date_added.timestamp())
322 if update.date_added
323 else UNSET,
324 },
325 )
326 # update/set external id lookup table
327 await self.set_external_ids(
328 db_id, update.external_ids if overwrite else cur_item.external_ids
329 )
330 # update/set provider_mappings table
331 provider_mappings = (
332 update.provider_mappings
333 if overwrite
334 else {*update.provider_mappings, *cur_item.provider_mappings}
335 )
336 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
337 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
338
339 async def _get_provider_podcast_episodes(
340 self, item_id: str, provider_instance_id_or_domain: str
341 ) -> AsyncGenerator[PodcastEpisode]:
342 """Return podcast episodes for the given provider podcast id."""
343 prov = self.mass.get_provider(provider_instance_id_or_domain)
344 if not isinstance(prov, MusicProvider):
345 return
346
347 # Get user who initiated the query. Querying the userid as well is most useful
348 # in a multi-user environment where a single instance provider is used.
349 user: User | None = None
350 if session_user := get_current_user():
351 # this is the active session user that triggered the action
352 user = session_user
353 elif provider_user := await self.mass.music._get_user_for_provider(
354 provider_mappings_or_instance_id=provider_instance_id_or_domain
355 ):
356 # based on configured provider filter we can try to find a user
357 user = provider_user
358
359 # fetched in one query on first use instead of one per episode: a podcast can have
360 # thousands of them
361 resume_rows: dict[str, Mapping[str, Any]] | None = None
362
363 async def load_resume_rows() -> dict[str, Mapping[str, Any]]:
364 match: dict[str, Any] = {
365 "provider": prov.instance_id,
366 "media_type": MediaType.PODCAST_EPISODE,
367 }
368 if user is not None:
369 match["userid"] = user.user_id
370 # limit=0 lifts get_rows' 500 row default, which combined with the ascending sort
371 # would drop the newest rows - the part-played episodes this lookup is for. That
372 # sort also picks the newest row per item_id in the map below, where without a
373 # userid filter several users can hold one
374 rows = await self.mass.music.database.get_rows(
375 DB_TABLE_PLAYLOG, match=match, order_by="timestamp", limit=0
376 )
377 return {row["item_id"]: row for row in rows}
378
379 async def set_resume_position(episode: PodcastEpisode) -> None:
380 nonlocal resume_rows
381 if episode.fully_played is not None or episode.resume_position_ms:
382 # provider supports resume info, we can skip
383 return
384 # for providers that do not natively support providing resume info,
385 # we fallback to the playlog db table
386 if resume_rows is None:
387 resume_rows = await load_resume_rows()
388 resume_info_db_row = resume_rows.get(episode.item_id)
389 if resume_info_db_row is None:
390 return
391 if resume_info_db_row["seconds_played"]:
392 episode.resume_position_ms = int(resume_info_db_row["seconds_played"] * 1000)
393 if resume_info_db_row["fully_played"] is not None:
394 episode.fully_played = bool(resume_info_db_row["fully_played"])
395
396 # grab the episodes from the provider. Providers cache their own listing, so resume
397 # info is applied here to keep per-user progress out of those caches
398 async for item in prov.get_podcast_episodes(item_id):
399 await set_resume_position(item)
400 await self._restore_probed_duration(item)
401 yield item
402
403 async def _restore_probed_duration(self, episode: PodcastEpisode) -> None:
404 """
405 Fill in the duration determined during an earlier playback, for feeds that omit it.
406
407 :param episode: The episode to fill the duration of, left untouched when it has one.
408 """
409 if episode.duration or not (uri := episode.uri):
410 return
411 if probed_duration := await get_probed_duration(self.mass, uri):
412 episode.duration = probed_duration
413
414 async def _restore_resume_position(
415 self, episode: PodcastEpisode, provider_instance_id: str
416 ) -> None:
417 """
418 Fill in resume position from the playlog for a single episode.
419
420 Skipped when the episode already has resume info set by the provider.
421
422 :param episode: The episode to enrich with resume info.
423 :param provider_instance_id: The provider instance the episode belongs to.
424 """
425 if episode.fully_played is not None or episode.resume_position_ms:
426 return
427 user: User | None = None
428 if session_user := get_current_user():
429 user = session_user
430 elif provider_user := await self.mass.music._get_user_for_provider(
431 provider_mappings_or_instance_id=provider_instance_id
432 ):
433 user = provider_user
434 match: dict[str, Any] = {
435 "provider": provider_instance_id,
436 "media_type": MediaType.PODCAST_EPISODE,
437 "item_id": episode.item_id,
438 }
439 if user is not None:
440 match["userid"] = user.user_id
441 # without a userid filter several users can hold a row, the newest one wins
442 rows = await self.mass.music.database.get_rows(
443 DB_TABLE_PLAYLOG, match=match, order_by="timestamp DESC", limit=1
444 )
445 row = rows[0] if rows else None
446 if row is None:
447 return
448 if row["seconds_played"]:
449 episode.resume_position_ms = int(row["seconds_played"] * 1000)
450 if row["fully_played"] is not None:
451 episode.fully_played = bool(row["fully_played"])
452
453 def _parse_summary_row(self, db_row: Mapping[str, Any]) -> PodcastSummary:
454 """Parse a raw summary db row into a PodcastSummary object."""
455 item = cast("PodcastSummary", super()._parse_summary_row(db_row))
456 item.version = db_row["version"] or ""
457 item.publisher = db_row["publisher"]
458 item.total_episodes = db_row["total_episodes"]
459 return item
460