music-assistant-server

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