/
/
/
1"""Podcast Index provider implementation."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator, Sequence
6from typing import Any, cast
7
8import aiohttp
9from music_assistant_models.config_entries import ConfigEntry
10from music_assistant_models.enums import ConfigEntryType, ContentType, MediaType, StreamType
11from music_assistant_models.errors import (
12 InvalidDataError,
13 LoginFailed,
14 MediaNotFoundError,
15 ProviderUnavailableError,
16)
17from music_assistant_models.media_items import (
18 AudioFormat,
19 BrowseFolder,
20 MediaItemType,
21 Podcast,
22 PodcastEpisode,
23 SearchResults,
24)
25from music_assistant_models.streamdetails import StreamDetails
26
27from music_assistant.constants import VERBOSE_LOG_LEVEL
28from music_assistant.controllers.cache import use_cache
29from music_assistant.helpers.podcast_parsers import (
30 enrich_episode_chapters,
31 rank_episodes_by_date,
32)
33from music_assistant.models.music_provider import MusicProvider
34
35from .constants import (
36 BROWSE_CATEGORIES,
37 BROWSE_RECENT,
38 BROWSE_TRENDING,
39 CONF_API_KEY,
40 CONF_API_SECRET,
41 CONF_STORED_PODCASTS,
42)
43from .helpers import make_api_request, parse_episode_from_data, parse_podcast_from_feed
44
45
46class PodcastIndexProvider(MusicProvider):
47 """Podcast Index provider for Music Assistant."""
48
49 api_key: str = ""
50 api_secret: str = ""
51
52 @property
53 def max_concurrent_streams(self) -> None:
54 """Allow unlimited concurrent upstream source streams."""
55 return None
56
57 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
58 """Return Config entries to setup this provider."""
59 return (
60 ConfigEntry(
61 key=CONF_STORED_PODCASTS,
62 type=ConfigEntryType.STRING,
63 multi_value=True,
64 default_value=[],
65 required=False,
66 hidden=True,
67 ),
68 )
69
70 async def handle_async_init(self) -> None:
71 """Handle async initialization of the provider."""
72 self.api_key = str(self.get_setup_value(CONF_API_KEY))
73 self.api_secret = str(self.get_setup_value(CONF_API_SECRET))
74
75 if not self.api_key or not self.api_secret:
76 raise LoginFailed("API key and secret are required")
77
78 # Test API connection
79 try:
80 await self._api_request("stats/current")
81 except LoginFailed, ProviderUnavailableError:
82 # Re-raise these specific errors as they have proper context
83 raise
84 except aiohttp.ClientConnectorError as err:
85 raise ProviderUnavailableError(
86 f"Failed to connect to Podcast Index API: {err}"
87 ) from err
88 except aiohttp.ServerTimeoutError as err:
89 raise ProviderUnavailableError(f"Podcast Index API timeout: {err}") from err
90 except Exception as err:
91 raise LoginFailed(f"Failed to connect to API: {err}") from err
92
93 async def search(
94 self, search_query: str, media_types: list[MediaType], limit: int = 10
95 ) -> SearchResults:
96 """
97 Perform search on Podcast Index.
98
99 Searches for podcasts by term. Future enhancement could include
100 category search if needed.
101 """
102 result = SearchResults()
103 if MediaType.PODCAST not in media_types:
104 return result
105
106 response = await self._api_request(
107 "search/byterm", params={"q": search_query, "max": limit}
108 )
109
110 podcasts = []
111 for feed_data in response.get("feeds", []):
112 podcast = parse_podcast_from_feed(feed_data, self.instance_id, self.domain)
113 if podcast:
114 podcasts.append(podcast)
115
116 result.podcasts = podcasts
117 return result
118
119 async def browse(self, path: str) -> Sequence[BrowseFolder | Podcast | PodcastEpisode]:
120 """Browse this provider's items."""
121 base = f"{self.instance_id}://"
122
123 if path == base:
124 # Return main browse categories
125 return [
126 BrowseFolder(
127 item_id=BROWSE_TRENDING,
128 provider=self.domain,
129 path=f"{base}{BROWSE_TRENDING}",
130 name="Trending Podcasts",
131 translation_key="trending_podcasts",
132 ),
133 BrowseFolder(
134 item_id=BROWSE_RECENT,
135 provider=self.domain,
136 path=f"{base}{BROWSE_RECENT}",
137 name="Recent Episodes",
138 translation_key="recent_episodes",
139 ),
140 BrowseFolder(
141 item_id=BROWSE_CATEGORIES,
142 provider=self.domain,
143 path=f"{base}{BROWSE_CATEGORIES}",
144 name="Categories",
145 translation_key="categories",
146 ),
147 ]
148
149 # Parse path after base
150 if path.startswith(base):
151 subpath_parts = path[len(base) :].split("/")
152 subpath = subpath_parts[0] if subpath_parts else ""
153
154 if subpath == BROWSE_TRENDING:
155 return await self._browse_trending()
156 if subpath == BROWSE_RECENT:
157 return await self._browse_recent_episodes()
158 if subpath == BROWSE_CATEGORIES:
159 if len(subpath_parts) > 1:
160 # Browse specific category - category name is directly in path
161 category_name = subpath_parts[1]
162 return await self._browse_category_podcasts(category_name)
163 # Browse categories
164 return await self._browse_categories()
165
166 return []
167
168 async def library_add(self, item: MediaItemType) -> bool:
169 """
170 Add podcast to library.
171
172 Retrieves the RSS feed URL for the podcast and adds it to the stored
173 podcasts configuration. Returns True if successfully added, False if
174 the podcast was already in the library or if the feed URL couldn't be found.
175 """
176 # Only handle podcasts - delegate others to base class
177 if not isinstance(item, Podcast):
178 return await super().library_add(item)
179
180 # Get the RSS URL from the podcast via API
181 try:
182 feed_url = await self._get_feed_url_for_podcast(item.item_id)
183 except Exception as err:
184 self.logger.warning(
185 "Failed to retrieve feed URL for podcast %s: %s", item.name, err, exc_info=True
186 )
187 return False
188
189 if not feed_url:
190 self.logger.warning(
191 "No feed URL found for podcast %s (ID: %s)", item.name, item.item_id
192 )
193 return False
194
195 stored_podcasts = cast("list[str]", self.get_config_value(CONF_STORED_PODCASTS))
196 if feed_url in stored_podcasts:
197 return False
198
199 self.logger.debug("Adding podcast %s to library", item.name)
200 self._update_config_value(CONF_STORED_PODCASTS, [*stored_podcasts, feed_url])
201 return True
202
203 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
204 """
205 Remove podcast from library.
206
207 Removes the podcast's RSS feed URL from the stored podcasts configuration.
208 Always returns True for idempotent operation. If feed URL retrieval fails,
209 logs a warning but still returns True to maintain the idempotent contract
210 as required by MA convention.
211 """
212 # Get the RSS URL for this podcast
213 try:
214 feed_url = await self._get_feed_url_for_podcast(prov_item_id)
215 except Exception as err:
216 self.logger.warning(
217 "Failed to retrieve feed URL for podcast removal %s: %s",
218 prov_item_id,
219 err,
220 exc_info=True,
221 )
222 # Still return True for idempotent operation
223 return True
224
225 if not feed_url:
226 return True
227
228 stored_podcasts = cast("list[str]", self.get_config_value(CONF_STORED_PODCASTS))
229 if feed_url not in stored_podcasts:
230 return True
231
232 self.logger.debug("Removing podcast %s from library", prov_item_id)
233 stored_podcasts = [x for x in stored_podcasts if x != feed_url]
234 self._update_config_value(CONF_STORED_PODCASTS, stored_podcasts)
235 return True
236
237 @use_cache(3600 * 24 * 14) # Cache for 14 days
238 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
239 """Get podcast details."""
240 try:
241 # Try by ID first
242 response = await self._api_request("podcasts/byfeedid", params={"id": prov_podcast_id})
243 if response.get("feed"):
244 podcast = parse_podcast_from_feed(response["feed"], self.instance_id, self.domain)
245 if podcast:
246 return podcast
247 except ProviderUnavailableError, InvalidDataError:
248 # Re-raise these specific errors
249 raise
250 except Exception as err:
251 self.logger.debug("Unexpected error getting podcast %s: %s", prov_podcast_id, err)
252
253 raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found")
254
255 async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
256 """Get episodes for a podcast."""
257 self.logger.debug("Getting episodes for podcast ID: %s", prov_podcast_id)
258
259 # Try to get the podcast name from the current context first
260 podcast_name = None
261 try:
262 podcast = await self.mass.music.podcasts.get_provider_item(
263 prov_podcast_id, self.instance_id
264 )
265 if podcast:
266 podcast_name = podcast.name
267 self.logger.debug("Got podcast name from MA context: %s", podcast_name)
268 except Exception as err:
269 self.logger.debug("Could not get podcast from MA context: %s", err)
270
271 # If we don't have the name, get it from the API
272 if not podcast_name:
273 try:
274 podcast_response = await self._api_request(
275 "podcasts/byfeedid", params={"id": prov_podcast_id}
276 )
277 if podcast_response.get("feed"):
278 podcast_name = podcast_response["feed"].get("title")
279 self.logger.debug("Got podcast name from API fallback: %s", podcast_name)
280 except Exception as err:
281 self.logger.warning("Could not get podcast name from API: %s", err)
282
283 try:
284 response = await self._api_request(
285 "episodes/byfeedid", params={"id": prov_podcast_id, "max": 1000}
286 )
287
288 episodes = response.get("items", [])
289 # rank on the publication date rather than trusting the listing order, so a feed
290 # that numbers only part of its episodes cannot mix two incompatible scales
291 positions = rank_episodes_by_date([ep.get("datePublished") or None for ep in episodes])
292 for position, episode_data in zip(positions, episodes, strict=True):
293 episode = parse_episode_from_data(
294 episode_data,
295 prov_podcast_id,
296 self.instance_id,
297 self.domain,
298 podcast_name,
299 position=position,
300 )
301 if episode:
302 yield episode
303
304 except ProviderUnavailableError, InvalidDataError:
305 # Re-raise these specific errors
306 raise
307 except Exception as err:
308 self.logger.warning(
309 "Unexpected error getting episodes for %s: %s", prov_podcast_id, err
310 )
311
312 @use_cache(43200) # Cache for 12 hours
313 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
314 """
315 Get podcast episode details using direct API lookup.
316
317 Uses the efficient episodes/byid endpoint for direct episode retrieval.
318 """
319 episode_data: dict[str, Any] | None = None
320 episode: PodcastEpisode | None = None
321 try:
322 podcast_id, episode_id = prov_episode_id.split("|", 1)
323 response = await self._api_request("episodes/byid", params={"id": episode_id})
324 episode_data = response.get("episode")
325 if episode_data:
326 episode = parse_episode_from_data(
327 episode_data, podcast_id, self.instance_id, self.domain
328 )
329 except ProviderUnavailableError, InvalidDataError:
330 # Re-raise these specific errors
331 raise
332 except ValueError as err:
333 # Handle malformed episode ID
334 raise InvalidDataError(f"Invalid episode ID format: {prov_episode_id}") from err
335 except Exception as err:
336 self.logger.warning("Unexpected error getting episode %s: %s", prov_episode_id, err)
337
338 if episode is None or episode_data is None:
339 raise MediaNotFoundError(f"Episode {prov_episode_id} not found")
340
341 # single-episode path only: fetch external podcast:chapters JSON (Podcasting 2.0)
342 # when present, to avoid a request per episode during full-podcast listing. Runs
343 # outside the resolution try so a best-effort chapter failure can never surface as
344 # the episode itself being not found.
345 await enrich_episode_chapters(
346 session=self.mass.http_session,
347 chapters_json_url=episode_data.get("chaptersUrl"),
348 mass_episode=episode,
349 )
350 return episode
351
352 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
353 """
354 Get stream details for a podcast episode.
355
356 Uses the Podcast Index episodes/byid endpoint for efficient direct lookup
357 rather than fetching all episodes for a podcast.
358 """
359 if media_type != MediaType.PODCAST_EPISODE:
360 raise MediaNotFoundError("Stream details only available for episodes")
361
362 try:
363 _, episode_id = item_id.split("|", 1)
364
365 # Use direct episode lookup for efficiency
366 response = await self._api_request("episodes/byid", params={"id": episode_id})
367 episode_data = response.get("episode")
368
369 if episode_data:
370 stream_url = episode_data.get("enclosureUrl")
371 if stream_url:
372 return StreamDetails(
373 provider=self.instance_id,
374 item_id=item_id,
375 audio_format=AudioFormat(
376 content_type=ContentType.try_parse(
377 episode_data.get("enclosureType") or "audio/mpeg"
378 ),
379 ),
380 media_type=MediaType.PODCAST_EPISODE,
381 stream_type=StreamType.HTTP,
382 path=stream_url,
383 allow_seek=True,
384 )
385
386 except ProviderUnavailableError, InvalidDataError:
387 # Re-raise these specific errors
388 raise
389 except ValueError as err:
390 # Handle malformed episode ID
391 raise InvalidDataError(f"Invalid episode ID format: {item_id}") from err
392 except Exception as err:
393 self.logger.warning("Unexpected error getting stream for %s: %s", item_id, err)
394
395 raise MediaNotFoundError(f"Stream not found for {item_id}")
396
397 async def _fetch_podcasts(
398 self, endpoint: str, params: dict[str, Any] | None = None
399 ) -> list[Podcast]:
400 """Fetch and parse podcasts from API endpoint."""
401 response = await self._api_request(endpoint, params)
402 podcasts = []
403 for feed_data in response.get("feeds", []):
404 podcast = parse_podcast_from_feed(feed_data, self.instance_id, self.domain)
405 if podcast:
406 podcasts.append(podcast)
407 return podcasts
408
409 async def _api_request(
410 self, endpoint: str, params: dict[str, Any] | None = None
411 ) -> dict[str, Any]:
412 """Make authenticated request to Podcast Index API."""
413 self.logger.log(
414 VERBOSE_LOG_LEVEL, "Making API request to %s with params: %s", endpoint, params
415 )
416 return await make_api_request(self.mass, self.api_key, self.api_secret, endpoint, params)
417
418 async def _get_feed_url_for_podcast(self, podcast_id: str) -> str | None:
419 """Get RSS feed URL for a podcast ID."""
420 try:
421 response = await self._api_request("podcasts/byfeedid", params={"id": podcast_id})
422 feed_data: dict[str, Any] = response.get("feed", {})
423 return feed_data.get("url")
424 except ProviderUnavailableError, InvalidDataError:
425 # Re-raise these specific errors
426 raise
427 except Exception as err:
428 self.logger.warning(
429 "Unexpected error getting feed URL for podcast %s: %s",
430 podcast_id,
431 err,
432 exc_info=True,
433 )
434 return None
435
436 @use_cache(7200) # Cache for 2 hours
437 async def _browse_trending(self) -> list[Podcast]:
438 """Browse trending podcasts."""
439 try:
440 return await self._fetch_podcasts("podcasts/trending", {"max": 50})
441 except ProviderUnavailableError, InvalidDataError:
442 raise
443 except Exception as err:
444 self.logger.warning(
445 "Unexpected error getting trending podcasts: %s", err, exc_info=True
446 )
447 return []
448
449 @use_cache(14400) # Cache for 4 hours
450 async def _browse_recent_episodes(self) -> list[PodcastEpisode]:
451 """Browse recent episodes."""
452 try:
453 response = await self._api_request("recent/episodes", params={"max": 50})
454
455 episodes = []
456 for episode_data in response.get("items", []):
457 # Extract podcast ID from episode data
458 podcast_id = str(episode_data.get("feedId", ""))
459 # Pass feedTitle to avoid unnecessary API calls
460 podcast_name = episode_data.get("feedTitle")
461 episode = parse_episode_from_data(
462 episode_data,
463 podcast_id,
464 self.instance_id,
465 self.domain,
466 podcast_name,
467 )
468 if episode:
469 episodes.append(episode)
470
471 return episodes
472
473 except ProviderUnavailableError, InvalidDataError:
474 # Re-raise these specific errors
475 raise
476 except Exception as err:
477 self.logger.warning("Unexpected error getting recent episodes: %s", err, exc_info=True)
478 return []
479
480 @use_cache(86400) # Cache for 24 hours
481 async def _browse_categories(self) -> list[BrowseFolder]:
482 """Browse podcast categories."""
483 try:
484 response = await self._api_request("categories/list")
485
486 categories = []
487 # Categories API returns feeds array with {id, name} objects
488 categories_data = response.get("feeds", [])
489
490 for category in categories_data:
491 cat_name = category.get("name", "Unknown Category")
492
493 categories.append(
494 BrowseFolder(
495 item_id=cat_name, # Use name as ID
496 provider=self.domain,
497 path=f"{self.instance_id}://{BROWSE_CATEGORIES}/{cat_name}",
498 name=cat_name,
499 )
500 )
501
502 # Sort by name
503 return sorted(categories, key=lambda x: x.name)
504
505 except ProviderUnavailableError, InvalidDataError:
506 # Re-raise these specific errors
507 raise
508 except Exception as err:
509 self.logger.warning("Unexpected error getting categories: %s", err, exc_info=True)
510 return []
511
512 @use_cache(43200) # Cache for 12 hours
513 async def _browse_category_podcasts(self, category_name: str) -> list[Podcast]:
514 """Browse podcasts in a specific category using search."""
515 try:
516 # Search for podcasts using the category name directly
517 search_response = await self._api_request(
518 "search/byterm", params={"q": category_name, "max": 50}
519 )
520
521 podcasts = []
522 for feed_data in search_response.get("feeds", []):
523 podcast = parse_podcast_from_feed(feed_data, self.instance_id, self.domain)
524 if podcast:
525 podcasts.append(podcast)
526
527 return podcasts
528
529 except ProviderUnavailableError, InvalidDataError:
530 raise
531 except Exception as err:
532 self.logger.warning(
533 "Unexpected error getting category podcasts: %s", err, exc_info=True
534 )
535 return []
536