/
/
/
1"""
2Overcast provider for Music Assistant.
3
4Imports podcast subscriptions and playback progress from an Overcast account
5using the account's extended OPML export. Synchronization is strictly one-way
6(Overcast -> Music Assistant): Overcast has no ingest API, so ``on_played`` is
7deliberately not implemented and no state is ever written back.
8
9The OPML export endpoint is rate limited by Overcast, therefore the export is
10cached aggressively and refreshed in the background, while the RSS feeds it
11refers to are fetched directly from the podcasts' own servers.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import math
18import time
19from typing import TYPE_CHECKING, Any, cast
20
21import aiohttp
22from music_assistant_models.config_entries import ConfigEntry
23from music_assistant_models.enums import (
24 ConfigEntryType,
25 ContentType,
26 MediaType,
27 StreamType,
28)
29from music_assistant_models.errors import (
30 LoginFailed,
31 MediaNotFoundError,
32 ResourceTemporarilyUnavailable,
33)
34from music_assistant_models.media_items import AudioFormat, Podcast, PodcastEpisode
35from music_assistant_models.streamdetails import StreamDetails
36from yarl import URL
37
38from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER, CONF_PASSWORD, CONF_USERNAME
39from music_assistant.controllers.cache import use_cache
40from music_assistant.helpers.aiohttp_client import create_clientsession
41from music_assistant.helpers.datetime import from_iso_string
42from music_assistant.helpers.podcast_parsers import (
43 enrich_episode_chapters,
44 find_episode_stream_url,
45 get_cached_podcast,
46 get_episode_positions,
47 get_stream_url_and_guid_from_episode,
48 parse_podcast,
49 parse_podcast_episode,
50 refresh_cached_podcast,
51)
52from music_assistant.helpers.throttle_retry import parse_retry_after
53from music_assistant.models.music_provider import MusicProvider
54
55from .constants import (
56 AUTH_REJECT_STATUSES,
57 BASE_URL,
58 CACHE_CATEGORY_OPML,
59 CACHE_KEY_LAST_APPLIED,
60 CONF_MAX_NUM_EPISODES,
61 CONF_SESSION_COOKIE,
62 LOGIN_URL,
63 OPML_CACHE_EXPIRATION,
64 OPML_EXPORT_URL,
65 PODCASTS_URL,
66 RATE_LIMIT_FALLBACK_BACKOFF,
67 SESSION_COOKIE_NAME,
68)
69from .helpers import OvercastSubscription, match_episode_state, parse_extended_opml
70
71if TYPE_CHECKING:
72 from collections.abc import AsyncGenerator
73 from datetime import datetime
74
75
76class OvercastProvider(MusicProvider):
77 """Provider that imports podcast subscriptions from an Overcast account."""
78
79 http_session: aiohttp.ClientSession
80 max_episodes: int
81 # newest playback state applied per feed url: a feed that could not be retrieved
82 # keeps its own watermark, so its states are still applied once it recovers
83 _feed_watermarks: dict[str, datetime]
84 # the parsed export, kept alongside the raw text it was parsed from so a
85 # refreshed export is re-parsed while repeated lookups are not
86 _opml_cache: tuple[str, dict[str, OvercastSubscription]] | None = None
87 # monotonic deadline set from a 429's Retry-After, see _rate_limit_remaining
88 _rate_limited_until: float | None = None
89
90 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
91 """
92 Return the (options) config entries for the Overcast provider.
93
94 The account credentials are collected by the interactive setup flow
95 (see ``setup_flow.py``); only the max-episodes limit is configured here.
96 """
97 return (
98 CONF_ENTRY_UNOFFICIAL_PROVIDER,
99 ConfigEntry(
100 key=CONF_MAX_NUM_EPISODES,
101 type=ConfigEntryType.INTEGER,
102 required=False,
103 default_value=0,
104 ),
105 )
106
107 async def handle_async_init(self) -> None:
108 """Handle async initialization of the provider."""
109 self.max_episodes = cast("int", self.config.get_value(CONF_MAX_NUM_EPISODES, 0))
110 # Dedicated session with its own cookie jar to support multi-instance
111 # (each instance has its own Overcast login cookie)
112 self.http_session = create_clientsession(self.mass, cookie_jar=aiohttp.CookieJar())
113 stored_cookie = self.get_setup_value(CONF_SESSION_COOKIE)
114 cookie_valid = False
115 if isinstance(stored_cookie, str) and stored_cookie:
116 self.http_session.cookie_jar.update_cookies(
117 {SESSION_COOKIE_NAME: stored_cookie}, response_url=URL(BASE_URL)
118 )
119 cookie_valid = await self._session_cookie_valid()
120 if not cookie_valid:
121 await self._login()
122
123 raw_watermarks = await self.mass.cache.get(
124 key=CACHE_KEY_LAST_APPLIED,
125 provider=self.instance_id,
126 category=CACHE_CATEGORY_OPML,
127 default={},
128 )
129 self._feed_watermarks = {
130 feed_url: from_iso_string(raw) for feed_url, raw in raw_watermarks.items()
131 }
132
133 async def unload(self, is_removed: bool = False) -> None:
134 """Handle unload/close of the provider."""
135 if not self.http_session.closed:
136 await self.http_session.close()
137
138 @property
139 def is_streaming_provider(self) -> bool:
140 """Return False: the library mirrors the user's own Overcast subscriptions."""
141 return False
142
143 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
144 """Retrieve the subscribed podcasts from the Overcast account."""
145 subscriptions = await self._get_opml_subscriptions()
146 for feed_url, subscription in subscriptions.items():
147 self.logger.debug("Adding podcast with feed %s to library", feed_url)
148 try:
149 parsed_podcast = await refresh_cached_podcast(
150 mass=self.mass,
151 provider_instance_id=self.instance_id,
152 feed_url=feed_url,
153 max_episodes=self.max_episodes,
154 )
155 except MediaNotFoundError as err:
156 self.report_skipped_sync_item(MediaType.PODCAST, feed_url, err)
157 continue
158 applied = await self._apply_playback_states(feed_url, subscription, parsed_podcast)
159 if applied is not None:
160 # stored right away: the caller may stop consuming this generator at
161 # any point, which would otherwise re-apply these states on the next sync
162 self._feed_watermarks[feed_url] = applied
163 await self._store_watermarks()
164 yield parse_podcast(
165 feed_url=feed_url,
166 parsed_feed=parsed_podcast,
167 instance_id=self.instance_id,
168 domain=self.domain,
169 )
170
171 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
172 """Get the podcast for the given feed url."""
173 parsed_podcast = await self._cache_get_podcast(prov_podcast_id)
174 return parse_podcast(
175 feed_url=prov_podcast_id,
176 parsed_feed=parsed_podcast,
177 instance_id=self.instance_id,
178 domain=self.domain,
179 )
180
181 async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
182 """Get all episodes of a podcast, including their Overcast playback state."""
183 podcast = await self._cache_get_podcast(prov_podcast_id)
184 subscription = await self._get_subscription(prov_podcast_id)
185 podcast_cover = podcast.get("cover_url")
186 podcast_name = podcast.get("title")
187 episodes = podcast.get("episodes", [])
188 positions = get_episode_positions(episodes)
189 for position, parsed_episode in zip(positions, episodes, strict=True):
190 mass_episode = parse_podcast_episode(
191 episode=parsed_episode,
192 prov_podcast_id=prov_podcast_id,
193 position=position,
194 podcast_cover=podcast_cover,
195 podcast_name=podcast_name,
196 instance_id=self.instance_id,
197 domain=self.domain,
198 )
199 if mass_episode is None:
200 # faulty episode
201 continue
202 try:
203 stream_url, _ = get_stream_url_and_guid_from_episode(episode=parsed_episode)
204 except ValueError:
205 # episode enclosure or stream url missing
206 continue
207 if subscription is not None:
208 state = match_episode_state(subscription, stream_url)
209 if state is not None and (state.played or state.progress_s):
210 mass_episode.resume_position_ms = (state.progress_s or 0) * 1000
211 mass_episode.fully_played = state.played
212 yield mass_episode
213
214 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
215 """Get a single podcast episode."""
216 podcast_id, guid_or_stream_url = prov_episode_id.split(" ", 1)
217 async for mass_episode in self.get_podcast_episodes(podcast_id):
218 _, episode_key = mass_episode.item_id.split(" ", 1)
219 if episode_key == guid_or_stream_url:
220 await self._enrich_episode_chapters(podcast_id, guid_or_stream_url, mass_episode)
221 return mass_episode
222 raise MediaNotFoundError("Did not find episode.")
223
224 async def get_resume_position(
225 self, item_id: str, media_type: MediaType
226 ) -> tuple[bool, int, datetime | None]:
227 """Return fully_played, resume position (ms) and its timestamp from Overcast."""
228 if media_type != MediaType.PODCAST_EPISODE:
229 raise NotImplementedError
230 podcast_id, guid_or_stream_url = item_id.split(" ", 1)
231 stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
232 if stream_url is None:
233 raise NotImplementedError
234 subscription = await self._get_subscription(podcast_id)
235 state = match_episode_state(subscription, stream_url) if subscription else None
236 if state is None or (not state.played and state.progress_s is None):
237 # No known Overcast progress; raise NotImplementedError such that MA
238 # falls back to the resume position stored in its own playlog.
239 raise NotImplementedError
240 return state.played, max((state.progress_s or 0) * 1000, 0), state.user_updated_at
241
242 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
243 """Get streamdetails for item."""
244 podcast_id, guid_or_stream_url = item_id.split(" ", 1)
245 stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
246 if stream_url is None:
247 raise MediaNotFoundError
248 return StreamDetails(
249 provider=self.instance_id,
250 item_id=item_id,
251 audio_format=AudioFormat(
252 content_type=ContentType.try_parse(stream_url),
253 ),
254 media_type=MediaType.PODCAST_EPISODE,
255 stream_type=StreamType.HTTP,
256 path=stream_url,
257 can_seek=True,
258 allow_seek=True,
259 )
260
261 async def _login(self) -> None:
262 """Authenticate with Overcast and persist the session cookie."""
263 email = str(self.get_setup_value(CONF_USERNAME))
264 password = str(self.get_setup_value(CONF_PASSWORD))
265 try:
266 async with self.http_session.post(
267 LOGIN_URL,
268 data={"email": email, "password": password},
269 allow_redirects=False,
270 ) as response:
271 status = response.status
272 location = response.headers.get("Location", "")
273 morsel = response.cookies.get(SESSION_COOKIE_NAME)
274 except (TimeoutError, aiohttp.ClientError) as err:
275 raise ResourceTemporarilyUnavailable("Overcast is unreachable") from err
276 if status != 302 or "/podcasts" not in location or morsel is None:
277 raise LoginFailed("Overcast login failed, check your email and password")
278 self._update_setup_data(CONF_SESSION_COOKIE, morsel.value)
279
280 async def _session_cookie_valid(self) -> bool:
281 """Check whether the restored session cookie is still accepted by Overcast."""
282 try:
283 async with self.http_session.get(PODCASTS_URL, allow_redirects=False) as response:
284 return response.status == 200
285 except (TimeoutError, aiohttp.ClientError) as err:
286 raise ResourceTemporarilyUnavailable("Overcast is unreachable") from err
287
288 @use_cache(
289 expiration=OPML_CACHE_EXPIRATION,
290 category=CACHE_CATEGORY_OPML,
291 allow_expired_cache=True,
292 cache_none=False,
293 )
294 async def _fetch_opml_text(self) -> str:
295 """Fetch the account's extended OPML export."""
296 opml_text = await self._request_opml()
297 if opml_text is None:
298 # the session expired, log in again and retry once
299 await self._login()
300 opml_text = await self._request_opml()
301 if opml_text is None:
302 raise LoginFailed("Overcast rejected the session right after a fresh login")
303 return opml_text
304
305 async def _request_opml(self) -> str | None:
306 """Return the raw OPML document, or None if the session cookie was rejected."""
307 if remaining := self._rate_limit_remaining():
308 # spend no request while Overcast is still refusing them: the export allows
309 # only ~10 per day and every episode listing would otherwise cost one
310 raise ResourceTemporarilyUnavailable(
311 "Overcast OPML export is rate limited", backoff_time=remaining
312 )
313 try:
314 async with self.http_session.get(OPML_EXPORT_URL, allow_redirects=False) as response:
315 if response.status == 200:
316 return await response.text()
317 if response.status == 429:
318 backoff = (
319 parse_retry_after(response.headers.get("Retry-After"))
320 or RATE_LIMIT_FALLBACK_BACKOFF
321 )
322 self._rate_limited_until = time.monotonic() + backoff
323 raise ResourceTemporarilyUnavailable(
324 "Overcast OPML export is rate limited", backoff_time=backoff
325 )
326 if response.status in AUTH_REJECT_STATUSES:
327 return None
328 raise ResourceTemporarilyUnavailable(
329 f"Overcast OPML export failed with HTTP {response.status}"
330 )
331 except (TimeoutError, aiohttp.ClientError) as err:
332 raise ResourceTemporarilyUnavailable("Overcast is unreachable") from err
333
334 def _rate_limit_remaining(self) -> int:
335 """Return the seconds left of a known Overcast rate limit window, 0 if none."""
336 if self._rate_limited_until is None:
337 return 0
338 remaining = self._rate_limited_until - time.monotonic()
339 if remaining <= 0:
340 self._rate_limited_until = None
341 return 0
342 return math.ceil(remaining)
343
344 async def _get_opml_subscriptions(self) -> dict[str, OvercastSubscription]:
345 opml_text = await self._fetch_opml_text()
346 if self._opml_cache is None or self._opml_cache[0] != opml_text:
347 parsed = await asyncio.to_thread(parse_extended_opml, opml_text)
348 self._opml_cache = (opml_text, parsed)
349 return self._opml_cache[1]
350
351 async def _get_subscription(self, feed_url: str) -> OvercastSubscription | None:
352 """Return the Overcast subscription for a feed, or None if unavailable."""
353 try:
354 subscriptions = await self._get_opml_subscriptions()
355 except (ResourceTemporarilyUnavailable, LoginFailed) as err:
356 # episodes can still be listed without playback state
357 self.logger.debug("Could not obtain Overcast playback states: %s", err)
358 return None
359 return subscriptions.get(feed_url)
360
361 async def _apply_playback_states(
362 self,
363 feed_url: str,
364 subscription: OvercastSubscription,
365 parsed_podcast: dict[str, Any],
366 ) -> datetime | None:
367 """
368 Push a feed's new Overcast playback states to the playlog.
369
370 :param feed_url: The podcast's feed url (also the provider item id).
371 :param subscription: The Overcast subscription holding the episode states.
372 :param parsed_podcast: The podcastparser dict of the feed.
373 :return: The newest state timestamp that was applied, or None if none were.
374 """
375 watermark = self._feed_watermarks.get(feed_url)
376 newest_applied: datetime | None = None
377 podcast_cover = parsed_podcast.get("cover_url")
378 podcast_name = parsed_podcast.get("title")
379 all_episodes = parsed_podcast.get("episodes", [])
380 positions = get_episode_positions(all_episodes)
381 for position, parsed_episode in zip(positions, all_episodes, strict=True):
382 try:
383 stream_url, _ = get_stream_url_and_guid_from_episode(episode=parsed_episode)
384 except ValueError:
385 continue
386 state = match_episode_state(subscription, stream_url)
387 if state is None or state.user_updated_at is None:
388 continue
389 if not state.played and not state.progress_s:
390 # never mark items unplayed: an absent state cannot be told apart
391 # from an episode that simply was never touched in Overcast
392 continue
393 if watermark is not None and state.user_updated_at <= watermark:
394 # already applied in a previous sync; skipping it also makes sure
395 # local progress made since then is not overwritten
396 continue
397 mass_episode = parse_podcast_episode(
398 episode=parsed_episode,
399 prov_podcast_id=feed_url,
400 position=position,
401 podcast_cover=podcast_cover,
402 podcast_name=podcast_name,
403 instance_id=self.instance_id,
404 domain=self.domain,
405 )
406 if mass_episode is None:
407 continue
408 if not state.played:
409 # never move the user backwards: the playlog write replaces whatever MA
410 # recorded itself, which may be a position further into the episode
411 _, local_position_ms = await self.mass.music.get_resume_position(mass_episode)
412 if local_position_ms > (state.progress_s or 0) * 1000:
413 continue
414 await self.mass.music.mark_item_played(
415 mass_episode,
416 fully_played=state.played,
417 seconds_played=state.progress_s or 0,
418 user_initiated=False,
419 )
420 if newest_applied is None or state.user_updated_at > newest_applied:
421 newest_applied = state.user_updated_at
422 return newest_applied
423
424 async def _store_watermarks(self) -> None:
425 """Persist the per-feed watermarks of the applied playback states."""
426 # watermarks of feeds that are no longer subscribed are kept on purpose,
427 # so re-subscribing does not re-apply the feed's entire playback history
428 await self.mass.cache.set(
429 key=CACHE_KEY_LAST_APPLIED,
430 provider=self.instance_id,
431 category=CACHE_CATEGORY_OPML,
432 data={feed_url: ts.isoformat() for feed_url, ts in self._feed_watermarks.items()},
433 )
434
435 async def _enrich_episode_chapters(
436 self, prov_podcast_id: str, guid_or_stream_url: str, mass_episode: PodcastEpisode
437 ) -> None:
438 """
439 Attach external ``podcast:chapters`` JSON to a resolved single episode, if any.
440
441 :param prov_podcast_id: Provider podcast id the episode belongs to.
442 :param guid_or_stream_url: Episode identifier used to locate the raw parsed episode.
443 :param mass_episode: The episode to enrich in place; left untouched on any failure.
444 """
445 if mass_episode.metadata.chapters:
446 return
447 podcast = await self._cache_get_podcast(prov_podcast_id)
448 for episode in podcast.get("episodes", []):
449 try:
450 stream_url, guid = get_stream_url_and_guid_from_episode(episode=episode)
451 except ValueError:
452 continue
453 if guid_or_stream_url in (guid, stream_url):
454 await enrich_episode_chapters(
455 session=self.mass.http_session,
456 chapters_json_url=episode.get("chapters_json_url"),
457 mass_episode=mass_episode,
458 )
459 return
460
461 async def _get_episode_stream_url(self, podcast_id: str, guid_or_stream_url: str) -> str | None:
462 parsed_podcast = await self._cache_get_podcast(podcast_id)
463 return find_episode_stream_url(
464 parsed_feed=parsed_podcast, guid_or_stream_url=guid_or_stream_url
465 )
466
467 async def _cache_get_podcast(self, prov_podcast_id: str) -> dict[str, Any]:
468 # raises MediaNotFoundError when the feed is gone
469 return await get_cached_podcast(
470 mass=self.mass,
471 provider_instance_id=self.instance_id,
472 feed_url=prov_podcast_id,
473 max_episodes=self.max_episodes,
474 )
475