/
/
/
1"""
2Podcast RSS Feed Music Provider for Music Assistant.
3
4A URL to a podcast feed can be configured. The contents of that specific podcast
5feed will be forwarded to music assistant. In order to have multiple podcast feeds,
6multiple instances with each one feed must exist.
7
8"""
9
10from __future__ import annotations
11
12from collections.abc import AsyncGenerator
13from typing import TYPE_CHECKING, Any
14
15import podcastparser
16from aiohttp.client_exceptions import ClientError
17from music_assistant_models.enums import (
18 ContentType,
19 MediaType,
20 ProviderFeature,
21 StreamType,
22)
23from music_assistant_models.errors import InvalidProviderURI, MediaNotFoundError
24from music_assistant_models.helpers import create_safe_string
25from music_assistant_models.media_items import (
26 AudioFormat,
27 MediaItemImage,
28 Podcast,
29 PodcastEpisode,
30 UniqueList,
31)
32from music_assistant_models.streamdetails import StreamDetails
33
34from music_assistant.controllers.cache import use_cache
35from music_assistant.helpers.podcast_parsers import (
36 enrich_episode_chapters,
37 get_cached_podcast,
38 get_episode_positions,
39 get_stream_url_from_episode,
40 parse_podcast,
41 parse_podcast_episode,
42 refresh_cached_podcast,
43)
44from music_assistant.models.music_provider import MusicProvider
45
46if TYPE_CHECKING:
47 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
48 from music_assistant_models.provider import ProviderManifest
49
50 from music_assistant.mass import MusicAssistant
51 from music_assistant.models import ProviderInstanceType
52
53CONF_FEED_URL = "feed_url"
54
55SUPPORTED_FEATURES = {
56 ProviderFeature.BROWSE,
57 ProviderFeature.LIBRARY_PODCASTS,
58}
59
60
61async def setup(
62 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
63) -> ProviderInstanceType:
64 """Initialize provider(instance) with given configuration."""
65 return PodcastMusicprovider(mass, manifest, config, SUPPORTED_FEATURES)
66
67
68class PodcastMusicprovider(MusicProvider):
69 """Podcast RSS Feed Music Provider."""
70
71 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
72 """Return Config entries to configure this provider."""
73 return ()
74
75 async def handle_async_init(self) -> None:
76 """Handle async initialization of the provider."""
77 feed_url = self.get_setup_value(CONF_FEED_URL)
78 if not feed_url:
79 msg = "No podcast feed set"
80 raise InvalidProviderURI(msg)
81 self.feed_url = podcastparser.normalize_feed_url(str(feed_url))
82 if self.feed_url is None:
83 raise MediaNotFoundError("The specified feed url cannot be used.")
84
85 self.podcast_id = create_safe_string(self.feed_url.replace("http", ""))
86
87 try:
88 self.parsed_podcast: dict[str, Any] = await self._cache_get_podcast()
89 except ClientError as exc:
90 raise MediaNotFoundError("Invalid URL") from exc
91
92 @property
93 def is_streaming_provider(self) -> bool:
94 """
95 Return True if the provider is a streaming provider.
96
97 This literally means that the catalog is not the same as the library contents.
98 For local based providers (files, plex), the catalog is the same as the library content.
99 It also means that data is if this provider is NOT a streaming provider,
100 data cross instances is unique, the catalog and library differs per instance.
101
102 Setting this to True will only query one instance of the provider for search and lookups.
103 Setting this to False will query all instances of this provider for search and lookups.
104 """
105 return False
106
107 @property
108 def instance_name_postfix(self) -> str | None:
109 """Return a (default) instance name postfix for this provider instance."""
110 return self.parsed_podcast.get("title")
111
112 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
113 """Retrieve library/subscribed podcasts from the provider."""
114 """
115 Only one podcast per rss feed is supported. The data format of the rss feed supports
116 only one podcast.
117 """
118 # on sync we renew
119 assert self.feed_url is not None
120 self.parsed_podcast = await refresh_cached_podcast(
121 mass=self.mass,
122 provider_instance_id=self.instance_id,
123 feed_url=self.feed_url,
124 )
125 yield await self._parse_podcast()
126
127 @use_cache(3600 * 24 * 7) # Cache for 7 days
128 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
129 """Get full artist details by id."""
130 if prov_podcast_id != self.podcast_id:
131 raise MediaNotFoundError(f"Podcast id not in provider: {prov_podcast_id}")
132 return await self._parse_podcast()
133
134 @use_cache(3600) # Cache for 1 hour
135 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
136 """Get (full) podcast episode details by id."""
137 episodes = self.parsed_podcast["episodes"]
138 positions = get_episode_positions(episodes)
139 for position, episode in zip(positions, episodes, strict=True):
140 if prov_episode_id == episode["guid"]:
141 if mass_episode := self._parse_episode(episode, position):
142 await enrich_episode_chapters(
143 session=self.mass.http_session,
144 chapters_json_url=episode.get("chapters_json_url"),
145 mass_episode=mass_episode,
146 )
147 return mass_episode
148 raise MediaNotFoundError("Episode not found")
149
150 async def get_podcast_episodes(
151 self,
152 prov_podcast_id: str,
153 ) -> AsyncGenerator[PodcastEpisode]:
154 """List all episodes for the podcast."""
155 if prov_podcast_id != self.podcast_id:
156 raise MediaNotFoundError(f"Podcast id not in provider: {prov_podcast_id}")
157 # yield newest-first like the other providers, so callers after the latest episode
158 # only have to take the first one
159 episodes: list[dict[str, Any]] = self.parsed_podcast["episodes"]
160 if episodes and episodes[0].get("published", 0) != 0:
161 episodes.sort(key=lambda x: x.get("published", 0), reverse=True)
162 positions = get_episode_positions(episodes)
163 for position, episode in zip(positions, episodes, strict=True):
164 if mass_episode := self._parse_episode(episode, position):
165 yield mass_episode
166
167 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
168 """Get streamdetails for a track/radio."""
169 for episode in self.parsed_podcast["episodes"]:
170 if item_id == episode["guid"]:
171 stream_url = get_stream_url_from_episode(episode=episode)
172 if stream_url is None:
173 raise MediaNotFoundError(f"Episode {item_id} has no playable stream")
174 return StreamDetails(
175 provider=self.instance_id,
176 item_id=item_id,
177 audio_format=AudioFormat(
178 content_type=ContentType.try_parse(stream_url),
179 ),
180 media_type=MediaType.PODCAST_EPISODE,
181 stream_type=StreamType.HTTP,
182 path=stream_url,
183 can_seek=True,
184 allow_seek=True,
185 extra_input_args=[
186 "-user_agent",
187 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
188 ],
189 )
190 raise MediaNotFoundError("Stream not found")
191
192 async def resolve_image(self, path: str) -> str | bytes:
193 """Resolve image for RSS provider with fallback to podcast cover."""
194 if not path.startswith("http"):
195 return path
196
197 try:
198 async with self.mass.http_session.get(path, raise_for_status=True) as response:
199 # Check if we got actual image content
200 content_type = response.headers.get("content-type", "").lower()
201 if not content_type.startswith(("image/", "application/octet-stream")):
202 # Not an image - likely redirected to error page
203 raise ClientError(f"Invalid content type: {content_type}")
204
205 return await response.read()
206
207 except ClientError, Exception:
208 # Try podcast cover fallback
209 podcast_cover = self.parsed_podcast.get("cover_url")
210 if podcast_cover and isinstance(podcast_cover, str) and podcast_cover != path:
211 async with self.mass.http_session.get(
212 podcast_cover, raise_for_status=True
213 ) as response:
214 return await response.read()
215
216 raise MediaNotFoundError(f"Episode image not found: {path}")
217
218 async def _parse_podcast(self) -> Podcast:
219 """Parse podcast information from podcast feed."""
220 assert self.feed_url is not None
221 return parse_podcast(
222 feed_url=self.feed_url,
223 parsed_feed=self.parsed_podcast,
224 instance_id=self.instance_id,
225 domain=self.domain,
226 mass_item_id=self.podcast_id,
227 )
228
229 def _parse_episode(self, episode_obj: dict[str, Any], position: int) -> PodcastEpisode | None:
230 episode_result = parse_podcast_episode(
231 episode=episode_obj,
232 prov_podcast_id=self.podcast_id,
233 position=position,
234 podcast_cover=self.parsed_podcast.get("cover_url"),
235 podcast_name=self.parsed_podcast.get("title"),
236 instance_id=self.instance_id,
237 domain=self.domain,
238 mass_item_id=episode_obj["guid"],
239 )
240 # Override remotely_accessible as these providers can have unreliable image URLs
241 if episode_result and episode_result.metadata.images:
242 new_images = []
243 for img in episode_result.metadata.images:
244 new_images.append(
245 MediaItemImage(
246 type=img.type,
247 path=img.path,
248 provider=img.provider,
249 remotely_accessible=False, # Force through imageproxy
250 )
251 )
252 episode_result.metadata.images = UniqueList(new_images)
253
254 return episode_result
255
256 async def _cache_get_podcast(self) -> dict[str, Any]:
257 assert self.feed_url is not None
258 return await get_cached_podcast(
259 mass=self.mass,
260 provider_instance_id=self.instance_id,
261 feed_url=self.feed_url,
262 )
263