/
/
/
1"""Helper functions for Podcast Index provider."""
2
3from __future__ import annotations
4
5import hashlib
6import time
7from datetime import UTC, datetime
8from typing import TYPE_CHECKING, Any
9
10import aiohttp
11from music_assistant_models.enums import ContentType, ImageType, LinkType, MediaType
12from music_assistant_models.errors import (
13 InvalidDataError,
14 LoginFailed,
15 ProviderUnavailableError,
16)
17from music_assistant_models.media_items import (
18 AudioFormat,
19 ItemMapping,
20 MediaItemImage,
21 MediaItemLink,
22 Podcast,
23 PodcastEpisode,
24 ProviderMapping,
25 UniqueList,
26)
27
28from music_assistant.helpers.podcast_parsers import parse_podcast_persons
29
30from .constants import API_BASE_URL
31
32if TYPE_CHECKING:
33 from music_assistant.mass import MusicAssistant
34
35
36async def make_api_request(
37 mass: MusicAssistant,
38 api_key: str,
39 api_secret: str,
40 endpoint: str,
41 params: dict[str, Any] | None = None,
42) -> dict[str, Any]:
43 """
44 Make authenticated request to Podcast Index API.
45
46 Handles authentication using SHA1 hash of API key, secret, and timestamp.
47 Maps HTTP errors appropriately: 401 -> LoginFailed, others -> ProviderUnavailableError.
48 """
49 # Prepare authentication headers
50 auth_date = str(int(time.time()))
51 auth_string = api_key + api_secret + auth_date
52 auth_hash = hashlib.sha1(auth_string.encode()).hexdigest()
53
54 headers = {
55 "X-Auth-Key": api_key,
56 "X-Auth-Date": auth_date,
57 "Authorization": auth_hash,
58 }
59
60 url = f"{API_BASE_URL}/{endpoint}"
61
62 try:
63 async with mass.http_session.get(url, headers=headers, params=params or {}) as response:
64 response.raise_for_status()
65
66 try:
67 data: dict[str, Any] = await response.json()
68 except aiohttp.ContentTypeError as err:
69 raise InvalidDataError("Invalid JSON response from API") from err
70
71 if str(data.get("status")).lower() != "true":
72 raise InvalidDataError(data.get("description") or "API error")
73
74 return data
75
76 except aiohttp.ClientConnectorError as err:
77 raise ProviderUnavailableError(f"Failed to connect to Podcast Index API: {err}") from err
78 except aiohttp.ServerTimeoutError as err:
79 raise ProviderUnavailableError(f"Podcast Index API timeout: {err}") from err
80 except aiohttp.ClientResponseError as err:
81 if err.status == 401:
82 raise LoginFailed(f"Authentication failed: {err.status}") from err
83 raise ProviderUnavailableError(f"API request failed: {err.status}") from err
84
85
86def parse_podcast_from_feed(
87 feed_data: dict[str, Any], instance_id: str, domain: str
88) -> Podcast | None:
89 """Parse podcast from API feed data."""
90 feed_url = feed_data.get("url")
91 podcast_id = feed_data.get("id")
92
93 if not feed_url or not podcast_id:
94 return None
95
96 podcast = Podcast(
97 item_id=str(podcast_id),
98 name=feed_data.get("title", "Unknown Podcast"),
99 publisher=feed_data.get("author") or feed_data.get("ownerName", "Unknown"),
100 provider=instance_id,
101 provider_mappings={
102 ProviderMapping(
103 item_id=str(podcast_id),
104 provider_domain=domain,
105 provider_instance=instance_id,
106 url=feed_url,
107 )
108 },
109 )
110
111 # Add metadata
112 podcast.metadata.description = feed_data.get("description", "")
113 podcast.metadata.explicit = bool(feed_data.get("explicit", False))
114
115 # Set episode count only if provided
116 episode_count = feed_data.get("episodeCount")
117 if episode_count is not None:
118 podcast.total_episodes = int(episode_count) or 0
119
120 # Add image - prefer 'image' field, fallback to 'artwork'
121 image_url = feed_data.get("image") or feed_data.get("artwork")
122 if image_url:
123 podcast.metadata.add_image(
124 MediaItemImage(
125 type=ImageType.THUMB,
126 path=image_url,
127 provider=instance_id,
128 remotely_accessible=True,
129 )
130 )
131
132 # Add categories as genres - categories is a dict {id: name}
133 categories = feed_data.get("categories", {})
134 if categories and isinstance(categories, dict):
135 podcast.metadata.genres = set(categories.values())
136
137 # Add language
138 language = feed_data.get("language", "")
139 if language:
140 podcast.metadata.languages = UniqueList([language])
141
142 return podcast
143
144
145def parse_episode_from_data(
146 episode_data: dict[str, Any],
147 podcast_id: str,
148 instance_id: str,
149 domain: str,
150 podcast_name: str | None = None,
151 position: int = 0,
152) -> PodcastEpisode | None:
153 """
154 Parse episode from API episode data.
155
156 :param position: The episode's listing position. Defaults to 0 (unknown).
157 """
158 episode_api_id = episode_data.get("id")
159 if not episode_api_id:
160 return None
161
162 episode_id = f"{podcast_id}|{episode_api_id}"
163
164 if podcast_name is None:
165 podcast_name = episode_data.get("feedTitle") or "Unknown Podcast"
166
167 raw_duration = episode_data.get("duration")
168 try:
169 duration = int(raw_duration) if raw_duration is not None else 0
170 except ValueError, TypeError:
171 duration = 0
172
173 episode = PodcastEpisode(
174 item_id=episode_id,
175 provider=instance_id,
176 name=episode_data.get("title", "Unknown Episode"),
177 duration=duration,
178 position=position,
179 podcast=ItemMapping(
180 item_id=podcast_id,
181 provider=instance_id,
182 name=podcast_name,
183 media_type=MediaType.PODCAST,
184 ),
185 provider_mappings={
186 ProviderMapping(
187 item_id=episode_id,
188 provider_domain=domain,
189 provider_instance=instance_id,
190 available=True,
191 audio_format=AudioFormat(
192 content_type=ContentType.try_parse(
193 episode_data.get("enclosureType") or "audio/mpeg"
194 ),
195 ),
196 url=episode_data.get("enclosureUrl"),
197 )
198 },
199 )
200
201 # Add metadata
202 episode.metadata.description = episode_data.get("description", "")
203 episode.metadata.explicit = bool(episode_data.get("explicit", 0))
204
205 # hosts/guests (Podcast Index persons array), mapped to performer names
206 if performers := parse_podcast_persons(episode_data.get("persons")):
207 episode.metadata.performers = set(performers)
208
209 # episode webpage
210 if link := episode_data.get("link"):
211 episode.metadata.links = {MediaItemLink(type=LinkType.WEBSITE, url=link)}
212
213 date_published = episode_data.get("datePublished")
214 if date_published:
215 episode.metadata.release_date = datetime.fromtimestamp(date_published, tz=UTC)
216
217 image_url = episode_data.get("image") or episode_data.get("feedImage")
218 if image_url:
219 episode.metadata.add_image(
220 MediaItemImage(
221 type=ImageType.THUMB,
222 path=image_url,
223 provider=instance_id,
224 remotely_accessible=True,
225 )
226 )
227
228 return episode
229