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