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