music-assistant-server

1"""iTunes Podcast search support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
10from music_assistant_models.enums import (
11    ConfigEntryType,
12    ContentType,
13    ImageType,
14    MediaType,
15    ProviderFeature,
16    StreamType,
17    TaskScheduleType,
18)
19from music_assistant_models.errors import MediaNotFoundError
20from music_assistant_models.media_items import (
21    AudioFormat,
22    BrowseFolder,
23    ItemMapping,
24    MediaItemImage,
25    MediaItemType,
26    Podcast,
27    PodcastEpisode,
28    ProviderMapping,
29    RecommendationFolder,
30    SearchResults,
31    UniqueList,
32)
33from music_assistant_models.streamdetails import StreamDetails
34
35from music_assistant.constants import CONF_ENTRY_LIBRARY_SYNC_PODCASTS
36from music_assistant.controllers.cache import use_cache
37from music_assistant.helpers.countries import get_country_codes
38from music_assistant.helpers.podcast_parsers import (
39    enrich_episode_chapters,
40    find_episode_stream_url,
41    get_cached_podcast,
42    parse_podcast,
43    parse_podcast_episode,
44    refresh_cached_podcast,
45)
46from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
47from music_assistant.models.music_provider import MusicProvider
48from music_assistant.providers.itunes_podcasts.schema import (
49    ITunesSearchResults,
50    PodcastSearchResult,
51    TopPodcastsHelper,
52    TopPodcastsResponse,
53)
54
55if TYPE_CHECKING:
56    from music_assistant_models.config_entries import ProviderConfig
57    from music_assistant_models.provider import ProviderManifest
58
59    from music_assistant.mass import MusicAssistant
60    from music_assistant.models import ProviderInstanceType
61
62
63CONF_LOCALE = "locale"
64CONF_EXPLICIT = "explicit"
65CONF_NUM_EPISODES = "num_episodes"
66
67# store to search when the server's language has no matching iTunes storefront
68DEFAULT_LOCALE = "us"
69
70# category 0 holds the parsed podcast feeds, see CACHE_CATEGORY_PODCAST_FEED
71CACHE_CATEGORY_RECOMMENDATIONS = 1
72CACHE_KEY_TOP_PODCASTS = "top-podcasts"
73RECOMMENDATION_ROW_TOP_PODCASTS = "itunes-top-podcasts"
74
75SUPPORTED_FEATURES = {
76    ProviderFeature.SEARCH,
77    ProviderFeature.RECOMMENDATIONS,
78    # This provider does not have a "real" library. Refer to method comment
79    # in get_library_podcasts
80    ProviderFeature.LIBRARY_PODCASTS,
81}
82
83CONF_ENTRY_LIBRARY_SYNC_PODCASTS_HIDDEN = ConfigEntry.from_dict(
84    {
85        **CONF_ENTRY_LIBRARY_SYNC_PODCASTS.to_dict(),
86        "hidden": True,
87        "default_value": True,
88    }
89)
90
91
92async def setup(
93    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
94) -> ProviderInstanceType:
95    """Initialize provider(instance) with given configuration."""
96    return ITunesPodcastsProvider(mass, manifest, config, SUPPORTED_FEATURES)
97
98
99class ITunesPodcastsProvider(MusicProvider):
100    """ITunesPodcastsProvider."""
101
102    throttler: ThrottlerManager
103
104    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
105        """Return Config entries to setup this provider."""
106        country_codes = await asyncio.to_thread(get_country_codes)
107
108        language_options = [
109            ConfigValueOption(key.lower(), title=val) for key, val in country_codes.items()
110        ]
111        # the store country decides which catalog is searched; default to the region of the
112        # server's language so the provider can be added without picking one first
113        region = self.mass.metadata.locale.split("_")[-1].upper()
114        return (
115            CONF_ENTRY_LIBRARY_SYNC_PODCASTS_HIDDEN,
116            ConfigEntry(
117                key=CONF_LOCALE,
118                type=ConfigEntryType.STRING,
119                required=True,
120                options=language_options,
121                default_value=region.lower() if region in country_codes else DEFAULT_LOCALE,
122            ),
123            ConfigEntry(
124                key=CONF_NUM_EPISODES,
125                type=ConfigEntryType.INTEGER,
126                required=False,
127                default_value=0,
128            ),
129            ConfigEntry(
130                key=CONF_EXPLICIT,
131                type=ConfigEntryType.BOOLEAN,
132                required=False,
133                default_value=True,
134            ),
135        )
136
137    @property
138    def is_streaming_provider(self) -> bool:
139        """Return True if the provider is a streaming provider."""
140        # For streaming providers return True here but for local file based providers return False.
141        return True
142
143    async def handle_async_init(self) -> None:
144        """Handle async initialization of the provider."""
145        self.max_episodes = int(str(self.config.get_value(CONF_NUM_EPISODES)))
146        # 20 requests per minute, be a bit below
147        self.throttler = ThrottlerManager(rate_limit=18, period=60)
148
149    @use_cache(3600 * 24 * 7)  # Cache for 7 days
150    async def search(
151        self, search_query: str, media_types: list[MediaType], limit: int = 10
152    ) -> SearchResults:
153        """Perform search on musicprovider."""
154        result = SearchResults()
155        if MediaType.PODCAST not in media_types:
156            return result
157
158        if limit < 1:
159            limit = 1
160        elif limit > 200:
161            limit = 200
162        country = str(self.config.get_value(CONF_LOCALE))
163        explicit = "Yes" if bool(self.config.get_value(CONF_EXPLICIT)) else "No"
164        params: dict[str, str | int] = {
165            "media": "podcast",
166            "entity": "podcast",
167            "country": country,
168            "attribute": "titleTerm",
169            "explicit": explicit,
170            "limit": limit,
171            "term": search_query,
172        }
173        url = "https://itunes.apple.com/search?"
174        result.podcasts = await self._perform_search(url, params)
175
176        return result
177
178    async def get_recommendations(self) -> list[RecommendationFolder]:
179        """
180        Get this provider's available recommendation rows, without items.
181
182        A single row with the top podcasts for the configured country.
183        """
184        return [
185            RecommendationFolder(
186                item_id=RECOMMENDATION_ROW_TOP_PODCASTS,
187                name="Trending Podcasts",
188                icon="mdi-trending-up",
189                translation_key="trending_podcasts",
190                provider=self.instance_id,
191            )
192        ]
193
194    async def get_recommendation_items(
195        self, item_id: str
196    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
197        """
198        Get the items for a single recommendation row.
199
200        :param item_id: The item_id of the row, as returned by get_recommendations.
201        """
202        if item_id != RECOMMENDATION_ROW_TOP_PODCASTS:
203            return UniqueList()
204        search_results = await self._cache_get_top_podcasts()
205        return UniqueList(self._get_podcast_list(search_results))
206
207    @throttle_with_retries
208    async def _perform_search(self, url: str, params: dict[str, str | int]) -> list[Podcast]:
209        response = await self.mass.http_session.get(url, params=params)
210        json_response = b""
211        if response.status == 200:
212            json_response = await response.read()
213        if not json_response:
214            return []
215        results = ITunesSearchResults.from_json(json_response).results
216        return self._get_podcast_list(results)
217
218    def _get_podcast_list(self, results: list[PodcastSearchResult]) -> list[Podcast]:
219        podcast_list: list[Podcast] = []
220        for result in results:
221            if result.feed_url is None or result.track_name is None:
222                self.logger.info(
223                    "The podcast '%s' does not have a feed url. Please see the docs for more info.",
224                    result.track_name,
225                )
226                continue
227            podcast = Podcast(
228                name=result.track_name,
229                item_id=result.feed_url,
230                publisher=result.artist_name,
231                provider=self.instance_id,
232                provider_mappings={
233                    ProviderMapping(
234                        item_id=result.feed_url,
235                        provider_domain=self.domain,
236                        provider_instance=self.instance_id,
237                    )
238                },
239            )
240            image_list = []
241            for artwork_url in [
242                result.artwork_url_600,
243                result.artwork_url_100,
244                result.artwork_url_60,
245                result.artwork_url_30,
246            ]:
247                if artwork_url is not None:
248                    image_list.append(
249                        MediaItemImage(
250                            type=ImageType.THUMB, path=artwork_url, provider=self.instance_id
251                        )
252                    )
253            podcast.metadata.images = UniqueList(image_list)
254            podcast_list.append(podcast)
255        return podcast_list
256
257    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
258        """
259        Get library podcasts.
260
261        We use get_library_podcasts to sync all feeds which have been added to the MA library
262        by the user via the search function. The provider itself does not offer a real library.
263
264        The item_id corresponds to the feed_url.
265        """
266        podcasts = await self.mass.music.podcasts.get_library_items_by_prov_id(
267            provider_instance=self.instance_id
268        )
269        for podcast in podcasts:
270            our_provider_mapping: ProviderMapping | None = None
271            for provider_mapping in podcast.provider_mappings:
272                if provider_mapping.provider_instance == self.instance_id:
273                    our_provider_mapping = provider_mapping
274                    break
275            if our_provider_mapping is None:
276                # We should never end up here.
277                self.logger.error("Podcast %s lacks a provider mapping.", podcast.name)
278                continue
279            feed_url = our_provider_mapping.item_id
280            parsed_podcast: dict[str, Any] | None = None
281            try:
282                parsed_podcast = await refresh_cached_podcast(
283                    mass=self.mass,
284                    provider_instance_id=self.instance_id,
285                    feed_url=feed_url,
286                    max_episodes=self.max_episodes,
287                    cache_expiration=self._get_cache_expiration(),
288                )
289                self.logger.debug("Synced podcast %s.", podcast.name)
290            except MediaNotFoundError:
291                # If we are not able to refresh the podcast, we must prevent the sync
292                # from deleting the podcast from the library - that is both a breaking change
293                # (pre March 2026) and certainly not desired just because of some downtime.
294                self.logger.warning("Was unable to sync podcast %s (%s).", podcast.name, feed_url)
295                podcast.item_id = feed_url
296                podcast.provider_mappings = {our_provider_mapping}
297                yield podcast
298                continue
299
300            yield parse_podcast(
301                feed_url=feed_url,
302                parsed_feed=parsed_podcast,
303                instance_id=self.instance_id,
304                domain=self.domain,
305            )
306
307    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
308        """Get podcast."""
309        parsed = await self._cache_get_podcast(prov_podcast_id)
310
311        return parse_podcast(
312            feed_url=prov_podcast_id,
313            parsed_feed=parsed,
314            instance_id=self.instance_id,
315            domain=self.domain,
316        )
317
318    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
319        """Get podcast episodes."""
320        podcast = await self._cache_get_podcast(prov_podcast_id)
321        podcast_cover = podcast.get("cover_url")
322        episodes = podcast.get("episodes", [])
323        for cnt, episode in enumerate(episodes):
324            if mass_episode := parse_podcast_episode(
325                episode=episode,
326                prov_podcast_id=prov_podcast_id,
327                episode_cnt=cnt,
328                podcast_cover=podcast_cover,
329                podcast_name=podcast.get("title"),
330                domain=self.domain,
331                instance_id=self.instance_id,
332            ):
333                yield mass_episode
334
335    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
336        """Get single podcast episode."""
337        podcast_id, guid_or_stream_url = prov_episode_id.split(" ")
338        podcast = await self._cache_get_podcast(podcast_id)
339        podcast_cover = podcast.get("cover_url")
340        for cnt, episode in enumerate(podcast.get("episodes", [])):
341            mass_episode = parse_podcast_episode(
342                episode=episode,
343                prov_podcast_id=podcast_id,
344                episode_cnt=cnt,
345                podcast_cover=podcast_cover,
346                podcast_name=podcast.get("title"),
347                domain=self.domain,
348                instance_id=self.instance_id,
349            )
350            if mass_episode is None:
351                continue
352            _, _guid_or_stream_url = mass_episode.item_id.split(" ")
353            # this is enough, as internal
354            if guid_or_stream_url == _guid_or_stream_url:
355                await enrich_episode_chapters(
356                    session=self.mass.http_session,
357                    chapters_json_url=episode.get("chapters_json_url"),
358                    mass_episode=mass_episode,
359                )
360                return mass_episode
361        raise MediaNotFoundError("Episode not found")
362
363    async def _get_episode_stream_url(self, podcast_id: str, guid_or_stream_url: str) -> str | None:
364        parsed_podcast = await self._cache_get_podcast(podcast_id)
365        return find_episode_stream_url(
366            parsed_feed=parsed_podcast, guid_or_stream_url=guid_or_stream_url
367        )
368
369    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
370        """Get streamdetails for item."""
371        podcast_id, guid_or_stream_url = item_id.split(" ")
372        stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
373        if stream_url is None:
374            raise MediaNotFoundError
375        return StreamDetails(
376            provider=self.instance_id,
377            item_id=item_id,
378            audio_format=AudioFormat(
379                content_type=ContentType.try_parse(stream_url),
380            ),
381            media_type=MediaType.PODCAST_EPISODE,
382            stream_type=StreamType.HTTP,
383            path=stream_url,
384            can_seek=True,
385            allow_seek=True,
386        )
387
388    @throttle_with_retries
389    async def _get_podcast_search_result_from_itunes_id(
390        self, itunes_id: int
391    ) -> PodcastSearchResult:
392        params = {"id": itunes_id}
393        url = "https://itunes.apple.com/lookup?"
394        response = await self.mass.http_session.get(url, params=params)
395        json_response = b""
396        if response.status == 200:
397            json_response = await response.read()
398        if not json_response:
399            raise MediaNotFoundError
400        search_results = ITunesSearchResults.from_json(json_response)
401        if search_results.result_count == 0:
402            raise MediaNotFoundError
403        if search_results.result_count > 1:
404            self.logger.warning("More than a single result for podcast.")
405        return search_results.results[0]
406
407    async def _cache_get_podcast(self, prov_podcast_id: str) -> dict[str, Any]:
408        # raises MediaNotFoundError if the feed is gone or invalid
409        return await get_cached_podcast(
410            mass=self.mass,
411            provider_instance_id=self.instance_id,
412            feed_url=prov_podcast_id,
413            max_episodes=self.max_episodes,
414            cache_expiration=self._get_cache_expiration(),
415        )
416
417    def _get_cache_expiration(self) -> int:
418        # Cache slightly longer than the effective sync interval to avoid fetching
419        # the same podcast feed repeatedly during recurring library sync.
420        schedule = self.mass.music.get_provider_sync_schedule(self.instance_id, MediaType.PODCAST)
421        library_sync_enabled = bool(self.config.get_value("library_sync_podcasts"))
422        if not library_sync_enabled or schedule is None or not schedule.enabled:
423            return 60 * 60 * 12  # 12h
424        if schedule.type == TaskScheduleType.HOURLY and schedule.every is not None:
425            return schedule.every * 60 * 60 + 600  # 10 minutes extra cache
426        if schedule.type == TaskScheduleType.DAILY and schedule.every is not None:
427            return schedule.every * 24 * 60 * 60 + 600
428        return 60 * 60 * 12  # 12h
429
430    async def _cache_set_top_podcasts(self, top_podcast_helper: TopPodcastsHelper) -> None:
431        await self.mass.cache.set(
432            key=CACHE_KEY_TOP_PODCASTS,
433            provider=self.instance_id,
434            category=CACHE_CATEGORY_RECOMMENDATIONS,
435            data=top_podcast_helper.to_dict(),
436            expiration=60 * 60 * 6,  # 6 hours
437        )
438
439    async def _cache_get_top_podcasts(self) -> list[PodcastSearchResult]:
440        parsed_top_podcasts = await self.mass.cache.get(
441            key=CACHE_KEY_TOP_PODCASTS,
442            provider=self.instance_id,
443            category=CACHE_CATEGORY_RECOMMENDATIONS,
444        )
445        if parsed_top_podcasts is not None:
446            helper = TopPodcastsHelper.from_dict(parsed_top_podcasts)
447            return helper.top_podcasts
448
449        # 15 results
450        # keep 20 requests max per minute in mind
451        # https://rss.marketingtools.apple.com/
452        country = str(self.config.get_value(CONF_LOCALE))
453        url = f"https://rss.marketingtools.apple.com/api/v2/{country}/podcasts/top/15/podcasts.json"
454        response = await self.mass.http_session.get(url)
455        json_response = b""
456        if response.status == 200:
457            json_response = await response.read()
458        if not json_response:
459            return []
460
461        top_podcasts_response = TopPodcastsResponse.from_json(json_response)
462
463        if top_podcasts_response.feed is None:
464            return []
465
466        include_explicit = bool(self.config.get_value(CONF_EXPLICIT))
467
468        helper = TopPodcastsHelper()
469        for top_podcast in top_podcasts_response.feed.results:
470            if not include_explicit and top_podcast.content_advisory_rating is not None:
471                # the spelling within the API is wrong.
472                if top_podcast.content_advisory_rating in [
473                    "explicit",
474                    "Explicit",
475                    "Explict",
476                    "explict",
477                ]:
478                    continue
479            try:
480                podcast_search_result = await self._get_podcast_search_result_from_itunes_id(
481                    int(top_podcast.id_)
482                )
483            except MediaNotFoundError:
484                continue
485            helper.top_podcasts.append(podcast_search_result)
486
487        await self._cache_set_top_podcasts(top_podcast_helper=helper)
488        return helper.top_podcasts
489