/
/
/
1"""
2Simplest client for gPodder.
3
4Should be compatible with Nextcloud App GPodder Sync, and the original api
5of gpodder.net (mygpo) or drop-in replacements like opodsync.
6Gpodder Sync uses guid optionally.
7"""
8
9import datetime
10import logging
11from contextlib import suppress
12from dataclasses import dataclass, field
13from typing import Any
14
15import aiohttp
16from aiohttp.client_exceptions import ClientResponseError
17from mashumaro.config import BaseConfig
18from mashumaro.mixins.json import DataClassJSONMixin
19from mashumaro.types import Discriminator
20
21from music_assistant.helpers.datetime import utc
22
23
24# https://gpoddernet.readthedocs.io/en/latest/api/reference/subscriptions.html#upload-subscription-changes
25@dataclass(kw_only=True)
26class SubscriptionsChangeRequest(DataClassJSONMixin):
27 """SubscriptionChangeRequest."""
28
29 add: list[str] = field(default_factory=list)
30 remove: list[str] = field(default_factory=list)
31
32
33# https://gpoddernet.readthedocs.io/en/latest/api/reference/subscriptions.html#upload-subscription-changes
34@dataclass(kw_only=True)
35class SubscriptionsGet(SubscriptionsChangeRequest):
36 """SubscriptionsGet."""
37
38 timestamp: int
39
40
41def action_tagger(cls: type[EpisodeAction]) -> list[str]:
42 """
43 Use action field to distinguish classes.
44
45 NC Gpodder uses upper case values, opodsync lower case.
46 This however does not work with a StrEnum, so plain string as action.
47 """
48 action = cls.__name__.replace("EpisodeAction", "")
49 return [action.upper(), action.lower()]
50
51
52@dataclass(kw_only=True)
53class EpisodeAction(DataClassJSONMixin):
54 """
55 General EpisodeAction.
56
57 See https://gpoddernet.readthedocs.io/en/latest/api/reference/events.html
58 """
59
60 class Config(BaseConfig):
61 """Config."""
62
63 discriminator = Discriminator(
64 field="action", include_subtypes=True, variant_tagger_fn=action_tagger
65 )
66 omit_none = True # only nextcloud supports guid
67
68 podcast: str
69 episode: str
70 timestamp: str = ""
71 guid: str | None = None
72
73
74@dataclass(kw_only=True)
75class EpisodeActionDownload(EpisodeAction):
76 """EpisodeActionDownload."""
77
78 action: str = "download"
79
80
81@dataclass(kw_only=True)
82class EpisodeActionDelete(EpisodeAction):
83 """EpisodeActionDelete."""
84
85 action: str = "delete"
86
87
88@dataclass(kw_only=True)
89class EpisodeActionNew(EpisodeAction):
90 """EpisodeActionNew."""
91
92 action: str = "new"
93
94
95@dataclass(kw_only=True)
96class EpisodeActionFlattr(EpisodeAction):
97 """EpisodeActionFlattr."""
98
99 action: str = "flattr"
100
101
102@dataclass(kw_only=True)
103class EpisodeActionPlay(EpisodeAction):
104 """EpisodeActionPlay."""
105
106 action: str = "play"
107
108 # all in seconds
109 started: int = 0
110 position: int = 0
111 total: int = 0
112
113
114@dataclass(kw_only=True)
115class EpisodeActionGet(DataClassJSONMixin):
116 """EpisodeActionGet."""
117
118 actions: list[EpisodeAction]
119 timestamp: int
120
121
122class GPodderClient:
123 """GPodderClient."""
124
125 def __init__(
126 self, session: aiohttp.ClientSession, logger: logging.Logger, verify_ssl: bool = True
127 ) -> None:
128 """Init for GPodderClient."""
129 self.session = session
130 self.verify_ssl = verify_ssl
131
132 self.is_nextcloud = False
133 self.base_url: str
134 self.token: str | None
135
136 self.username: str
137 self.device: str
138 self.auth: aiohttp.BasicAuth | None = None # only for gpodder
139
140 self.logger = logger
141
142 self._nextcloud_prefix = "index.php/apps/gpoddersync"
143
144 def init_nc(self, base_url: str, nc_token: str | None = None) -> None:
145 """Init values for a nextcloud client."""
146 self.is_nextcloud = True
147 self.token = nc_token
148 self.base_url = base_url.rstrip("/")
149
150 async def init_gpodder(self, username: str, password: str, device: str, base_url: str) -> None:
151 """Init via basic auth."""
152 self.username = username
153 self.device = device
154 self.base_url = base_url.rstrip("/")
155 self.auth = aiohttp.BasicAuth(username, password)
156 await self._post(endpoint=f"api/2/auth/{username}/login.json")
157
158 @property
159 def headers(self) -> dict[str, str]:
160 """Session headers."""
161 if self.token is None:
162 raise RuntimeError("Token not set.")
163 return {"Authorization": f"Bearer {self.token}"}
164
165 async def _post(
166 self,
167 endpoint: str,
168 data: dict[str, Any] | list[Any] | None = None,
169 ) -> bytes:
170 """POST request."""
171 try:
172 response = await self.session.post(
173 f"{self.base_url}/{endpoint}",
174 json=data,
175 ssl=self.verify_ssl,
176 headers=self.headers if self.is_nextcloud else None,
177 raise_for_status=True,
178 auth=self.auth,
179 )
180 except ClientResponseError as exc:
181 self.logger.debug(exc)
182 raise RuntimeError(f"API POST call to {endpoint} failed.") from exc
183 if response.status != 200:
184 self.logger.debug(f"Call failed with status {response.status}")
185 raise RuntimeError(f"Api post call failed to {endpoint} failed!")
186 return await response.read()
187
188 async def _get(self, endpoint: str, params: dict[str, str | int] | None = None) -> bytes:
189 """GET request."""
190 response = await self.session.get(
191 f"{self.base_url}/{endpoint}",
192 params=params,
193 ssl=self.verify_ssl,
194 headers=self.headers if self.is_nextcloud else None,
195 auth=self.auth,
196 )
197 status = response.status
198 if response.content_type == "application/json" and status == 200:
199 return await response.read()
200 if status == 404:
201 return b""
202 self.logger.debug(f"Call failed with status {response.status}")
203 raise RuntimeError(f"API GET call to {endpoint} failed.")
204
205 async def get_subscriptions(self, since: int = 0) -> SubscriptionsGet | None:
206 """
207 Get subscriptions.
208
209 since is unix time epoch - this may return none if there are no
210 subscriptions.
211 """
212 if self.is_nextcloud:
213 endpoint = f"{self._nextcloud_prefix}/subscriptions"
214 else:
215 endpoint = f"api/2/subscriptions/{self.username}/{self.device}.json"
216
217 response = await self._get(endpoint, params={"since": since})
218 if not response:
219 return None
220 return SubscriptionsGet.from_json(response)
221
222 async def get_episode_actions(
223 self, since: int = 0
224 ) -> tuple[list[EpisodeActionPlay | EpisodeActionNew | EpisodeActionDelete], int | None]:
225 """
226 Get progresses or deletions. Timestamp is second return value.
227
228 gpodder net may filter by podcast
229 https://gpoddernet.readthedocs.io/en/latest/api/reference/events.html
230 -> we do not use this for now, since nextcloud implementation is not
231 capable of it. Also, implementation in drop-in replacements varies.
232
233 Play holds progress information.
234 New is a marked unplayed.
235 Delete is used if the user deletes a previously downloaded episode.
236 """
237 params: dict[str, str | int] = {"since": since}
238 if self.is_nextcloud:
239 endpoint = f"{self._nextcloud_prefix}/episode_action"
240 else:
241 endpoint = f"api/2/episodes/{self.username}.json"
242 params["device"] = self.device
243 response = await self._get(endpoint, params=params)
244 if not response:
245 return [], None
246 actions_response = EpisodeActionGet.from_json(response)
247
248 # play has progress information
249 # new means, there is no progress (i.e. mark unplayed)
250 actions = [
251 x
252 for x in actions_response.actions
253 if isinstance(x, EpisodeActionPlay | EpisodeActionNew | EpisodeActionDelete)
254 ]
255
256 with suppress(ValueError):
257 actions = sorted(actions, key=lambda x: datetime.datetime.fromisoformat(x.timestamp))[
258 ::-1
259 ]
260
261 return actions, actions_response.timestamp
262
263 async def update_subscriptions(
264 self, add: list[str] | None = None, remove: list[str] | None = None
265 ) -> None:
266 """Update subscriptions."""
267 if add is None:
268 add = []
269 if remove is None:
270 remove = []
271 request = SubscriptionsChangeRequest(add=add, remove=remove)
272 if self.is_nextcloud:
273 endpoint = f"{self._nextcloud_prefix}/subscription_change/create"
274 else:
275 endpoint = f"api/2/subscriptions/{self.username}/{self.device}.json"
276
277 await self._post(endpoint=endpoint, data=request.to_dict())
278
279 async def update_progress(
280 self,
281 *,
282 podcast_id: str,
283 episode_id: str,
284 guid: str | None,
285 position_s: float,
286 duration_s: float,
287 ) -> None:
288 """Update progress."""
289 utc_timestamp = utc().replace(microsecond=0, tzinfo=None).isoformat()
290
291 episode_action: EpisodeActionNew | EpisodeActionPlay
292 if position_s == 0:
293 # mark unplayed
294 episode_action = EpisodeActionNew(
295 podcast=podcast_id, episode=episode_id, timestamp=utc_timestamp
296 )
297 else:
298 episode_action = EpisodeActionPlay(
299 podcast=podcast_id,
300 episode=episode_id,
301 timestamp=utc_timestamp,
302 position=int(position_s),
303 started=0,
304 total=int(duration_s),
305 )
306
307 # It is a bit unclear here, if other gpodder alternatives then nextcloud support the guid
308 # for episodes. I didn't see that in the source for opodsync at least...
309 if self.is_nextcloud:
310 episode_action.guid = guid
311 endpoint = f"{self._nextcloud_prefix}/episode_action/create"
312 else:
313 endpoint = f"api/2/episodes/{self.username}.json"
314 await self._post(endpoint=endpoint, data=[episode_action.to_dict()])
315