/
/
/
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, LoginFailed:
248 raise
249 except Exception as err:
250 self.logger.debug("Unexpected error getting podcast %s: %s", prov_podcast_id, err)
251
252 raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found")
253
254 async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
255 """Get episodes for a podcast."""
256 self.logger.debug("Getting episodes for podcast ID: %s", prov_podcast_id)
257
258 # Try to get the podcast name from the current context first
259 podcast_name = None
260 try:
261 podcast = await self.mass.music.podcasts.get_provider_item(
262 prov_podcast_id, self.instance_id
263 )
264 if podcast:
265 podcast_name = podcast.name
266 self.logger.debug("Got podcast name from MA context: %s", podcast_name)
267 except Exception as err:
268 self.logger.debug("Could not get podcast from MA context: %s", err)
269
270 # If we don't have the name, get it from the API
271 if not podcast_name:
272 try:
273 podcast_response = await self._api_request(
274 "podcasts/byfeedid", params={"id": prov_podcast_id}
275 )
276 if podcast_response.get("feed"):
277 podcast_name = podcast_response["feed"].get("title")
278 self.logger.debug("Got podcast name from API fallback: %s", podcast_name)
279 except Exception as err:
280 self.logger.warning("Could not get podcast name from API: %s", err)
281
282 try:
283 response = await self._api_request(
284 "episodes/byfeedid", params={"id": prov_podcast_id, "max": 1000}
285 )
286
287 episodes = response.get("items", [])
288 # rank on the publication date rather than trusting the listing order, so a feed
289 # that numbers only part of its episodes cannot mix two incompatible scales
290 positions = rank_episodes_by_date([ep.get("datePublished") or None for ep in episodes])
291 for position, episode_data in zip(positions, episodes, strict=True):
292 episode = parse_episode_from_data(
293 episode_data,
294 prov_podcast_id,
295 self.instance_id,
296 self.domain,
297 podcast_name,
298 position=position,
299 )
300 if episode:
301 yield episode
302
303 except ProviderUnavailableError, InvalidDataError, LoginFailed:
304 raise
305 except Exception as err:
306 self.logger.warning(
307 "Unexpected error getting episodes for %s: %s", prov_podcast_id, err
308 )
309
310 @use_cache(43200) # Cache for 12 hours
311 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
312 """
313 Get podcast episode details using direct API lookup.
314
315 Uses the efficient episodes/byid endpoint for direct episode retrieval.
316 """
317 episode_data: dict[str, Any] | None = None
318 episode: PodcastEpisode | None = None
319 try:
320 podcast_id, episode_id = prov_episode_id.split("|", 1)
321 response = await self._api_request("episodes/byid", params={"id": episode_id})
322 episode_data = response.get("episode")
323 if episode_data:
324 episode = parse_episode_from_data(
325 episode_data, podcast_id, self.instance_id, self.domain
326 )
327 except ProviderUnavailableError, InvalidDataError, LoginFailed:
328 raise
329 except ValueError as err:
330 # Handle malformed episode ID
331 raise InvalidDataError(f"Invalid episode ID format: {prov_episode_id}") from err
332 except Exception as err:
333 self.logger.warning("Unexpected error getting episode %s: %s", prov_episode_id, err)
334
335 if episode is None or episode_data is None:
336 raise MediaNotFoundError(f"Episode {prov_episode_id} not found")
337
338 # single-episode path only: fetch external podcast:chapters JSON (Podcasting 2.0)
339 # when present, to avoid a request per episode during full-podcast listing. Runs
340 # outside the resolution try so a best-effort chapter failure can never surface as
341 # the episode itself being not found.
342 await enrich_episode_chapters(
343 session=self.mass.http_session,
344 chapters_json_url=episode_data.get("chaptersUrl"),
345 mass_episode=episode,
346 )
347 return episode
348
349 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
350 """
351 Get stream details for a podcast episode.
352
353 Uses the Podcast Index episodes/byid endpoint for efficient direct lookup
354 rather than fetching all episodes for a podcast.
355 """
356 if media_type != MediaType.PODCAST_EPISODE:
357 raise MediaNotFoundError("Stream details only available for episodes")
358
359 try:
360 _, episode_id = item_id.split("|", 1)
361
362 # Use direct episode lookup for efficiency
363 response = await self._api_request("episodes/byid", params={"id": episode_id})
364 if not (episode_data := response.get("episode")):
365 self.logger.debug(
366 "Podcast Index has no episode %s, it may have left the index", episode_id
367 )
368 elif not (stream_url := episode_data.get("enclosureUrl")):
369 self.logger.debug(
370 "Episode %s carries no audio url, so there is nothing to play", episode_id
371 )
372 else:
373 content_type = episode_data.get("enclosureType") or "audio/mpeg"
374 self.logger.debug("Streaming episode %s as %s", episode_id, content_type)
375 return StreamDetails(
376 provider=self.instance_id,
377 item_id=item_id,
378 audio_format=AudioFormat(content_type=ContentType.try_parse(content_type)),
379 media_type=MediaType.PODCAST_EPISODE,
380 stream_type=StreamType.HTTP,
381 path=stream_url,
382 allow_seek=True,
383 )
384
385 except ProviderUnavailableError, InvalidDataError, LoginFailed:
386 raise
387 except ValueError as err:
388 # Handle malformed episode ID
389 raise InvalidDataError(f"Invalid episode ID format: {item_id}") from err
390 except Exception as err:
391 self.logger.warning("Unexpected error getting stream for %s: %s", item_id, err)
392
393 raise MediaNotFoundError(f"Stream not found for {item_id}")
394
395 async def _fetch_podcasts(
396 self, endpoint: str, params: dict[str, Any] | None = None
397 ) -> list[Podcast]:
398 """Fetch and parse podcasts from API endpoint."""
399 response = await self._api_request(endpoint, params)
400 podcasts = []
401 for feed_data in response.get("feeds", []):
402 podcast = parse_podcast_from_feed(feed_data, self.instance_id, self.domain)
403 if podcast:
404 podcasts.append(podcast)
405 return podcasts
406
407 async def _api_request(
408 self, endpoint: str, params: dict[str, Any] | None = None
409 ) -> dict[str, Any]:
410 """Make authenticated request to Podcast Index API."""
411 self.logger.log(
412 VERBOSE_LOG_LEVEL, "Making API request to %s with params: %s", endpoint, params
413 )
414 return await make_api_request(
415 self.mass, self.api_key, self.api_secret, endpoint, params, logger=self.logger
416 )
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, LoginFailed:
425 raise
426 except Exception as err:
427 self.logger.warning(
428 "Unexpected error getting feed URL for podcast %s: %s",
429 podcast_id,
430 err,
431 exc_info=True,
432 )
433 return None
434
435 @use_cache(7200) # Cache for 2 hours
436 async def _browse_trending(self) -> list[Podcast]:
437 """Browse trending podcasts."""
438 try:
439 return await self._fetch_podcasts("podcasts/trending", {"max": 50})
440 except ProviderUnavailableError, InvalidDataError, LoginFailed:
441 raise
442 except Exception as err:
443 self.logger.warning(
444 "Unexpected error getting trending podcasts: %s", err, exc_info=True
445 )
446 return []
447
448 @use_cache(14400) # Cache for 4 hours
449 async def _browse_recent_episodes(self) -> list[PodcastEpisode]:
450 """Browse recent episodes."""
451 try:
452 response = await self._api_request("recent/episodes", params={"max": 50})
453
454 episodes = []
455 for episode_data in response.get("items", []):
456 # Extract podcast ID from episode data
457 podcast_id = str(episode_data.get("feedId", ""))
458 # Pass feedTitle to avoid unnecessary API calls
459 podcast_name = episode_data.get("feedTitle")
460 episode = parse_episode_from_data(
461 episode_data,
462 podcast_id,
463 self.instance_id,
464 self.domain,
465 podcast_name,
466 )
467 if episode:
468 episodes.append(episode)
469
470 return episodes
471
472 except ProviderUnavailableError, InvalidDataError, LoginFailed:
473 raise
474 except Exception as err:
475 self.logger.warning("Unexpected error getting recent episodes: %s", err, exc_info=True)
476 return []
477
478 @use_cache(86400) # Cache for 24 hours
479 async def _browse_categories(self) -> list[BrowseFolder]:
480 """Browse podcast categories."""
481 try:
482 response = await self._api_request("categories/list")
483
484 categories = []
485 # Categories API returns feeds array with {id, name} objects
486 categories_data = response.get("feeds", [])
487
488 for category in categories_data:
489 cat_name = category.get("name", "Unknown Category")
490
491 categories.append(
492 BrowseFolder(
493 item_id=cat_name, # Use name as ID
494 provider=self.domain,
495 path=f"{self.instance_id}://{BROWSE_CATEGORIES}/{cat_name}",
496 name=cat_name,
497 )
498 )
499
500 # Sort by name
501 return sorted(categories, key=lambda x: x.name)
502
503 except ProviderUnavailableError, InvalidDataError, LoginFailed:
504 raise
505 except Exception as err:
506 self.logger.warning("Unexpected error getting categories: %s", err, exc_info=True)
507 return []
508
509 @use_cache(43200) # Cache for 12 hours
510 async def _browse_category_podcasts(self, category_name: str) -> list[Podcast]:
511 """Browse podcasts in a specific category using search."""
512 try:
513 # Search for podcasts using the category name directly
514 search_response = await self._api_request(
515 "search/byterm", params={"q": category_name, "max": 50}
516 )
517
518 podcasts = []
519 for feed_data in search_response.get("feeds", []):
520 podcast = parse_podcast_from_feed(feed_data, self.instance_id, self.domain)
521 if podcast:
522 podcasts.append(podcast)
523
524 return podcasts
525
526 except ProviderUnavailableError, InvalidDataError, LoginFailed:
527 raise
528 except Exception as err:
529 self.logger.warning(
530 "Unexpected error getting category podcasts: %s", err, exc_info=True
531 )
532 return []
533