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