/
/
/
1"""Helpers/utilities for the Internet Archive provider."""
2
3from __future__ import annotations
4
5import json
6import re
7from typing import TYPE_CHECKING, Any
8from urllib.parse import quote
9
10import aiohttp
11from music_assistant_models.errors import (
12 InvalidDataError,
13 MediaNotFoundError,
14 RateLimited,
15 ResourceTemporarilyUnavailable,
16)
17
18from .constants import (
19 IA_DETAILS_URL,
20 IA_DOWNLOAD_URL,
21 IA_METADATA_URL,
22 IA_SEARCH_URL,
23 PREFERRED_AUDIO_FORMATS,
24 SUPPORTED_AUDIO_FORMATS,
25)
26
27if TYPE_CHECKING:
28 from music_assistant import MusicAssistant
29
30
31class InternetArchiveClient:
32 """Client for communicating with the Internet Archive API."""
33
34 def __init__(self, mass: MusicAssistant) -> None:
35 """Initialize the Internet Archive client."""
36 self.mass = mass
37
38 async def _get_json(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
39 """Make a GET request and return JSON response with proper error handling."""
40 try:
41 async with self.mass.http_session.get(
42 url, params=params, timeout=aiohttp.ClientTimeout(total=30)
43 ) as response:
44 if response.status == 429:
45 # Rate limited - let throttler handle this
46 backoff_time = int(response.headers.get("Retry-After", 60))
47 raise RateLimited(
48 "Internet Archive rate limit exceeded", backoff_time=backoff_time
49 )
50
51 if response.status == 404:
52 raise MediaNotFoundError("Item not found on Internet Archive")
53
54 if response.status >= 500:
55 raise ResourceTemporarilyUnavailable(
56 "Internet Archive server error", backoff_time=30
57 )
58
59 response.raise_for_status()
60 json_data = await response.json()
61
62 if not isinstance(json_data, dict):
63 raise InvalidDataError(f"Expected JSON object, got {type(json_data).__name__}")
64
65 return json_data
66
67 except aiohttp.ClientError as err:
68 raise ResourceTemporarilyUnavailable(f"Network error: {err}") from err
69 except TimeoutError as err:
70 raise ResourceTemporarilyUnavailable(f"Request timeout: {err}") from err
71 except json.JSONDecodeError as err:
72 raise InvalidDataError(f"Invalid JSON response: {err}") from err
73
74 async def search(
75 self,
76 query: str,
77 mediatype: str | None = None,
78 collection: str | None = None,
79 rows: int = 50,
80 page: int = 1,
81 sort: str | None = None,
82 ) -> dict[str, Any]:
83 """
84 Search the Internet Archive using the advanced search API.
85
86 Args:
87 query: Search query string
88 mediatype: Optional media type filter (e.g., 'audio')
89 collection: Optional collection filter (e.g., 'etree')
90 rows: Number of results per page (max 200)
91 page: Page number for pagination
92 sort: Sort order (e.g., 'downloads desc', 'date desc')
93
94 Returns:
95 Search response dictionary containing results and metadata
96 """
97 params: dict[str, Any] = {
98 "output": "json",
99 "rows": min(rows, 200), # IA limits to 200 per request
100 "page": page,
101 "q": query,
102 }
103 if sort:
104 params["sort"] = sort
105
106 return await self._get_json(IA_SEARCH_URL, params)
107
108 async def get_metadata(self, identifier: str) -> dict[str, Any]:
109 """Get metadata for a specific Internet Archive item."""
110 url = f"{IA_METADATA_URL}/{identifier}"
111 return await self._get_json(url)
112
113 async def get_files(self, identifier: str) -> list[dict[str, Any]]:
114 """Get file list for an Internet Archive item."""
115 metadata = await self.get_metadata(identifier)
116 return list(metadata.get("files", []))
117
118 async def get_audio_files(self, identifier: str) -> list[dict[str, Any]]:
119 """
120 Get audio files for an item with format preference and deduplication.
121
122 Filters for supported audio formats, removes derivative low-quality files,
123 deduplicates by base filename, and selects the best quality format for
124 each unique track.
125
126 Args:
127 identifier: Internet Archive item identifier
128
129 Returns:
130 List of audio file information dictionaries, sorted by filename
131 for proper track ordering
132 """
133 files = await self.get_files(identifier)
134 files_by_basename: dict[str, list[dict[str, Any]]] = {}
135
136 for file_info in files:
137 filename = file_info.get("name", "")
138 file_format = file_info.get("format", "").lower()
139
140 if not self._is_supported_audio_format(file_format):
141 continue
142 if self._is_derivative_file(file_info, filename):
143 continue
144
145 base_name = self._get_base_filename(filename)
146 files_by_basename.setdefault(base_name, []).append(file_info)
147
148 preferred_files: list[dict[str, Any]] = []
149 for format_versions in files_by_basename.values():
150 best_file = self._select_best_audio_format(format_versions)
151 if best_file:
152 preferred_files.append(best_file)
153
154 return sorted(preferred_files, key=lambda x: x.get("name", ""))
155
156 def _is_supported_audio_format(self, file_format: str) -> bool:
157 """Check if the file format is a supported audio format."""
158 return any(fmt in file_format for fmt in SUPPORTED_AUDIO_FORMATS)
159
160 def _is_derivative_file(self, file_info: dict[str, Any], filename: str) -> bool:
161 """Check if a file is a derivative (low-quality) version."""
162 return file_info.get("source", "") == "derivative" and any(
163 skip in filename.lower() for skip in ("_64kb", "_vbr", "_sample", "_preview")
164 )
165
166 def _get_base_filename(self, filename: str) -> str:
167 """Extract base filename without extension and quality indicators for deduplication."""
168 # Remove extension first
169 base = filename.rsplit(".", 1)[0] if "." in filename else filename
170
171 # Remove common quality indicators from Internet Archive files
172 quality_patterns = [
173 r"_320kb$",
174 r"_256kb$",
175 r"_192kb$",
176 r"_128kb$",
177 r"_64kb$",
178 r"_vbr$",
179 r"_original$",
180 r"_sample$",
181 r"_preview$",
182 ]
183
184 for pattern in quality_patterns:
185 base = re.sub(pattern, "", base, flags=re.IGNORECASE)
186
187 return base
188
189 def _select_best_audio_format(
190 self, format_versions: list[dict[str, Any]]
191 ) -> dict[str, Any] | None:
192 """
193 Select the best audio format from available versions.
194
195 Prefers higher quality formats based on PREFERRED_AUDIO_FORMATS ordering.
196 Falls back to first available if no preferred format is found.
197
198 Args:
199 format_versions: List of file info dictionaries for the same track
200
201 Returns:
202 Best quality file info dictionary, or None if no valid files
203 """
204 for preferred_format in PREFERRED_AUDIO_FORMATS:
205 for file_info in format_versions:
206 if preferred_format in file_info.get("format", "").lower():
207 return file_info
208 return format_versions[0] if format_versions else None
209
210 def get_download_url(self, identifier: str, filename: str) -> str:
211 """
212 Get download URL for a specific file.
213
214 Args:
215 identifier: Internet Archive item identifier
216 filename: Name of the file to download
217
218 Returns:
219 Full download URL for the file
220 """
221 return f"{IA_DOWNLOAD_URL}/{identifier}/{quote(filename)}"
222
223 def get_item_url(self, identifier: str) -> str:
224 """
225 Get the details page URL for an Internet Archive item.
226
227 Args:
228 identifier: Internet Archive item identifier
229
230 Returns:
231 Full URL to the item's details page
232 """
233 return f"{IA_DETAILS_URL}/{identifier}"
234
235
236def parse_duration(duration_str: str) -> int | None:
237 """
238 Parse duration string to seconds.
239
240 Handles various duration formats commonly found in Internet Archive metadata:
241 - "1:23:45" (hours:minutes:seconds)
242 - "12:34" (minutes:seconds)
243 - "123" (seconds only)
244
245 Args:
246 duration_str: Duration string to parse
247
248 Returns:
249 Duration in seconds, or None if parsing fails
250 """
251 if not duration_str:
252 return None
253 try:
254 if ":" in duration_str:
255 parts = duration_str.split(":")
256 if len(parts) == 3: # h:m:s
257 hours, minutes, seconds = map(float, parts)
258 return int(hours * 3600 + minutes * 60 + seconds)
259 if len(parts) == 2: # m:s
260 minutes, seconds = map(float, parts)
261 return int(minutes * 60 + seconds)
262 return None
263 return int(float(duration_str))
264 except ValueError, TypeError:
265 return None
266
267
268def clean_text(text: str | list[str] | None) -> str:
269 """
270 Clean and normalize text fields from Internet Archive metadata.
271
272 Internet Archive metadata can contain text as strings or lists of strings.
273 This function normalizes the input to a clean string.
274
275 Args:
276 text: Text to clean (string, list of strings, or None)
277
278 Returns:
279 Cleaned text string, or empty string if no valid text found
280 """
281 if not text:
282 return ""
283 if isinstance(text, list):
284 for item in text:
285 if isinstance(item, str) and item.strip():
286 return item.strip()
287 return ""
288 return text.strip()
289
290
291def extract_year(date_str: str | list[str] | None) -> int | None:
292 """
293 Extract year from Internet Archive date string.
294
295 Internet Archive dates can be in various formats. This function attempts
296 to extract a 4-digit year from the date string.
297
298 Args:
299 date_str: Date string or list to extract year from
300
301 Returns:
302 4-digit year as integer, or None if extraction fails
303 """
304 date_text = clean_text(date_str)
305 if not date_text:
306 return None
307 try:
308 match = re.search(r"\b(19\d{2}|20\d{2})\b", date_text)
309 return int(match.group(1)) if match else None
310 except ValueError, TypeError:
311 return None
312
313
314def get_image_url(identifier: str, filename: str | None = None) -> str | None:
315 """
316 Get image URL for an Internet Archive item.
317
318 Args:
319 identifier: Internet Archive item identifier
320 filename: Optional specific image filename
321
322 Returns:
323 Full URL to the image, or None if identifier is missing
324 """
325 if not identifier:
326 return None
327 if filename:
328 return f"{IA_DOWNLOAD_URL}/{identifier}/{quote(filename)}"
329 return f"{IA_DOWNLOAD_URL}/{identifier}/__ia_thumb.jpg"
330