/
/
/
1"""Helper functions for Podcast Index provider."""
2
3from __future__ import annotations
4
5import hashlib
6import logging
7import time
8from datetime import UTC, datetime
9from typing import TYPE_CHECKING, Any
10
11import aiohttp
12from music_assistant_models.enums import ContentType, ImageType, LinkType, MediaType
13from music_assistant_models.errors import (
14 InvalidDataError,
15 LoginFailed,
16 ProviderUnavailableError,
17)
18from music_assistant_models.media_items import (
19 AudioFormat,
20 ItemMapping,
21 MediaItemImage,
22 MediaItemLink,
23 Podcast,
24 PodcastEpisode,
25 ProviderMapping,
26 UniqueList,
27)
28
29from music_assistant.helpers.podcast_parsers import parse_podcast_persons
30
31from .constants import (
32 API_BASE_URL,
33 HTTP_STATUS_ERROR,
34 HTTP_STATUS_UNAUTHORIZED,
35 MAX_ERROR_DETAIL_LENGTH,
36)
37
38if TYPE_CHECKING:
39 from music_assistant.mass import MusicAssistant
40
41
42async def make_api_request(
43 mass: MusicAssistant,
44 api_key: str,
45 api_secret: str,
46 endpoint: str,
47 params: dict[str, Any] | None = None,
48 logger: logging.Logger | None = None,
49) -> dict[str, Any]:
50 """
51 Make an authenticated request to the Podcast Index API and return its payload.
52
53 :param mass: The Music Assistant instance whose http session is used.
54 :param api_key: The API key issued by Podcast Index.
55 :param api_secret: The API secret issued by Podcast Index.
56 :param endpoint: The API endpoint to call, without a leading slash.
57 :param params: Optional query parameters for the endpoint.
58 :param logger: Optional logger, used to record the outcome of the call.
59 :raises LoginFailed: The credentials were rejected.
60 :raises ProviderUnavailableError: The API could not be reached or refused the call.
61 :raises InvalidDataError: The API answered with something unusable.
62 """
63 auth_date = str(int(time.time()))
64 auth_string = api_key + api_secret + auth_date
65 auth_hash = hashlib.sha1(auth_string.encode()).hexdigest()
66
67 headers = {
68 "X-Auth-Key": api_key,
69 "X-Auth-Date": auth_date,
70 "Authorization": auth_hash,
71 }
72
73 url = f"{API_BASE_URL}/{endpoint}"
74
75 try:
76 async with mass.http_session.get(url, headers=headers, params=params or {}) as response:
77 if response.status >= HTTP_STATUS_ERROR:
78 # the body says why a key was refused, which the bare status never does
79 detail = _response_detail(await response.text(), api_key, api_secret)
80 if logger:
81 logger.debug("%s failed with HTTP %s%s", endpoint, response.status, detail)
82 if response.status == HTTP_STATUS_UNAUTHORIZED:
83 raise LoginFailed(f"Authentication failed (HTTP {response.status}){detail}")
84 raise ProviderUnavailableError(
85 f"API request failed (HTTP {response.status}){detail}"
86 )
87
88 try:
89 data: dict[str, Any] = await response.json()
90 except aiohttp.ContentTypeError as err:
91 raise InvalidDataError("Invalid JSON response from API") from err
92
93 if str(data.get("status")).lower() != "true":
94 description = data.get("description") or "API error"
95 if logger:
96 logger.debug("%s was refused by the API: %s", endpoint, description)
97 raise InvalidDataError(description)
98
99 if logger:
100 # the single item endpoints report no count, so do not invent one for them
101 if (count := data.get("count")) is None:
102 logger.debug("%s succeeded", endpoint)
103 else:
104 logger.debug("%s returned %s items", endpoint, count)
105 return data
106
107 except aiohttp.ClientConnectorError as err:
108 raise ProviderUnavailableError(f"Failed to connect to Podcast Index API: {err}") from err
109 except aiohttp.ServerTimeoutError as err:
110 raise ProviderUnavailableError(f"Podcast Index API timeout: {err}") from err
111
112
113def parse_podcast_from_feed(
114 feed_data: dict[str, Any], instance_id: str, domain: str
115) -> Podcast | None:
116 """Parse podcast from API feed data."""
117 feed_url = feed_data.get("url")
118 podcast_id = feed_data.get("id")
119
120 if not feed_url or not podcast_id:
121 return None
122
123 podcast = Podcast(
124 item_id=str(podcast_id),
125 name=feed_data.get("title", "Unknown Podcast"),
126 publisher=feed_data.get("author") or feed_data.get("ownerName", "Unknown"),
127 provider=instance_id,
128 provider_mappings={
129 ProviderMapping(
130 item_id=str(podcast_id),
131 provider_domain=domain,
132 provider_instance=instance_id,
133 url=feed_url,
134 )
135 },
136 )
137
138 # Add metadata
139 podcast.metadata.description = feed_data.get("description", "")
140 podcast.metadata.explicit = bool(feed_data.get("explicit", False))
141
142 # Set episode count only if provided
143 episode_count = feed_data.get("episodeCount")
144 if episode_count is not None:
145 podcast.total_episodes = int(episode_count) or 0
146
147 # Add image - prefer 'image' field, fallback to 'artwork'
148 image_url = feed_data.get("image") or feed_data.get("artwork")
149 if image_url:
150 podcast.metadata.add_image(
151 MediaItemImage(
152 type=ImageType.THUMB,
153 path=image_url,
154 provider=instance_id,
155 remotely_accessible=True,
156 )
157 )
158
159 # Add categories as genres - categories is a dict {id: name}
160 categories = feed_data.get("categories", {})
161 if categories and isinstance(categories, dict):
162 podcast.metadata.genres = set(categories.values())
163
164 # Add language
165 language = feed_data.get("language", "")
166 if language:
167 podcast.metadata.languages = UniqueList([language])
168
169 return podcast
170
171
172def parse_episode_from_data(
173 episode_data: dict[str, Any],
174 podcast_id: str,
175 instance_id: str,
176 domain: str,
177 podcast_name: str | None = None,
178 position: int = 0,
179) -> PodcastEpisode | None:
180 """
181 Parse episode from API episode data.
182
183 :param position: The episode's listing position. Defaults to 0 (unknown).
184 """
185 episode_api_id = episode_data.get("id")
186 if not episode_api_id:
187 return None
188
189 episode_id = f"{podcast_id}|{episode_api_id}"
190
191 if podcast_name is None:
192 podcast_name = episode_data.get("feedTitle") or "Unknown Podcast"
193
194 raw_duration = episode_data.get("duration")
195 try:
196 duration = int(raw_duration) if raw_duration is not None else 0
197 except ValueError, TypeError:
198 duration = 0
199
200 episode = PodcastEpisode(
201 item_id=episode_id,
202 provider=instance_id,
203 name=episode_data.get("title", "Unknown Episode"),
204 duration=duration,
205 position=position,
206 podcast=ItemMapping(
207 item_id=podcast_id,
208 provider=instance_id,
209 name=podcast_name,
210 media_type=MediaType.PODCAST,
211 ),
212 provider_mappings={
213 ProviderMapping(
214 item_id=episode_id,
215 provider_domain=domain,
216 provider_instance=instance_id,
217 available=True,
218 audio_format=AudioFormat(
219 content_type=ContentType.try_parse(
220 episode_data.get("enclosureType") or "audio/mpeg"
221 ),
222 ),
223 url=episode_data.get("enclosureUrl"),
224 )
225 },
226 )
227
228 # Add metadata
229 episode.metadata.description = episode_data.get("description", "")
230 episode.metadata.explicit = bool(episode_data.get("explicit", 0))
231
232 # hosts/guests (Podcast Index persons array), mapped to performer names
233 if performers := parse_podcast_persons(episode_data.get("persons")):
234 episode.metadata.performers = set(performers)
235
236 # episode webpage
237 if link := episode_data.get("link"):
238 episode.metadata.links = {MediaItemLink(type=LinkType.WEBSITE, url=link)}
239
240 date_published = episode_data.get("datePublished")
241 if date_published:
242 episode.metadata.release_date = datetime.fromtimestamp(date_published, tz=UTC)
243
244 image_url = episode_data.get("image") or episode_data.get("feedImage")
245 if image_url:
246 episode.metadata.add_image(
247 MediaItemImage(
248 type=ImageType.THUMB,
249 path=image_url,
250 provider=instance_id,
251 remotely_accessible=True,
252 )
253 )
254
255 return episode
256
257
258def _response_detail(body: str, *secrets: str) -> str:
259 """
260 Return an error body condensed into a readable suffix, empty when it says nothing.
261
262 :param body: The raw response body.
263 :param secrets: Values to mask, as the body is quoted back to the user.
264 """
265 detail = " ".join(body.split())
266 if not detail:
267 return ""
268 for secret in secrets:
269 if secret:
270 detail = detail.replace(secret, "***")
271 if len(detail) > MAX_ERROR_DETAIL_LENGTH:
272 detail = f"{detail[:MAX_ERROR_DETAIL_LENGTH]}..."
273 return f": {detail}"
274