/
/
/
1"""
2gPodder provider for Music Assistant.
3
4Tested against opodsync, https://github.com/kd2org/opodsync
5and nextcloud-gpodder, https://github.com/thrillfall/nextcloud-gpodder
6gpodder.net is not supported due to responsiveness/ frequent downtimes of domain.
7
8Note:
9 - it can happen, that we have the guid and use that for identification, but the sync state
10 provider, eg. opodsync might use only the stream url. So always make sure, to compare both
11 when relying on an external service
12 - The service calls have a timestamp (int, unix epoch s), which give the changes since then.
13"""
14
15from __future__ import annotations
16
17import time
18from collections.abc import AsyncGenerator
19from datetime import datetime
20from typing import TYPE_CHECKING, Any, cast
21
22from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
23from music_assistant_models.enums import (
24 ConfigEntryType,
25 ContentType,
26 MediaType,
27 ProviderFeature,
28 StreamType,
29)
30from music_assistant_models.errors import (
31 LoginFailed,
32 MediaNotFoundError,
33 ResourceTemporarilyUnavailable,
34)
35from music_assistant_models.media_items import AudioFormat, MediaItemType, Podcast, PodcastEpisode
36from music_assistant_models.streamdetails import StreamDetails
37
38from music_assistant.helpers.datetime import from_utc_timestamp
39from music_assistant.helpers.podcast_parsers import (
40 enrich_episode_chapters,
41 find_episode_stream_url,
42 get_cached_podcast,
43 get_episode_positions,
44 get_stream_url_and_guid_from_episode,
45 parse_podcast,
46 parse_podcast_episode,
47 refresh_cached_podcast,
48)
49from music_assistant.models.music_provider import MusicProvider
50
51from .client import EpisodeActionDelete, EpisodeActionNew, EpisodeActionPlay, GPodderClient
52
53if TYPE_CHECKING:
54 from music_assistant_models.provider import ProviderManifest
55
56 from music_assistant.mass import MusicAssistant
57 from music_assistant.models import ProviderInstanceType
58
59# Config for "classic" gpodder api
60CONF_URL = "url"
61CONF_USERNAME = "username"
62CONF_PASSWORD = "password"
63CONF_DEVICE_ID = "device_id"
64
65# Config for nextcloud
66CONF_TOKEN_NC = "token"
67CONF_URL_NC = "url_nc"
68
69# General config
70CONF_VERIFY_SSL = "verify_ssl"
71CONF_MAX_NUM_EPISODES = "max_num_episodes"
72
73
74# category 0 holds the individual parsed podcasts, see CACHE_CATEGORY_PODCAST_FEED
75CACHE_CATEGORY_OTHER = 1
76CACHE_KEY_TIMESTAMP = (
77 "timestamp" # tuple of two ints, timestamp_subscriptions and timestamp_actions
78)
79CACHE_KEY_FEEDS = "feeds" # list[str] : all available rss feed urls
80
81SUPPORTED_FEATURES = {
82 ProviderFeature.LIBRARY_PODCASTS,
83 ProviderFeature.BROWSE,
84}
85
86
87async def setup(
88 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
89) -> ProviderInstanceType:
90 """Initialize provider(instance) with given configuration."""
91 return GPodder(mass, manifest, config, SUPPORTED_FEATURES)
92
93
94class GPodder(MusicProvider):
95 """gPodder MusicProvider."""
96
97 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
98 """
99 Return the (options) config entries for the gPodder provider.
100
101 The server/account connection (gpodder API or Nextcloud) is set up by the interactive
102 setup flow (see ``setup_flow.py``); only the max-episodes limit is configured here.
103 """
104 return (
105 ConfigEntry(
106 key=CONF_MAX_NUM_EPISODES,
107 type=ConfigEntryType.INTEGER,
108 required=False,
109 default_value=0,
110 ),
111 )
112
113 async def handle_async_init(self) -> None:
114 """Pass config values to client and initialize."""
115 base_url = str(self.get_setup_value(CONF_URL))
116 _username = self.get_setup_value(CONF_USERNAME)
117 _password = self.get_setup_value(CONF_PASSWORD)
118 _device_id = self.get_setup_value(CONF_DEVICE_ID)
119 nc_url = str(self.get_setup_value(CONF_URL_NC))
120 nc_token = self.get_setup_value(CONF_TOKEN_NC)
121 verify_ssl = bool(self.get_setup_value(CONF_VERIFY_SSL, True))
122
123 self.max_episodes = cast("int", self.config.get_value(CONF_MAX_NUM_EPISODES, 0))
124
125 self._client = GPodderClient(
126 session=self.mass.http_session, logger=self.logger, verify_ssl=verify_ssl
127 )
128
129 if nc_token is not None:
130 assert nc_url is not None
131 self._client.init_nc(base_url=nc_url, nc_token=str(nc_token))
132 else:
133 if _username is None or _password is None or _device_id is None:
134 raise LoginFailed("Must provide username, password and device_id.")
135 username = str(_username)
136 password = str(_password)
137 device_id = str(_device_id)
138
139 if base_url.rstrip("/") == "https://gpodder.net":
140 raise LoginFailed("Do not use gpodder.net. See docs for explanation.")
141 try:
142 await self._client.init_gpodder(
143 username=username, password=password, base_url=base_url, device=device_id
144 )
145 except RuntimeError as exc:
146 raise LoginFailed("Login failed.") from exc
147
148 timestamps = await self.mass.cache.get(
149 key=CACHE_KEY_TIMESTAMP,
150 provider=self.instance_id,
151 category=CACHE_CATEGORY_OTHER,
152 default=None,
153 )
154 if timestamps is None:
155 self.timestamp_subscriptions: int = 0
156 self.timestamp_actions: int = 0
157 else:
158 self.timestamp_subscriptions, self.timestamp_actions = timestamps
159
160 self.logger.debug(
161 "Our timestamps are (subscriptions, actions) (%s, %s)",
162 self.timestamp_subscriptions,
163 self.timestamp_actions,
164 )
165
166 feeds = await self.mass.cache.get(
167 key=CACHE_KEY_FEEDS,
168 provider=self.instance_id,
169 category=CACHE_CATEGORY_OTHER,
170 default=None,
171 )
172 if feeds is None:
173 self.feeds: set[str] = set()
174 else:
175 self.feeds = set(feeds) # feeds is a list here
176
177 # we are syncing the playlog, but not event based. A simple check in on_played,
178 # should be sufficient
179 self.progress_guard_timestamp = 0.0
180
181 @property
182 def is_streaming_provider(self) -> bool:
183 """Return True if the provider is a streaming provider."""
184 # For streaming providers return True here but for local file based providers return False.
185 # While the streams are remote, the user controls what is added.
186 return False
187
188 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
189 """Retrieve library/subscribed podcasts from the provider."""
190 try:
191 subscriptions = await self._client.get_subscriptions()
192 except RuntimeError:
193 raise ResourceTemporarilyUnavailable(backoff_time=30)
194 if subscriptions is None:
195 return
196
197 for feed_url in subscriptions.add:
198 self.feeds.add(feed_url)
199 for feed_url in subscriptions.remove:
200 try:
201 self.feeds.remove(feed_url)
202 except KeyError:
203 # a podcast might have been added and removed in our absence...
204 continue
205
206 episode_actions, timestamp_action = await self._client.get_episode_actions()
207 for feed_url in self.feeds:
208 self.logger.debug("Adding podcast with feed %s to library", feed_url)
209 # parse podcast
210 try:
211 parsed_podcast = await refresh_cached_podcast(
212 mass=self.mass,
213 provider_instance_id=self.instance_id,
214 feed_url=feed_url,
215 max_episodes=self.max_episodes,
216 )
217 except MediaNotFoundError as err:
218 self.report_skipped_sync_item(MediaType.PODCAST, feed_url, err)
219 continue
220
221 # playlog
222 # be safe, if there should be multiple episodeactions. client already sorts
223 # progresses in descending order.
224 _already_processed = set()
225 _episode_actions = [x for x in episode_actions if x.podcast == feed_url]
226 for _action in _episode_actions:
227 if _action.episode not in _already_processed:
228 _already_processed.add(_action.episode)
229 # we do not have to add the progress, these would make calls twice,
230 # and we only use the object to propagate to playlog
231 self.progress_guard_timestamp = time.time()
232 _episode_ids: list[str] = []
233 if _action.guid is not None:
234 _episode_ids.append(f"{feed_url} {_action.guid}")
235 _episode_ids.append(f"{feed_url} {_action.episode}")
236 mass_episode: PodcastEpisode | None = None
237 for _episode_id in _episode_ids:
238 try:
239 mass_episode = await self.get_podcast_episode(
240 _episode_id, add_progress=False
241 )
242 break
243 except MediaNotFoundError:
244 continue
245 if mass_episode is None:
246 self.logger.debug(
247 f"Was unable to use progress for episode {_action.episode}."
248 )
249 continue
250 match _action:
251 case EpisodeActionNew():
252 await self.mass.music.mark_item_unplayed(mass_episode)
253 case EpisodeActionPlay():
254 await self.mass.music.mark_item_played(
255 mass_episode,
256 fully_played=_action.position >= _action.total,
257 seconds_played=_action.position,
258 user_initiated=False,
259 )
260
261 # cache
262 yield parse_podcast(
263 feed_url=feed_url,
264 parsed_feed=parsed_podcast,
265 instance_id=self.instance_id,
266 domain=self.domain,
267 )
268
269 self.timestamp_subscriptions = subscriptions.timestamp
270 if timestamp_action is not None:
271 self.timestamp_actions = timestamp_action
272 await self._cache_set_timestamps()
273 await self._cache_set_feeds()
274
275 async def get_podcast(self, prov_podcast_id: str) -> Podcast:
276 """Get Podcast."""
277 parsed_podcast = await self._cache_get_podcast(prov_podcast_id)
278
279 return parse_podcast(
280 feed_url=prov_podcast_id,
281 parsed_feed=parsed_podcast,
282 instance_id=self.instance_id,
283 domain=self.domain,
284 )
285
286 async def get_podcast_episodes(
287 self, prov_podcast_id: str, add_progress: bool = True
288 ) -> AsyncGenerator[PodcastEpisode]:
289 """Get Podcast episodes. Add progress information."""
290 if add_progress:
291 episode_actions, timestamp = await self._client.get_episode_actions()
292 else:
293 episode_actions, timestamp = [], None
294
295 podcast = await self._cache_get_podcast(prov_podcast_id)
296 podcast_cover = podcast.get("cover_url")
297 parsed_episodes = podcast.get("episodes", [])
298
299 if timestamp is not None:
300 self.timestamp_actions = timestamp
301 await self._cache_set_timestamps()
302
303 positions = get_episode_positions(parsed_episodes)
304 for position, parsed_episode in zip(positions, parsed_episodes, strict=True):
305 mass_episode = parse_podcast_episode(
306 episode=parsed_episode,
307 prov_podcast_id=prov_podcast_id,
308 position=position,
309 podcast_cover=podcast_cover,
310 podcast_name=podcast.get("title"),
311 domain=self.domain,
312 instance_id=self.instance_id,
313 )
314 if mass_episode is None:
315 # faulty episode
316 continue
317 try:
318 stream_url, guid = get_stream_url_and_guid_from_episode(episode=parsed_episode)
319 except ValueError:
320 # episode enclosure or stream url missing
321 continue
322
323 for action in episode_actions:
324 # we have to test both, as we are comparing to external input.
325 _test = [action.guid, action.episode]
326 if prov_podcast_id == action.podcast and (guid in _test or stream_url in _test):
327 self.progress_guard_timestamp = time.time()
328 if isinstance(action, EpisodeActionNew):
329 mass_episode.resume_position_ms = 0
330 mass_episode.fully_played = False
331
332 # propagate to playlog
333 await self.mass.music.mark_item_unplayed(
334 mass_episode,
335 )
336 elif isinstance(action, EpisodeActionPlay):
337 fully_played = action.position >= action.total
338 resume_position_s = action.position
339 mass_episode.resume_position_ms = resume_position_s * 1000
340 mass_episode.fully_played = fully_played
341
342 # propagate progress to playlog
343 await self.mass.music.mark_item_played(
344 mass_episode,
345 fully_played=fully_played,
346 seconds_played=resume_position_s,
347 user_initiated=False,
348 )
349 elif isinstance(action, EpisodeActionDelete):
350 for mapping in mass_episode.provider_mappings:
351 mapping.available = False
352 break
353 yield mass_episode
354
355 async def get_podcast_episode(
356 self, prov_episode_id: str, add_progress: bool = True
357 ) -> PodcastEpisode:
358 """Get Podcast Episode. Add progress information."""
359 podcast_id, guid_or_stream_url = prov_episode_id.split(" ")
360 async for mass_episode in self.get_podcast_episodes(podcast_id, add_progress=add_progress):
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 self._enrich_episode_chapters(podcast_id, guid_or_stream_url, mass_episode)
365 return mass_episode
366 raise MediaNotFoundError("Did not find episode.")
367
368 async def get_resume_position(
369 self, item_id: str, media_type: MediaType
370 ) -> tuple[bool, int, datetime | None]:
371 """Return: finished, position_ms."""
372 assert media_type == MediaType.PODCAST_EPISODE
373 podcast_id, guid_or_stream_url = item_id.split(" ")
374 stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
375 try:
376 progresses, timestamp = await self._client.get_episode_actions(
377 since=self.timestamp_actions
378 )
379 except RuntimeError:
380 self.logger.warning("Was unable to obtain progresses.")
381 raise NotImplementedError # fallback to internal position.
382 for action in progresses:
383 _test = [action.guid, action.episode]
384 # progress is external, compare guid and stream_url
385 if action.podcast == podcast_id and (
386 guid_or_stream_url in _test or stream_url in _test
387 ):
388 dt_timestamp: datetime | None = None
389 if timestamp is not None:
390 self.timestamp_actions = timestamp
391 await self._cache_set_timestamps()
392 dt_timestamp = from_utc_timestamp(timestamp)
393 if isinstance(action, EpisodeActionNew | EpisodeActionDelete):
394 # no progress, it might have been actively reset
395 # in case of delete, we start from start.
396 return False, 0, None
397 _progress = (action.position >= action.total, max(action.position * 1000, 0))
398 self.logger.debug("Found an updated external resume position.")
399 return action.position >= action.total, max(action.position * 1000, 0), dt_timestamp
400 self.logger.debug("Did not find an updated resume position, falling back to stored.")
401 # If we did not find a resume position, nothing changed since our last timestamp
402 # we raise NotImplementedError, such that MA falls back to the already stored
403 # resume_position in its playlog.
404 raise NotImplementedError
405
406 async def on_played(
407 self,
408 media_type: MediaType,
409 prov_item_id: str,
410 fully_played: bool,
411 position: int,
412 media_item: MediaItemType,
413 is_playing: bool = False,
414 ) -> None:
415 """Update progress."""
416 if media_item is None or not isinstance(media_item, PodcastEpisode):
417 return
418 if media_type != MediaType.PODCAST_EPISODE:
419 return
420 if time.time() - self.progress_guard_timestamp <= 5:
421 return
422 podcast_id, guid_or_stream_url = prov_item_id.split(" ")
423 stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
424 assert stream_url is not None
425 duration = media_item.duration
426 try:
427 await self._client.update_progress(
428 podcast_id=podcast_id,
429 episode_id=stream_url,
430 guid=guid_or_stream_url,
431 position_s=position,
432 duration_s=duration,
433 )
434 self.logger.debug(f"Updated progress to {position / duration * 100:.2f}%")
435 except RuntimeError as exc:
436 self.logger.debug(exc)
437 self.logger.debug("Failed to update progress.")
438
439 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
440 """Get streamdetails for item."""
441 podcast_id, guid_or_stream_url = item_id.split(" ")
442 stream_url = await self._get_episode_stream_url(podcast_id, guid_or_stream_url)
443 if stream_url is None:
444 raise MediaNotFoundError
445 return StreamDetails(
446 provider=self.instance_id,
447 item_id=item_id,
448 audio_format=AudioFormat(
449 content_type=ContentType.try_parse(stream_url),
450 ),
451 media_type=MediaType.PODCAST_EPISODE,
452 stream_type=StreamType.HTTP,
453 path=stream_url,
454 can_seek=True,
455 allow_seek=True,
456 )
457
458 async def _enrich_episode_chapters(
459 self, prov_podcast_id: str, guid_or_stream_url: str, mass_episode: PodcastEpisode
460 ) -> None:
461 """
462 Attach external ``podcast:chapters`` JSON to a resolved single episode, if any.
463
464 :param prov_podcast_id: Provider podcast id the episode belongs to.
465 :param guid_or_stream_url: Episode identifier used to locate the raw parsed episode.
466 :param mass_episode: The episode to enrich in place; left untouched on any failure.
467 """
468 if mass_episode.metadata.chapters:
469 return
470 podcast = await self._cache_get_podcast(prov_podcast_id)
471 for episode in podcast.get("episodes", []):
472 try:
473 stream_url, guid = get_stream_url_and_guid_from_episode(episode=episode)
474 except ValueError:
475 continue
476 if guid_or_stream_url in (guid, stream_url):
477 await enrich_episode_chapters(
478 session=self.mass.http_session,
479 chapters_json_url=episode.get("chapters_json_url"),
480 mass_episode=mass_episode,
481 )
482 return
483
484 async def _get_episode_stream_url(self, podcast_id: str, guid_or_stream_url: str) -> str | None:
485 parsed_podcast = await self._cache_get_podcast(podcast_id)
486 return find_episode_stream_url(
487 parsed_feed=parsed_podcast, guid_or_stream_url=guid_or_stream_url
488 )
489
490 async def _cache_get_podcast(self, prov_podcast_id: str) -> dict[str, Any]:
491 # raises MediaNotFoundError when the feed is gone
492 return await get_cached_podcast(
493 mass=self.mass,
494 provider_instance_id=self.instance_id,
495 feed_url=prov_podcast_id,
496 max_episodes=self.max_episodes,
497 )
498
499 async def _cache_set_timestamps(self) -> None:
500 # seven days default
501 await self.mass.cache.set(
502 key=CACHE_KEY_TIMESTAMP,
503 provider=self.instance_id,
504 category=CACHE_CATEGORY_OTHER,
505 data=[self.timestamp_subscriptions, self.timestamp_actions],
506 )
507
508 async def _cache_set_feeds(self) -> None:
509 # seven days default
510 await self.mass.cache.set(
511 key=CACHE_KEY_FEEDS,
512 provider=self.instance_id,
513 category=CACHE_CATEGORY_OTHER,
514 data=list(self.feeds),
515 )
516