music-assistant-server

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) -> list[dict[str, Any]]:
85        """
86        Return all episodes for a podcast.
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 episodes
98
99    async def get_in_progress_episodes(self) -> list[dict[str, Any]]:
100        """Return episodes currently in progress."""
101        data = await self._request("POST", f"{API_BASE_URL}/user/in_progress")
102        episodes: list[dict[str, Any]] = data.get("episodes", [])
103        self.logger.debug("Retrieved %d in-progress episodes", len(episodes))
104        return episodes
105
106    async def get_up_next_episodes(self) -> list[dict[str, Any]]:
107        """Return the Up Next queue episodes."""
108        data = await self._request("POST", f"{API_BASE_URL}/up_next/list")
109        episodes = data.get("episodes", [])
110        # the up_next endpoint returns a uuid-keyed map; normalise to a list carrying the uuid
111        if isinstance(episodes, dict):
112            return [{"uuid": uuid, **episode} for uuid, episode in episodes.items()]
113        return cast("list[dict[str, Any]]", episodes)
114
115    async def get_new_releases(self) -> list[dict[str, Any]]:
116        """Return new release episodes from subscriptions."""
117        data = await self._request("POST", f"{API_BASE_URL}/user/new_releases")
118        episodes: list[dict[str, Any]] = data.get("episodes", [])
119        self.logger.debug("Retrieved %d new release episodes", len(episodes))
120        return episodes
121
122    async def get_starred_episodes(self) -> list[dict[str, Any]]:
123        """Return starred episodes."""
124        data = await self._request("POST", f"{API_BASE_URL}/user/starred")
125        episodes: list[dict[str, Any]] = data.get("episodes", [])
126        self.logger.debug("Retrieved %d starred episodes", len(episodes))
127        return episodes
128
129    async def get_history(self) -> list[dict[str, Any]]:
130        """Return listening history episodes."""
131        data = await self._request("POST", f"{API_BASE_URL}/user/history")
132        episodes: list[dict[str, Any]] = data.get("episodes", [])
133        self.logger.debug("Retrieved %d history episodes", len(episodes))
134        return episodes
135
136    async def get_episode_details(self, episode_uuid: str) -> dict[str, Any]:
137        """
138        Return detailed episode info including correct duration and playback status.
139
140        :param episode_uuid: The episode UUID.
141        """
142        # /user/episode returns camelCase keys: uuid, title, url, fileType, duration (seconds),
143        # published, episodeNumber, playedUpTo (resume seconds), playingStatus (1=unplayed,
144        # 2=in progress, 3=played), starred, podcastUuid. No show notes or episode artwork.
145        data = await self._request(
146            "POST", f"{API_BASE_URL}/user/episode", json={"uuid": episode_uuid}
147        )
148        self.logger.debug(
149            "Episode %s: duration=%s, status=%s, playedUpTo=%s",
150            episode_uuid,
151            data.get("duration"),
152            data.get("playingStatus"),
153            data.get("playedUpTo"),
154        )
155        return data
156
157    async def search_podcasts(self, query: str) -> list[dict[str, Any]]:
158        """
159        Search for podcasts.
160
161        :param query: The search term.
162        """
163        data = await self._request("POST", f"{API_BASE_URL}/discover/search", json={"term": query})
164        podcasts: list[dict[str, Any]] = data.get("podcasts", [])
165        self.logger.debug("Found %d podcasts for query '%s'", len(podcasts), query)
166        return podcasts
167
168    async def update_episode_progress(
169        self, podcast_uuid: str, episode_uuid: str, position_seconds: int
170    ) -> None:
171        """
172        Update playback progress for an episode (marks it in progress).
173
174        :param podcast_uuid: The podcast UUID.
175        :param episode_uuid: The episode UUID.
176        :param position_seconds: Current playback position in seconds.
177        """
178        await self._request(
179            "POST",
180            f"{API_BASE_URL}/sync/update_episode",
181            json={
182                "uuid": episode_uuid,
183                "podcast": podcast_uuid,
184                "status": 2,  # 2=in_progress
185                "position": str(position_seconds),
186            },
187        )
188
189    async def mark_episode_played(self, podcast_uuid: str, episode_uuid: str) -> None:
190        """
191        Mark an episode as played.
192
193        :param podcast_uuid: The podcast UUID.
194        :param episode_uuid: The episode UUID.
195        """
196        await self._request(
197            "POST",
198            f"{API_BASE_URL}/sync/update_episode",
199            json={"uuid": episode_uuid, "podcast": podcast_uuid, "status": 3},  # 3=played
200        )
201
202    async def mark_episode_unplayed(self, podcast_uuid: str, episode_uuid: str) -> None:
203        """
204        Mark an episode as unplayed and reset its position.
205
206        :param podcast_uuid: The podcast UUID.
207        :param episode_uuid: The episode UUID.
208        """
209        await self._request(
210            "POST",
211            f"{API_BASE_URL}/sync/update_episode",
212            json={
213                "uuid": episode_uuid,
214                "podcast": podcast_uuid,
215                "status": 1,  # 1=unplayed
216                "position": "0",
217            },
218        )
219
220    async def archive_episode(
221        self, podcast_uuid: str, episode_uuid: str, archive: bool = True
222    ) -> None:
223        """
224        Archive or unarchive an episode.
225
226        :param podcast_uuid: The podcast UUID.
227        :param episode_uuid: The episode UUID.
228        :param archive: True to archive, False to unarchive.
229        """
230        await self._request(
231            "POST",
232            f"{API_BASE_URL}/sync/update_episodes_archive",
233            json={
234                "episodes": [{"uuid": episode_uuid, "podcast": podcast_uuid}],
235                "archive": archive,
236            },
237        )
238
239    async def remove_from_up_next(self, episode_uuid: str) -> None:
240        """
241        Remove an episode from the Up Next queue.
242
243        :param episode_uuid: The episode UUID to remove.
244        """
245        await self._request(
246            "POST",
247            f"{API_BASE_URL}/up_next/remove",
248            json={"version": 2, "uuids": [episode_uuid]},
249        )
250
251    async def play_now(
252        self,
253        episode_uuid: str,
254        podcast_uuid: str,
255        title: str,
256        url: str,
257        published: str | None = None,
258    ) -> None:
259        """
260        Add an episode to the top of the Up Next queue.
261
262        :param episode_uuid: The episode UUID.
263        :param podcast_uuid: The podcast UUID.
264        :param title: The episode title.
265        :param url: The episode audio URL.
266        :param published: The episode publish date (ISO format), optional.
267        """
268        episode: dict[str, Any] = {
269            "uuid": episode_uuid,
270            "podcast": podcast_uuid,
271            "title": title,
272            "url": url,
273        }
274        if published:
275            episode["published"] = published
276        await self._request(
277            "POST", f"{API_BASE_URL}/up_next/play_now", json={"version": 2, "episode": episode}
278        )
279
280    async def add_to_history(
281        self,
282        episode_uuid: str,
283        podcast_uuid: str,
284        title: str,
285        url: str,
286        published: str | None = None,
287    ) -> None:
288        """
289        Record an episode in the listening history.
290
291        :param episode_uuid: The episode UUID.
292        :param podcast_uuid: The podcast UUID.
293        :param title: The episode title.
294        :param url: The episode audio URL.
295        :param published: The episode publish date (ISO format), optional.
296        """
297        payload: dict[str, Any] = {
298            "action": 1,
299            "podcast": podcast_uuid,
300            "episode": episode_uuid,
301            "title": title,
302            "url": url,
303        }
304        if published:
305            payload["published"] = published
306        await self._request("POST", f"{API_BASE_URL}/history/do", json=payload)
307
308    async def subscribe_podcast(self, podcast_uuid: str) -> None:
309        """
310        Subscribe to a podcast.
311
312        :param podcast_uuid: The UUID of the podcast to subscribe to.
313        """
314        await self._request(
315            "POST", f"{API_BASE_URL}/user/podcast/subscribe", json={"uuid": podcast_uuid}
316        )
317
318    async def unsubscribe_podcast(self, podcast_uuid: str) -> None:
319        """
320        Unsubscribe from a podcast.
321
322        :param podcast_uuid: The UUID of the podcast to unsubscribe from.
323        """
324        await self._request(
325            "POST", f"{API_BASE_URL}/user/podcast/unsubscribe", json={"uuid": podcast_uuid}
326        )
327
328    def _headers(self) -> dict[str, str]:
329        if not self.token:
330            raise LoginFailed("Not logged in to Pocket Casts")
331        return {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
332
333    @throttle_with_retries
334    async def _request(
335        self, method: str, url: str, *, auth: bool = True, **kwargs: Any
336    ) -> dict[str, Any]:
337        """
338        Perform a request against the Pocket Casts API and return the decoded JSON body.
339
340        :param method: The HTTP method to use.
341        :param url: The full request URL.
342        :param auth: Whether to send the authorization header.
343        """
344        headers = self._headers() if auth else None
345        try:
346            async with self.session.request(method, url, headers=headers, **kwargs) as response:
347                if response.status in (401, 403):
348                    raise LoginFailed(f"Pocket Casts authentication failed ({response.status})")
349                if response.status == 429 or response.status >= 500:
350                    # transient: let the throttler back off and retry
351                    raise ResourceTemporarilyUnavailable(
352                        f"Pocket Casts temporarily unavailable ({response.status})",
353                        backoff_time=parse_retry_after(response.headers.get("Retry-After")),
354                    )
355                if response.status != 200:
356                    text = await response.text()
357                    raise ProviderUnavailableError(
358                        f"Pocket Casts request to {url} failed ({response.status}): {text}"
359                    )
360                return cast("dict[str, Any]", await response.json(loads=json_loads))
361        except aiohttp.ClientError as err:
362            raise ResourceTemporarilyUnavailable(
363                f"Network error contacting Pocket Casts: {err}"
364            ) from err
365