/
/
/
1"""API client for the Pocket Casts service."""
2
3from __future__ import annotations
4
5import logging
6from typing import Any, cast
7
8import aiohttp
9from music_assistant_models.errors import (
10 LoginFailed,
11 ProviderUnavailableError,
12 ResourceTemporarilyUnavailable,
13)
14
15from music_assistant.helpers.json import json_loads
16from music_assistant.helpers.throttle_retry import (
17 ThrottlerManager,
18 parse_retry_after,
19 throttle_with_retries,
20)
21
22API_BASE_URL = "https://api.pocketcasts.com"
23PODCAST_API_URL = "https://podcast-api.pocketcasts.com"
24
25
26class PocketCastsClient:
27 """Client for the Pocket Casts API."""
28
29 throttler = ThrottlerManager(rate_limit=5, period=1)
30
31 def __init__(self, session: aiohttp.ClientSession, logger: logging.Logger) -> None:
32 """
33 Initialize the client.
34
35 :param session: The aiohttp session to use for requests (typically mass.http_session).
36 :param logger: The provider logger, used for throttle/retry messages.
37 """
38 self.token: str | None = None
39 self.user_uuid: str | None = None
40 self.session = session
41 self.logger = logger
42
43 async def login(self, email: str, password: str) -> None:
44 """
45 Authenticate with Pocket Casts and store the session token.
46
47 :param email: The account email address.
48 :param password: The account password.
49 """
50 data = await self._request(
51 "POST",
52 f"{API_BASE_URL}/user/login",
53 auth=False,
54 data={"email": email, "password": password},
55 )
56 self.token = data.get("token")
57 self.user_uuid = data.get("uuid")
58 if not self.token:
59 raise LoginFailed("No token in Pocket Casts login response")
60 self.logger.info("Successfully logged in to Pocket Casts")
61
62 async def get_subscribed_podcasts(self) -> list[dict[str, Any]]:
63 """Return the user's subscribed podcasts."""
64 data = await self._request("POST", f"{API_BASE_URL}/user/podcast/list")
65 podcasts: list[dict[str, Any]] = data.get("podcasts", [])
66 self.logger.debug("Retrieved %d subscribed podcasts", len(podcasts))
67 return podcasts
68
69 async def get_podcast(self, podcast_uuid: str) -> dict[str, Any]:
70 """
71 Return full details (including episodes) for a podcast by UUID.
72
73 :param podcast_uuid: The podcast UUID.
74 """
75 data = await self._request(
76 "GET",
77 f"{PODCAST_API_URL}/podcast/full/{podcast_uuid}",
78 auth=False,
79 allow_redirects=True,
80 )
81 podcast: dict[str, Any] = data.get("podcast", {})
82 return podcast
83
84 async def get_podcast_episodes(self, podcast_uuid: str) -> tuple[str, list[dict[str, Any]]]:
85 """
86 Return a podcast's title and all of its episodes.
87
88 :param podcast_uuid: The podcast UUID.
89 """
90 podcast = await self.get_podcast(podcast_uuid)
91 # full-podcast episodes use snake_case keys: uuid, title, url, file_type, file_size,
92 # duration (seconds), published, type, slug, has_generated_transcript. Note this is a
93 # different (leaner) schema than the /user/episode endpoint - no playback status,
94 # episode number, show notes or artwork.
95 episodes: list[dict[str, Any]] = podcast.get("episodes", [])
96 self.logger.debug("Retrieved %d episodes for podcast %s", len(episodes), podcast_uuid)
97 return str(podcast.get("title", "")), episodes
98
99 async def get_show_notes(self, podcast_uuid: str) -> dict[str, dict[str, Any]]:
100 """
101 Return the show notes and artwork, keyed by episode UUID, for a podcast.
102
103 Episodes carrying neither are left out.
104
105 :param podcast_uuid: The podcast UUID.
106 """
107 data = await self._request(
108 "GET",
109 f"{PODCAST_API_URL}/mobile/show_notes/full/{podcast_uuid}",
110 auth=False,
111 allow_redirects=True,
112 )
113 # one call covers every episode, and no other endpoint carries these two fields. The
114 # rest of the response is dropped here to keep the cached entry small.
115 show_notes: dict[str, dict[str, Any]] = {}
116 for episode in data.get("podcast", {}).get("episodes", []):
117 if not (uuid := episode.get("uuid")):
118 continue
119 details: dict[str, Any] = {}
120 if description := episode.get("show_notes"):
121 details["description"] = description
122 if image := episode.get("image"):
123 details["image"] = image
124 if details:
125 show_notes[uuid] = details
126 self.logger.debug(
127 "Retrieved show notes for %d episodes of podcast %s", len(show_notes), podcast_uuid
128 )
129 return show_notes
130
131 async def get_in_progress_episodes(self) -> list[dict[str, Any]]:
132 """Return episodes currently in progress."""
133 data = await self._request("POST", f"{API_BASE_URL}/user/in_progress")
134 episodes: list[dict[str, Any]] = data.get("episodes", [])
135 self.logger.debug("Retrieved %d in-progress episodes", len(episodes))
136 return episodes
137
138 async def get_up_next_episodes(self) -> list[dict[str, Any]]:
139 """Return the Up Next queue episodes."""
140 data = await self._request("POST", f"{API_BASE_URL}/up_next/list")
141 episodes = data.get("episodes", [])
142 # the up_next endpoint returns a uuid-keyed map; normalise to a list carrying the uuid
143 if isinstance(episodes, dict):
144 return [{"uuid": uuid, **episode} for uuid, episode in episodes.items()]
145 return cast("list[dict[str, Any]]", episodes)
146
147 async def get_new_releases(self) -> list[dict[str, Any]]:
148 """Return new release episodes from subscriptions."""
149 data = await self._request("POST", f"{API_BASE_URL}/user/new_releases")
150 episodes: list[dict[str, Any]] = data.get("episodes", [])
151 self.logger.debug("Retrieved %d new release episodes", len(episodes))
152 return episodes
153
154 async def get_starred_episodes(self) -> list[dict[str, Any]]:
155 """Return starred episodes."""
156 data = await self._request("POST", f"{API_BASE_URL}/user/starred")
157 episodes: list[dict[str, Any]] = data.get("episodes", [])
158 self.logger.debug("Retrieved %d starred episodes", len(episodes))
159 return episodes
160
161 async def get_history(self) -> list[dict[str, Any]]:
162 """Return listening history episodes."""
163 data = await self._request("POST", f"{API_BASE_URL}/user/history")
164 episodes: list[dict[str, Any]] = data.get("episodes", [])
165 self.logger.debug("Retrieved %d history episodes", len(episodes))
166 return episodes
167
168 async def get_episode_details(self, episode_uuid: str) -> dict[str, Any]:
169 """
170 Return detailed episode info including correct duration and playback status.
171
172 :param episode_uuid: The episode UUID.
173 """
174 # /user/episode returns camelCase keys: uuid, title, url, fileType, duration (seconds),
175 # published, episodeNumber, playedUpTo (resume seconds), playingStatus (1=unplayed,
176 # 2=in progress, 3=played), starred, podcastUuid. No show notes or episode artwork.
177 data = await self._request(
178 "POST", f"{API_BASE_URL}/user/episode", json={"uuid": episode_uuid}
179 )
180 self.logger.debug(
181 "Episode %s: duration=%s, status=%s, playedUpTo=%s",
182 episode_uuid,
183 data.get("duration"),
184 data.get("playingStatus"),
185 data.get("playedUpTo"),
186 )
187 return data
188
189 async def search_podcasts(self, query: str) -> list[dict[str, Any]]:
190 """
191 Search for podcasts.
192
193 :param query: The search term.
194 """
195 data = await self._request("POST", f"{API_BASE_URL}/discover/search", json={"term": query})
196 podcasts: list[dict[str, Any]] = data.get("podcasts", [])
197 self.logger.debug("Found %d podcasts for query '%s'", len(podcasts), query)
198 return podcasts
199
200 async def update_episode_progress(
201 self, podcast_uuid: str, episode_uuid: str, position_seconds: int
202 ) -> None:
203 """
204 Update playback progress for an episode (marks it in progress).
205
206 :param podcast_uuid: The podcast UUID.
207 :param episode_uuid: The episode UUID.
208 :param position_seconds: Current playback position in seconds.
209 """
210 await self._request(
211 "POST",
212 f"{API_BASE_URL}/sync/update_episode",
213 json={
214 "uuid": episode_uuid,
215 "podcast": podcast_uuid,
216 "status": 2, # 2=in_progress
217 "position": str(position_seconds),
218 },
219 )
220
221 async def mark_episode_played(self, podcast_uuid: str, episode_uuid: str) -> None:
222 """
223 Mark an episode as played.
224
225 :param podcast_uuid: The podcast UUID.
226 :param episode_uuid: The episode UUID.
227 """
228 await self._request(
229 "POST",
230 f"{API_BASE_URL}/sync/update_episode",
231 json={"uuid": episode_uuid, "podcast": podcast_uuid, "status": 3}, # 3=played
232 )
233
234 async def mark_episode_unplayed(self, podcast_uuid: str, episode_uuid: str) -> None:
235 """
236 Mark an episode as unplayed and reset its position.
237
238 :param podcast_uuid: The podcast UUID.
239 :param episode_uuid: The episode UUID.
240 """
241 await self._request(
242 "POST",
243 f"{API_BASE_URL}/sync/update_episode",
244 json={
245 "uuid": episode_uuid,
246 "podcast": podcast_uuid,
247 "status": 1, # 1=unplayed
248 "position": "0",
249 },
250 )
251
252 async def archive_episode(
253 self, podcast_uuid: str, episode_uuid: str, archive: bool = True
254 ) -> None:
255 """
256 Archive or unarchive an episode.
257
258 :param podcast_uuid: The podcast UUID.
259 :param episode_uuid: The episode UUID.
260 :param archive: True to archive, False to unarchive.
261 """
262 await self._request(
263 "POST",
264 f"{API_BASE_URL}/sync/update_episodes_archive",
265 json={
266 "episodes": [{"uuid": episode_uuid, "podcast": podcast_uuid}],
267 "archive": archive,
268 },
269 )
270
271 async def remove_from_up_next(self, episode_uuid: str) -> None:
272 """
273 Remove an episode from the Up Next queue.
274
275 :param episode_uuid: The episode UUID to remove.
276 """
277 await self._request(
278 "POST",
279 f"{API_BASE_URL}/up_next/remove",
280 json={"version": 2, "uuids": [episode_uuid]},
281 )
282
283 async def play_now(
284 self,
285 episode_uuid: str,
286 podcast_uuid: str,
287 title: str,
288 url: str,
289 published: str | None = None,
290 ) -> None:
291 """
292 Add an episode to the top of the Up Next queue.
293
294 :param episode_uuid: The episode UUID.
295 :param podcast_uuid: The podcast UUID.
296 :param title: The episode title.
297 :param url: The episode audio URL.
298 :param published: The episode publish date (ISO format), optional.
299 """
300 episode: dict[str, Any] = {
301 "uuid": episode_uuid,
302 "podcast": podcast_uuid,
303 "title": title,
304 "url": url,
305 }
306 if published:
307 episode["published"] = published
308 await self._request(
309 "POST", f"{API_BASE_URL}/up_next/play_now", json={"version": 2, "episode": episode}
310 )
311
312 async def add_to_history(
313 self,
314 episode_uuid: str,
315 podcast_uuid: str,
316 title: str,
317 url: str,
318 published: str | None = None,
319 ) -> None:
320 """
321 Record an episode in the listening history.
322
323 :param episode_uuid: The episode UUID.
324 :param podcast_uuid: The podcast UUID.
325 :param title: The episode title.
326 :param url: The episode audio URL.
327 :param published: The episode publish date (ISO format), optional.
328 """
329 payload: dict[str, Any] = {
330 "action": 1,
331 "podcast": podcast_uuid,
332 "episode": episode_uuid,
333 "title": title,
334 "url": url,
335 }
336 if published:
337 payload["published"] = published
338 await self._request("POST", f"{API_BASE_URL}/history/do", json=payload)
339
340 async def subscribe_podcast(self, podcast_uuid: str) -> None:
341 """
342 Subscribe to a podcast.
343
344 :param podcast_uuid: The UUID of the podcast to subscribe to.
345 """
346 await self._request(
347 "POST", f"{API_BASE_URL}/user/podcast/subscribe", json={"uuid": podcast_uuid}
348 )
349
350 async def unsubscribe_podcast(self, podcast_uuid: str) -> None:
351 """
352 Unsubscribe from a podcast.
353
354 :param podcast_uuid: The UUID of the podcast to unsubscribe from.
355 """
356 await self._request(
357 "POST", f"{API_BASE_URL}/user/podcast/unsubscribe", json={"uuid": podcast_uuid}
358 )
359
360 def _headers(self) -> dict[str, str]:
361 if not self.token:
362 raise LoginFailed("Not logged in to Pocket Casts")
363 return {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
364
365 @throttle_with_retries
366 async def _request(
367 self, method: str, url: str, *, auth: bool = True, **kwargs: Any
368 ) -> dict[str, Any]:
369 """
370 Perform a request against the Pocket Casts API and return the decoded JSON body.
371
372 :param method: The HTTP method to use.
373 :param url: The full request URL.
374 :param auth: Whether to send the authorization header.
375 """
376 headers = self._headers() if auth else None
377 try:
378 async with self.session.request(method, url, headers=headers, **kwargs) as response:
379 if response.status in (401, 403):
380 raise LoginFailed(f"Pocket Casts authentication failed ({response.status})")
381 if response.status == 429 or response.status >= 500:
382 # transient: let the throttler back off and retry
383 raise ResourceTemporarilyUnavailable(
384 f"Pocket Casts temporarily unavailable ({response.status})",
385 backoff_time=parse_retry_after(response.headers.get("Retry-After")),
386 )
387 if response.status != 200:
388 text = await response.text()
389 raise ProviderUnavailableError(
390 f"Pocket Casts request to {url} failed ({response.status}): {text}"
391 )
392 return cast("dict[str, Any]", await response.json(loads=json_loads))
393 except aiohttp.ClientError as err:
394 raise ResourceTemporarilyUnavailable(
395 f"Network error contacting Pocket Casts: {err}"
396 ) from err
397