/
/
/
1"""Helper for parsing and using audible api."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import html
8import json
9import logging
10import os
11import re
12from collections.abc import AsyncGenerator
13from contextlib import suppress
14from datetime import UTC, datetime, timedelta
15from os import PathLike
16from typing import TYPE_CHECKING, Any
17from urllib.parse import parse_qs, urlparse
18
19import audible
20import audible.exceptions
21import audible.register
22from audible import AsyncClient
23
24if TYPE_CHECKING:
25 from aiohttp import ClientSession
26
27 from music_assistant.models.music_provider import MusicProvider
28from music_assistant_models.enums import ContentType, ImageType, MediaType, StreamType
29from music_assistant_models.errors import (
30 LoginFailed,
31 MediaNotFoundError,
32 ProviderUnavailableError,
33)
34from music_assistant_models.media_items import (
35 Audiobook,
36 AudioFormat,
37 ItemMapping,
38 MediaItemChapter,
39 MediaItemImage,
40 Podcast,
41 PodcastEpisode,
42 ProviderMapping,
43 UniqueList,
44)
45from music_assistant_models.streamdetails import StreamDetails
46
47from music_assistant.helpers.datetime import utc
48from music_assistant.helpers.podcast_parsers import rank_episodes_by_date
49from music_assistant.mass import MusicAssistant
50
51CACHE_DOMAIN = "audible"
52CACHE_CATEGORY_API = 0
53CACHE_CATEGORY_AUDIOBOOK = 1
54CACHE_CATEGORY_CHAPTERS = 2
55CACHE_CATEGORY_PODCAST = 3
56CACHE_CATEGORY_PODCAST_EPISODES = 4
57
58# Content delivery types
59AUDIOBOOK_CONTENT_TYPES = ("SinglePartBook", "MultiPartBook")
60# Podcasts are normally reported as "PodcastParent", but (older) Audible Original
61# series are still reported with the legacy "Periodical" delivery type.
62PODCAST_CONTENT_TYPES = ("PodcastParent", "Periodical")
63# legacy series report their episodes as show issues rather than podcast episodes
64SHOW_CONTENT_TYPE = "Show"
65
66_AUTH_CACHE: dict[str, audible.Authenticator] = {}
67
68
69async def refresh_access_token_compat(
70 refresh_token: str, domain: str, http_session: ClientSession, with_username: bool = False
71) -> dict[str, Any]:
72 """
73 Refresh tokens with compatibility for new Audible API format.
74
75 The Audible API changed from returning 'access_token' to 'actor_access_token'.
76 This function handles both formats for backward compatibility.
77
78 :param refresh_token: The refresh token obtained after device registration.
79 :param domain: The top level domain (e.g., com, de).
80 :param http_session: The HTTP client session to use for requests.
81 :param with_username: If True, use audible domain instead of amazon.
82 :return: Dict with access_token and expires timestamp.
83 """
84 logger = logging.getLogger("audible_helper")
85
86 body = {
87 "app_name": "Audible",
88 "app_version": "3.56.2",
89 "source_token": refresh_token,
90 "requested_token_type": "access_token",
91 "source_token_type": "refresh_token",
92 }
93
94 target_domain = "audible" if with_username else "amazon"
95 url = f"https://api.{target_domain}.{domain}/auth/token"
96
97 async with http_session.post(url, data=body) as resp:
98 resp.raise_for_status()
99 resp_dict = await resp.json()
100
101 expires_in_sec = int(resp_dict.get("expires_in", 3600))
102 expires = (utc() + timedelta(seconds=expires_in_sec)).timestamp()
103
104 # Handle new format (actor_access_token) or fall back to legacy (access_token)
105 access_token = resp_dict.get("actor_access_token") or resp_dict.get("access_token")
106
107 if not access_token:
108 logger.error("Token refresh response missing both actor_access_token and access_token")
109 raise LoginFailed("Token refresh failed: no access token in response")
110
111 logger.debug(
112 "Token refreshed successfully using %s format",
113 "new (actor)" if "actor_access_token" in resp_dict else "legacy",
114 )
115
116 return {"access_token": access_token, "expires": expires}
117
118
119async def cached_authenticator_from_file(
120 path: str, locale: str | None = None
121) -> audible.Authenticator:
122 """
123 Get an authenticator from file with caching and signing auth validation.
124
125 :param path: Path to the authenticator JSON file.
126 :param locale: The configured marketplace locale; when the stored file disagrees,
127 the configured locale wins and the file is corrected.
128 :return: The cached or loaded Authenticator instance.
129 """
130 logger = logging.getLogger("audible_helper")
131 auth = _AUTH_CACHE.get(path)
132 if auth is None:
133 logger.debug("Loading authenticator from file %s and caching it", path)
134 auth = await asyncio.to_thread(audible.Authenticator.from_file, path)
135
136 # Verify signing auth is available (not affected by API changes)
137 if auth.adp_token and auth.device_private_key:
138 logger.debug("Signing auth available - using stable RSA-signed requests")
139 else:
140 logger.warning(
141 "Signing auth not available - only bearer auth will work. "
142 "Consider re-authenticating for more stable auth."
143 )
144
145 _AUTH_CACHE[path] = auth
146
147 # auth files written by older versions can hold the marketplace from before a
148 # locale change; the configured locale is authoritative, so correct the file
149 if locale and (auth.locale is None or auth.locale.country_code != locale):
150 logger.warning(
151 "Marketplace in auth file (%s) does not match the configured locale (%s), correcting",
152 auth.locale.country_code if auth.locale else None,
153 locale,
154 )
155 auth.locale = audible.localization.Locale(locale)
156 await asyncio.to_thread(auth.to_file, path)
157
158 return auth
159
160
161def evict_cached_authenticator(path: str) -> None:
162 """
163 Drop the cached authenticator for the given file path, if any.
164
165 :param path: Path to the authenticator JSON file.
166 """
167 _AUTH_CACHE.pop(path, None)
168
169
170async def deregister_auth_file(path: str) -> None:
171 """
172 Deregister the virtual device registration stored in the given auth file.
173
174 :param path: Path to the authenticator JSON file.
175 """
176 auth = _AUTH_CACHE.pop(path, None)
177 if auth is None:
178 auth = await asyncio.to_thread(audible.Authenticator.from_file, path)
179 await asyncio.to_thread(auth.deregister_device)
180
181
182class AudibleHelper:
183 """Helper for parsing and using audible api."""
184
185 def __init__(
186 self,
187 mass: MusicAssistant,
188 client: AsyncClient,
189 provider_domain: str,
190 provider_instance: str,
191 provider: MusicProvider,
192 logger: logging.Logger | None = None,
193 ):
194 """
195 Initialize the Audible Helper.
196
197 :param mass: The MusicAssistant instance.
198 :param client: An authenticated Audible API client.
199 :param provider_domain: Domain of the owning provider.
200 :param provider_instance: Instance id of the owning provider.
201 :param provider: The owning provider, used to report library items it had to skip.
202 :param logger: Logger to use, defaults to a module level logger.
203 """
204 self.mass = mass
205 self.client = client
206 self.provider_domain = provider_domain
207 self.provider_instance = provider_instance
208 self.provider = provider
209 self.logger = logger or logging.getLogger("audible_helper")
210 self._acr_cache: dict[tuple[str, MediaType], str] = {}
211
212 async def _fetch_library_items(
213 self,
214 response_groups: str,
215 content_types: tuple[str, ...],
216 ) -> AsyncGenerator[dict[str, Any]]:
217 """Fetch items from the library with pagination."""
218 page = 1
219 page_size = 50
220 total_processed = 0
221 max_iterations = 100
222 iteration = 0
223
224 while iteration < max_iterations:
225 iteration += 1
226 self.logger.debug(
227 "Audible: Fetching library page %s (processed so far: %s)",
228 page,
229 total_processed,
230 )
231
232 library = await self._call_api(
233 "library",
234 use_cache=False,
235 response_groups=response_groups,
236 page=page,
237 num_results=page_size,
238 )
239
240 items = library.get("items", [])
241
242 if not items:
243 break
244
245 items_processed_this_page = 0
246 for item in items:
247 # Filter by content type if specified
248 if content_types and item.get("content_delivery_type") not in content_types:
249 continue
250
251 yield item
252 items_processed_this_page += 1
253 total_processed += 1
254
255 self.logger.debug(
256 "Audible: Processed %s items on page %s", items_processed_this_page, page
257 )
258
259 page += 1
260 if len(items) < page_size:
261 break
262
263 if iteration >= max_iterations:
264 self.logger.warning(
265 "Audible: Reached maximum iteration limit (%s) with %s items processed",
266 max_iterations,
267 total_processed,
268 )
269
270 async def _process_audiobook_item(self, audiobook_data: dict[str, Any]) -> Audiobook | None:
271 """Process a single audiobook item from the library."""
272 # Ensure asin is a valid string
273 asin = str(audiobook_data.get("asin", ""))
274 cached_book = None
275 if asin:
276 cached_book = await self.mass.cache.get(
277 key=asin,
278 provider=self.provider_instance,
279 category=CACHE_CATEGORY_AUDIOBOOK,
280 default=None,
281 )
282
283 try:
284 if cached_book is not None:
285 return self._parse_audiobook(cached_book)
286 return self._parse_audiobook(audiobook_data)
287 except Exception as exc:
288 self.provider.report_skipped_sync_item(MediaType.AUDIOBOOK, asin or None, exc)
289 return None
290
291 async def get_library(self) -> AsyncGenerator[Audiobook]:
292 """Fetch the user's library with pagination."""
293 response_groups = [
294 "contributors",
295 "media",
296 "product_attrs",
297 "product_desc",
298 "product_details",
299 "product_extended_attrs",
300 ]
301
302 async for item in self._fetch_library_items(
303 ",".join(response_groups), AUDIOBOOK_CONTENT_TYPES
304 ):
305 if album := await self._process_audiobook_item(item):
306 yield album
307
308 async def get_audiobook(self, asin: str, use_cache: bool = True) -> Audiobook:
309 """
310 Fetch the full audiobook by asin with all details including chapters.
311
312 This method fetches complete audiobook details including chapters and resume position.
313 Use this when the user requests full details for a specific audiobook.
314 """
315 if use_cache:
316 cached_book = await self.mass.cache.get(
317 key=asin,
318 provider=self.provider_instance,
319 category=CACHE_CATEGORY_AUDIOBOOK,
320 default=None,
321 )
322 if cached_book is not None:
323 book = self._parse_audiobook(cached_book)
324 # Enrich with chapters and resume position
325 await self._enrich_audiobook(book, asin)
326 return book
327 response = await self._call_api(
328 f"library/{asin}",
329 response_groups="""
330 contributors, media, price, product_attrs, product_desc, product_details,
331 product_extended_attrs,is_finished
332 """,
333 )
334
335 if response is None:
336 raise MediaNotFoundError(f"Audiobook with ASIN {asin} not found")
337
338 item_data = response.get("item")
339 if item_data is None:
340 raise MediaNotFoundError(f"Audiobook data for ASIN {asin} is empty")
341
342 await self.mass.cache.set(
343 key=asin,
344 provider=self.provider_instance,
345 category=CACHE_CATEGORY_AUDIOBOOK,
346 data=item_data,
347 )
348 book = self._parse_audiobook(item_data)
349 # Enrich with chapters and resume position
350 await self._enrich_audiobook(book, asin)
351 return book
352
353 async def _enrich_audiobook(self, book: Audiobook, asin: str) -> None:
354 """
355 Enrich audiobook with chapters and resume position.
356
357 This makes additional API calls and should only be used for full audiobook details,
358 not during library sync.
359 """
360 # Fetch chapters
361 chapters_data = await self._fetch_chapters(asin=asin)
362 if chapters_data:
363 chapters: list[MediaItemChapter] = [
364 self._parse_chapter_data(chapter, idx) for idx, chapter in enumerate(chapters_data)
365 ]
366 book.metadata.chapters = chapters
367 # Update duration from chapters if available (more accurate)
368 try:
369 duration = int(sum(chapter.get("length_ms", 0) for chapter in chapters_data) / 1000)
370 if duration > 0:
371 book.duration = duration
372 except Exception as exc:
373 self.logger.warning(f"Error calculating duration from chapters for {asin}: {exc}")
374
375 # Fetch resume position
376 book.resume_position_ms = await self.get_last_position(asin=asin)
377
378 async def get_stream(
379 self, asin: str, media_type: MediaType = MediaType.AUDIOBOOK
380 ) -> StreamDetails:
381 """
382 Get stream details for an audiobook or podcast episode.
383
384 :param asin: The ASIN of the content.
385 :param media_type: The type of media (audiobook or podcast episode).
386 """
387 if not asin:
388 self.logger.error("Invalid ASIN provided to get_stream")
389 raise ValueError("Invalid ASIN provided to get_stream")
390
391 duration = 0
392 # For audiobooks, try to get duration from chapters
393 if media_type == MediaType.AUDIOBOOK:
394 chapters = await self._fetch_chapters(asin=asin)
395 if chapters:
396 try:
397 duration = int(sum(chapter.get("length_ms", 0) for chapter in chapters) / 1000)
398 except Exception as exc:
399 self.logger.warning(f"Error calculating duration for ASIN {asin}: {exc}")
400
401 try:
402 # Podcasts use Mpeg (non-DRM MP3), audiobooks use HLS
403 if media_type == MediaType.PODCAST_EPISODE:
404 playback_info = await self.client.post(
405 f"content/{asin}/licenserequest",
406 body={
407 "consumption_type": "Streaming",
408 "drm_type": "Mpeg",
409 "quality": "High",
410 },
411 )
412 else:
413 playback_info = await self.client.post(
414 f"content/{asin}/licenserequest",
415 body={
416 "quality": "High",
417 "response_groups": "content_reference,certificate",
418 "consumption_type": "Streaming",
419 "supported_media_features": {
420 "codecs": ["mp4a.40.2", "mp4a.40.42"],
421 "drm_types": [
422 "Hls",
423 ],
424 },
425 "spatial": False,
426 },
427 )
428
429 content_license = playback_info.get("content_license", {})
430 if not content_license:
431 self.logger.error(f"No content_license in playback_info for ASIN {asin}")
432 raise ValueError(f"Missing content_license for ASIN {asin}")
433
434 content_metadata = content_license.get("content_metadata", {})
435 content_reference = content_metadata.get("content_reference", {})
436 size = content_reference.get("content_size_in_bytes", 0)
437
438 stream_url = content_license.get("license_response")
439 if not stream_url:
440 self.logger.error(f"No license_response (stream URL) for ASIN {asin}")
441 raise ValueError(f"Missing stream URL for ASIN {asin}")
442
443 acr = content_license.get("acr", "")
444 if acr:
445 self._acr_cache[(asin, media_type)] = acr
446
447 content_type = (
448 ContentType.MP3 if media_type == MediaType.PODCAST_EPISODE else ContentType.AAC
449 )
450 except Exception as exc:
451 self.logger.error(f"Error getting stream details for ASIN {asin}: {exc}")
452 raise ValueError(f"Failed to get stream details: {exc}") from exc
453
454 return StreamDetails(
455 provider=self.provider_instance,
456 size=size,
457 item_id=f"{asin}",
458 audio_format=AudioFormat(content_type=content_type),
459 media_type=media_type,
460 stream_type=StreamType.HTTP,
461 path=stream_url,
462 can_seek=True,
463 allow_seek=True,
464 duration=duration,
465 data={"acr": acr},
466 )
467
468 async def _fetch_chapters(self, asin: str) -> list[dict[str, Any]]:
469 """Fetch chapter data for an audiobook."""
470 if not asin or asin == "error":
471 self.logger.warning(
472 "Invalid ASIN provided to _fetch_chapters, returning empty chapter list"
473 )
474 return []
475
476 chapters_data: list[Any] = await self.mass.cache.get(
477 key=asin, provider=self.provider_instance, category=CACHE_CATEGORY_CHAPTERS, default=[]
478 )
479
480 if not chapters_data:
481 try:
482 response = await self._call_api(
483 f"content/{asin}/metadata",
484 response_groups="chapter_info, always-returned, content_reference, content_url",
485 chapter_titles_type="Flat",
486 )
487
488 if not response:
489 self.logger.warning(f"Failed to get metadata for ASIN {asin}")
490 return []
491
492 content_metadata = response.get("content_metadata")
493 if not content_metadata:
494 self.logger.warning(f"No content_metadata for ASIN {asin}")
495 return []
496
497 chapter_info = content_metadata.get("chapter_info")
498 if not chapter_info:
499 self.logger.warning(f"No chapter_info for ASIN {asin}")
500 return []
501
502 chapters_data = chapter_info.get("chapters") or []
503
504 await self.mass.cache.set(
505 key=asin,
506 data=chapters_data,
507 provider=self.provider_instance,
508 category=CACHE_CATEGORY_CHAPTERS,
509 )
510 except Exception as exc:
511 self.logger.error(f"Error fetching chapters for ASIN {asin}: {exc}")
512 chapters_data = []
513
514 return chapters_data
515
516 @staticmethod
517 def _parse_audible_timestamp(raw_ts: Any) -> datetime | None:
518 """
519 Parse an Audible timestamp value into a timezone-aware datetime.
520
521 :param raw_ts: The raw timestamp value from the Audible annotation payload.
522 """
523 if not raw_ts:
524 return None
525 try:
526 parsed = datetime.fromisoformat(str(raw_ts))
527 except ValueError, TypeError:
528 return None
529 if parsed.tzinfo is None:
530 parsed = parsed.replace(tzinfo=UTC)
531 return parsed
532
533 async def _fetch_last_position(self, asin: str) -> tuple[int, datetime | None] | None:
534 """
535 Fetch the last-heard position for a single ASIN from Audible.
536
537 :param asin: The audiobook ASIN to query.
538 """
539 response = await self._call_api("annotations/lastpositions", asins=asin)
540 if not response:
541 return None
542
543 annotations = response.get("asin_last_position_heard_annots")
544 if not annotations or not isinstance(annotations, list):
545 return None
546
547 annotation = annotations[0]
548 if not isinstance(annotation, dict):
549 return None
550
551 last_position = annotation.get("last_position_heard")
552 if not isinstance(last_position, dict):
553 return None
554
555 position_ms = int(last_position.get("position_ms", 0))
556
557 timestamp: datetime | None = None
558 for field in ("last_updated", "reported_time", "last_updated_time", "timestamp"):
559 timestamp = self._parse_audible_timestamp(
560 last_position.get(field) or annotation.get(field)
561 )
562 if timestamp is not None:
563 break
564
565 return position_ms, timestamp
566
567 async def get_last_position(self, asin: str) -> int:
568 """
569 Fetch the last-heard position in milliseconds for the given ASIN.
570
571 :param asin: The audiobook ASIN to query.
572 """
573 if not asin or asin == "error":
574 return 0
575 try:
576 result = await self._fetch_last_position(asin)
577 except (ProviderUnavailableError, KeyError, TypeError, ValueError) as exc:
578 self.logger.error("Error getting last position for ASIN %s: %s", asin, exc)
579 return 0
580 return result[0] if result else 0
581
582 async def set_last_position(
583 self, asin: str, pos: int, media_type: MediaType = MediaType.AUDIOBOOK
584 ) -> None:
585 """
586 Report last position to Audible.
587
588 :param asin: The content ID (audiobook or podcast episode).
589 :param pos: Position in seconds.
590 :param media_type: The type of media (audiobook or podcast episode).
591 """
592 if not asin or asin == "error" or pos <= 0:
593 return
594
595 try:
596 position_ms = pos * 1000
597
598 # Try to get ACR from cache first
599 acr = self._acr_cache.get((asin, media_type))
600 if not acr:
601 stream_details = await self.get_stream(asin=asin, media_type=media_type)
602 acr = stream_details.data.get("acr")
603
604 if not acr:
605 self.logger.warning(f"No ACR available for ASIN {asin}, cannot report position")
606 return
607
608 await self.client.put(
609 f"lastpositions/{asin}", body={"acr": acr, "asin": asin, "position_ms": position_ms}
610 )
611
612 self.logger.debug(f"Successfully reported position {position_ms}ms for ASIN {asin}")
613
614 except (KeyError, TypeError) as exc:
615 self.logger.error(
616 f"Error accessing data while reporting position for ASIN {asin}: {exc}"
617 )
618 except TimeoutError as exc:
619 self.logger.error(f"Timeout while reporting position for ASIN {asin}: {exc}")
620 except ConnectionError as exc:
621 self.logger.error(f"Connection error while reporting position for ASIN {asin}: {exc}")
622 except Exception as exc:
623 self.logger.error(f"Unexpected error reporting position for ASIN {asin}: {exc}")
624
625 async def get_audible_resume_position(self, asin: str) -> tuple[bool, int, datetime | None]:
626 """
627 Return resume state for the given ASIN from Audible.
628
629 :param asin: The audiobook ASIN to query.
630 """
631 if not asin or asin == "error":
632 raise NotImplementedError
633 try:
634 result = await self._fetch_last_position(asin)
635 except (ProviderUnavailableError, KeyError, TypeError, ValueError) as exc:
636 self.logger.debug("Audible lastpositions fetch failed for %s: %s", asin, exc)
637 raise NotImplementedError from exc
638 if not result or result[0] == 0:
639 raise NotImplementedError
640 position_ms, timestamp = result
641 return False, position_ms, timestamp
642
643 async def _call_api(self, path: str, **kwargs: Any) -> Any:
644 response = None
645 use_cache = kwargs.pop("use_cache", False)
646 params_str = json.dumps(kwargs, sort_keys=True)
647 params_hash = hashlib.md5(params_str.encode()).hexdigest()
648 cache_key_with_params = f"{path}:{params_hash}"
649 if use_cache:
650 response = await self.mass.cache.get(
651 key=cache_key_with_params,
652 provider=self.provider_instance,
653 category=CACHE_CATEGORY_API,
654 )
655 if not response:
656 try:
657 response = await self.client.get(path, **kwargs)
658 except audible.exceptions.RequestError as exc:
659 raise ProviderUnavailableError(
660 f"Audible API request failed for '{path}': {exc}"
661 ) from exc
662 await self.mass.cache.set(
663 key=cache_key_with_params, provider=self.provider_instance, data=response
664 )
665 return response
666
667 def _parse_contributors(
668 self, contributors_list: list[dict[str, Any]] | None, default_name: str
669 ) -> list[str]:
670 """Parse contributors (authors, narrators) from API response."""
671 result: list[str] = []
672 contributors: list[dict[str, Any]] = contributors_list or []
673 if isinstance(contributors, list):
674 for contributor in contributors:
675 if contributor and isinstance(contributor, dict):
676 result.append(contributor.get("name", default_name))
677 return result
678
679 def _create_images(self, image_path: str | None) -> list[MediaItemImage]:
680 """Create image objects if image path exists."""
681 images: list[MediaItemImage] = []
682 if image_path:
683 images.append(
684 MediaItemImage(
685 type=ImageType.THUMB,
686 path=image_path,
687 provider=self.provider_instance,
688 remotely_accessible=True,
689 )
690 )
691 images.append(
692 MediaItemImage(
693 type=ImageType.CLEARART,
694 path=image_path,
695 provider=self.provider_instance,
696 remotely_accessible=True,
697 )
698 )
699 return images
700
701 def _parse_chapter_data(self, chapter_data: dict[str, Any], index: int) -> MediaItemChapter:
702 """Parse chapter data into MediaItemChapter object."""
703 try:
704 start = int(chapter_data.get("start_offset_sec", 0))
705 except TypeError, ValueError:
706 start = 0
707
708 try:
709 length = int(chapter_data.get("length_ms", 0)) / 1000
710 except TypeError, ValueError:
711 length = 0
712
713 raw_title = chapter_data.get("title")
714 chapter_title: str
715 if raw_title is None:
716 chapter_title = f"Chapter {index + 1}"
717 elif isinstance(raw_title, str):
718 chapter_title = raw_title
719 else:
720 chapter_title = str(raw_title)
721
722 return MediaItemChapter(position=index, name=chapter_title, start=start, end=start + length)
723
724 def _parse_audiobook(self, audiobook_data: dict[str, Any] | None) -> Audiobook:
725 """
726 Parse audiobook data from API response.
727
728 NOTE: This is a pure parser - no API calls allowed here.
729 Chapters and resume position are fetched lazily when needed.
730 """
731 if audiobook_data is None:
732 self.logger.error("Received None audiobook_data in _parse_audiobook")
733 raise MediaNotFoundError("Audiobook data not found")
734
735 asin = audiobook_data.get("asin", "")
736 title = audiobook_data.get("title", "")
737
738 # Parse authors and narrators
739 narrators = self._parse_contributors(audiobook_data.get("narrators"), "Unknown Narrator")
740 authors = self._parse_contributors(audiobook_data.get("authors"), "Unknown Author")
741
742 # Get duration from runtime_length_min (provided by 'media' response group)
743 # Chapters are fetched lazily when streaming, not during library sync
744 runtime_minutes = audiobook_data.get("runtime_length_min", 0)
745 duration = runtime_minutes * 60 if runtime_minutes else 0
746
747 # Create audiobook object
748 book = Audiobook(
749 item_id=asin,
750 provider=self.provider_instance,
751 name=title,
752 duration=duration,
753 provider_mappings={
754 ProviderMapping(
755 item_id=asin,
756 provider_domain=self.provider_domain,
757 provider_instance=self.provider_instance,
758 )
759 },
760 publisher=audiobook_data.get("publisher_name"),
761 authors=UniqueList(authors),
762 narrators=UniqueList(narrators),
763 )
764
765 # Set metadata
766 book.metadata.copyright = audiobook_data.get("copyright")
767 book.metadata.description = _html_to_txt(
768 str(audiobook_data.get("extended_product_description", ""))
769 )
770 book.metadata.languages = UniqueList([audiobook_data.get("language") or ""])
771 if release_date := audiobook_data.get("release_date"):
772 with suppress(ValueError):
773 parsed_date = datetime.strptime(release_date, "%Y-%m-%d").astimezone(UTC)
774 book.metadata.release_date = parsed_date
775
776 # Set review if available
777 reviews = audiobook_data.get("editorial_reviews", [])
778 if reviews and reviews[0]:
779 book.metadata.review = _html_to_txt(str(reviews[0]))
780
781 # Set genres
782 book.metadata.genres = {
783 genre.replace("_", " ") for genre in (audiobook_data.get("platinum_keywords") or [])
784 }
785
786 # Add images
787 image_path = audiobook_data.get("product_images", {}).get("500")
788 book.metadata.images = UniqueList(self._create_images(image_path))
789
790 # Chapters are not fetched during parsing - they are fetched lazily when streaming
791 # This avoids N+1 API calls during library sync
792
793 return book
794
795 async def _process_podcast_item(self, podcast_data: dict[str, Any]) -> Podcast | None:
796 """Process a single podcast item from the library."""
797 asin = str(podcast_data.get("asin", ""))
798 cached_podcast = None
799 if asin:
800 cached_podcast = await self.mass.cache.get(
801 key=asin,
802 provider=self.provider_instance,
803 category=CACHE_CATEGORY_PODCAST,
804 default=None,
805 )
806
807 try:
808 if cached_podcast is not None:
809 return self._parse_podcast(cached_podcast)
810 return self._parse_podcast(podcast_data)
811 except Exception as exc:
812 self.provider.report_skipped_sync_item(MediaType.PODCAST, asin or None, exc)
813 return None
814
815 async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
816 """Fetch podcasts from the user's library with pagination."""
817 response_groups = [
818 "contributors",
819 "media",
820 "product_attrs",
821 "product_desc",
822 "product_details",
823 "product_extended_attrs",
824 ]
825
826 async for item in self._fetch_library_items(
827 ",".join(response_groups), PODCAST_CONTENT_TYPES
828 ):
829 if podcast := await self._process_podcast_item(item):
830 yield podcast
831
832 async def get_podcast(self, asin: str, use_cache: bool = True) -> Podcast:
833 """
834 Fetch full podcast details by ASIN.
835
836 :param asin: The ASIN of the podcast.
837 :param use_cache: Whether to use cached data if available.
838 """
839 if use_cache:
840 cached_podcast = await self.mass.cache.get(
841 key=asin,
842 provider=self.provider_instance,
843 category=CACHE_CATEGORY_PODCAST,
844 default=None,
845 )
846 if cached_podcast is not None:
847 return self._parse_podcast(cached_podcast)
848
849 response = await self._call_api(
850 f"library/{asin}",
851 response_groups="""
852 contributors, media, price, product_attrs, product_desc, product_details,
853 product_extended_attrs, relationships
854 """,
855 )
856
857 if response is None:
858 raise MediaNotFoundError(f"Podcast with ASIN {asin} not found")
859
860 item_data = response.get("item")
861 if item_data is None:
862 raise MediaNotFoundError(f"Podcast data for ASIN {asin} is empty")
863
864 await self.mass.cache.set(
865 key=asin,
866 provider=self.provider_instance,
867 category=CACHE_CATEGORY_PODCAST,
868 data=item_data,
869 )
870 return self._parse_podcast(item_data)
871
872 async def get_podcast_episodes(self, podcast_asin: str) -> AsyncGenerator[PodcastEpisode]:
873 """
874 Fetch all episodes for a podcast.
875
876 :param podcast_asin: The ASIN of the parent podcast.
877 """
878 podcast = await self.get_podcast(podcast_asin)
879
880 # Fetch episodes - they're typically in relationships or we need to query children
881 response_groups = [
882 "contributors",
883 "media",
884 "product_attrs",
885 "product_desc",
886 "product_details",
887 "relationships",
888 ]
889
890 page = 1
891 page_size = 50
892 all_items: list[dict[str, Any]] = []
893
894 while True:
895 # Query for children of the podcast parent
896 response = await self._call_api(
897 "library",
898 use_cache=False,
899 response_groups=",".join(response_groups),
900 parent_asin=podcast_asin,
901 page=page,
902 num_results=page_size,
903 )
904
905 items = response.get("items", [])
906 if not items:
907 break
908
909 all_items.extend(items)
910
911 page += 1
912 if len(items) < page_size:
913 break
914
915 if all(ep.get("content_type") == SHOW_CONTENT_TYPE for ep in all_items):
916 # a legacy series is released in one go, so its publication timestamps record the
917 # ingestion rather than the episode order; the newest-first listing is all we have
918 positions = [len(all_items) - idx for idx in range(len(all_items))]
919 else:
920 # the API lists most shows newest-first but serialised ones oldest-first, so rank on
921 # the publication timestamp; release_date is date only and cannot separate episodes
922 # that a serialised show published on the same day
923 positions = rank_episodes_by_date([ep.get("publication_datetime") for ep in all_items])
924 for position, episode_data in zip(positions, all_items, strict=True):
925 try:
926 yield self._parse_podcast_episode(episode_data, podcast, position)
927 except (AttributeError, KeyError, TypeError, ValueError) as exc:
928 asin = episode_data.get("asin", "unknown")
929 self.logger.warning(f"Error parsing podcast episode {asin}: {exc}")
930
931 async def get_podcast_episode(self, episode_asin: str) -> PodcastEpisode:
932 """
933 Fetch full podcast episode details by ASIN.
934
935 :param episode_asin: The ASIN of the podcast episode.
936 """
937 response = await self._call_api(
938 f"library/{episode_asin}",
939 response_groups="""
940 contributors, media, price, product_attrs, product_desc, product_details,
941 product_extended_attrs, relationships
942 """,
943 )
944
945 if response is None:
946 raise MediaNotFoundError(f"Podcast episode with ASIN {episode_asin} not found")
947
948 item_data = response.get("item")
949 if item_data is None:
950 raise MediaNotFoundError(f"Podcast episode data for ASIN {episode_asin} is empty")
951
952 # Try to get parent podcast info from relationships
953 podcast: Podcast | None = None
954 relationships = item_data.get("relationships", [])
955 for rel in relationships:
956 if rel.get("relationship_type") == "parent":
957 parent_asin = rel.get("asin")
958 if parent_asin:
959 with suppress(MediaNotFoundError):
960 podcast = await self.get_podcast(parent_asin)
961 break
962
963 return self._parse_podcast_episode(item_data, podcast, 0)
964
965 def _parse_podcast(self, podcast_data: dict[str, Any] | None) -> Podcast:
966 """
967 Parse podcast data from API response.
968
969 :param podcast_data: Raw podcast data from the Audible API.
970 """
971 if podcast_data is None:
972 self.logger.error("Received None podcast_data in _parse_podcast")
973 raise MediaNotFoundError("Podcast data not found")
974
975 asin = podcast_data.get("asin", "")
976 title = podcast_data.get("title", "")
977 publisher = podcast_data.get("publisher_name", "")
978
979 # Create podcast object
980 podcast = Podcast(
981 item_id=asin,
982 provider=self.provider_instance,
983 name=title,
984 publisher=publisher,
985 provider_mappings={
986 ProviderMapping(
987 item_id=asin,
988 provider_domain=self.provider_domain,
989 provider_instance=self.provider_instance,
990 )
991 },
992 )
993
994 # Set metadata
995 podcast.metadata.description = _html_to_txt(
996 str(
997 podcast_data.get("publisher_summary", "")
998 or podcast_data.get("extended_product_description", "")
999 )
1000 )
1001 podcast.metadata.languages = UniqueList([podcast_data.get("language") or ""])
1002
1003 # Set genres
1004 podcast.metadata.genres = {
1005 genre.replace("_", " ") for genre in (podcast_data.get("platinum_keywords") or [])
1006 }
1007
1008 # Add images
1009 image_path = podcast_data.get("product_images", {}).get("500")
1010 podcast.metadata.images = UniqueList(self._create_images(image_path))
1011
1012 return podcast
1013
1014 def _parse_podcast_episode(
1015 self,
1016 episode_data: dict[str, Any] | None,
1017 podcast: Podcast | None,
1018 position: int,
1019 ) -> PodcastEpisode:
1020 """
1021 Parse podcast episode data from API response.
1022
1023 :param episode_data: Raw episode data from the Audible API.
1024 :param podcast: Parent podcast object (optional).
1025 :param position: Position/index of the episode in the podcast.
1026 """
1027 if episode_data is None:
1028 self.logger.error("Received None episode_data in _parse_podcast_episode")
1029 raise MediaNotFoundError("Podcast episode data not found")
1030
1031 asin = episode_data.get("asin", "")
1032 title = episode_data.get("title", "")
1033
1034 # Get duration from runtime_length_min
1035 runtime_minutes = episode_data.get("runtime_length_min", 0)
1036 duration = runtime_minutes * 60 if runtime_minutes else 0
1037
1038 # Create podcast reference - use Podcast object or create ItemMapping
1039 podcast_ref: Podcast | ItemMapping
1040 if podcast is not None:
1041 podcast_ref = podcast
1042 else:
1043 # Try to get parent_asin from relationships for ItemMapping
1044 parent_asin = ""
1045 relationships = episode_data.get("relationships", [])
1046 for rel in relationships:
1047 if rel.get("relationship_type") == "parent":
1048 parent_asin = rel.get("asin", "")
1049 break
1050
1051 if not parent_asin:
1052 self.logger.warning(
1053 "No parent_asin found for podcast episode %s; parent podcast is unknown",
1054 asin,
1055 )
1056
1057 podcast_ref = ItemMapping(
1058 item_id=parent_asin or "",
1059 provider=self.provider_instance,
1060 name="Unknown Podcast",
1061 media_type=MediaType.PODCAST,
1062 )
1063
1064 # Create episode object
1065 episode = PodcastEpisode(
1066 item_id=asin,
1067 provider=self.provider_instance,
1068 name=title,
1069 duration=duration,
1070 position=position,
1071 podcast=podcast_ref,
1072 provider_mappings={
1073 ProviderMapping(
1074 item_id=asin,
1075 provider_domain=self.provider_domain,
1076 provider_instance=self.provider_instance,
1077 )
1078 },
1079 )
1080
1081 # Set metadata
1082 episode.metadata.description = _html_to_txt(
1083 str(
1084 episode_data.get("publisher_summary", "")
1085 or episode_data.get("extended_product_description", "")
1086 )
1087 )
1088
1089 # Add images
1090 image_path = episode_data.get("product_images", {}).get("500")
1091 episode.metadata.images = UniqueList(self._create_images(image_path))
1092
1093 return episode
1094
1095 async def get_authors(self) -> dict[str, str]:
1096 """
1097 Get all unique authors from the library.
1098
1099 Returns dict mapping author ASIN to author name.
1100 """
1101 authors: dict[str, str] = {}
1102 async for item in self._fetch_library_items(
1103 "contributors,product_attrs", AUDIOBOOK_CONTENT_TYPES
1104 ):
1105 for author in item.get("authors") or []:
1106 asin = author.get("asin")
1107 name = author.get("name")
1108 if asin and name:
1109 authors[asin] = name
1110 return authors
1111
1112 async def get_series(self) -> dict[str, str]:
1113 """
1114 Get all unique series from the library.
1115
1116 Returns dict mapping series ASIN to series title.
1117 """
1118 series: dict[str, str] = {}
1119 async for item in self._fetch_library_items(
1120 "series,product_attrs", AUDIOBOOK_CONTENT_TYPES
1121 ):
1122 for s in item.get("series") or []:
1123 asin = s.get("asin")
1124 title = s.get("title")
1125 if asin and title:
1126 series[asin] = title
1127 return series
1128
1129 async def get_narrators(self) -> dict[str, str]:
1130 """
1131 Get all unique narrators from the library.
1132
1133 Returns dict mapping narrator ASIN to narrator name.
1134 """
1135 narrators: dict[str, str] = {}
1136 async for item in self._fetch_library_items(
1137 "contributors,product_attrs", AUDIOBOOK_CONTENT_TYPES
1138 ):
1139 for narrator in item.get("narrators") or []:
1140 asin = narrator.get("asin")
1141 name = narrator.get("name")
1142 if asin and name:
1143 narrators[asin] = name
1144 return narrators
1145
1146 async def get_genres(self) -> set[str]:
1147 """Get all unique genres from the library."""
1148 genres: set[str] = set()
1149 async for item in self._fetch_library_items("product_attrs", AUDIOBOOK_CONTENT_TYPES):
1150 for keyword in item.get("thesaurus_subject_keywords") or []:
1151 genres.add(keyword.replace("_", " ").replace("-", " ").title())
1152 return genres
1153
1154 async def get_publishers(self) -> set[str]:
1155 """Get all unique publishers from the library."""
1156 publishers: set[str] = set()
1157 async for item in self._fetch_library_items("product_attrs", AUDIOBOOK_CONTENT_TYPES):
1158 publisher = item.get("publisher_name")
1159 if publisher:
1160 publishers.add(publisher)
1161 return publishers
1162
1163 async def get_audiobooks_by_author(self, author_asin: str) -> list[Audiobook]:
1164 """Get all audiobooks by a specific author, sorted by release date."""
1165 audiobooks: list[tuple[str, Audiobook]] = []
1166 async for item in self._fetch_library_items(
1167 "contributors,media,product_attrs,product_desc,series", AUDIOBOOK_CONTENT_TYPES
1168 ):
1169 for author in item.get("authors") or []:
1170 if author.get("asin") == author_asin:
1171 release_date = item.get("release_date") or "0000-00-00"
1172 audiobooks.append((release_date, self._parse_audiobook(item)))
1173 break
1174 audiobooks.sort(key=lambda x: x[0], reverse=True)
1175 return [book for _, book in audiobooks]
1176
1177 async def get_audiobooks_by_narrator(self, narrator_asin: str) -> list[Audiobook]:
1178 """Get all audiobooks by a specific narrator, sorted by release date."""
1179 audiobooks: list[tuple[str, Audiobook]] = []
1180 async for item in self._fetch_library_items(
1181 "contributors,media,product_attrs,product_desc,series", AUDIOBOOK_CONTENT_TYPES
1182 ):
1183 for narrator in item.get("narrators") or []:
1184 if narrator.get("asin") == narrator_asin:
1185 release_date = item.get("release_date") or "0000-00-00"
1186 audiobooks.append((release_date, self._parse_audiobook(item)))
1187 break
1188 audiobooks.sort(key=lambda x: x[0], reverse=True)
1189 return [book for _, book in audiobooks]
1190
1191 async def get_audiobooks_by_genre(self, genre: str) -> list[Audiobook]:
1192 """Get all audiobooks matching a genre, sorted by release date."""
1193 audiobooks: list[tuple[str, Audiobook]] = []
1194 genre_key = genre.lower().replace(" ", "_")
1195 genre_key_alt = genre.lower().replace(" ", "-")
1196 async for item in self._fetch_library_items(
1197 "contributors,media,product_attrs,product_desc,series", AUDIOBOOK_CONTENT_TYPES
1198 ):
1199 keywords = item.get("thesaurus_subject_keywords") or []
1200 if genre_key in keywords or genre_key_alt in keywords:
1201 release_date = item.get("release_date") or "0000-00-00"
1202 audiobooks.append((release_date, self._parse_audiobook(item)))
1203 audiobooks.sort(key=lambda x: x[0], reverse=True)
1204 return [book for _, book in audiobooks]
1205
1206 async def get_audiobooks_by_publisher(self, publisher: str) -> list[Audiobook]:
1207 """Get all audiobooks from a specific publisher, sorted by release date."""
1208 audiobooks: list[tuple[str, Audiobook]] = []
1209 async for item in self._fetch_library_items(
1210 "contributors,media,product_attrs,product_desc,series", AUDIOBOOK_CONTENT_TYPES
1211 ):
1212 if item.get("publisher_name") == publisher:
1213 release_date = item.get("release_date") or "0000-00-00"
1214 audiobooks.append((release_date, self._parse_audiobook(item)))
1215 audiobooks.sort(key=lambda x: x[0], reverse=True)
1216 return [book for _, book in audiobooks]
1217
1218 async def get_audiobooks_by_series(self, series_asin: str) -> list[Audiobook]:
1219 """Get all audiobooks in a specific series, ordered by sequence."""
1220 audiobooks: list[tuple[float, Audiobook]] = []
1221 async for item in self._fetch_library_items(
1222 "contributors,media,product_attrs,product_desc,series", AUDIOBOOK_CONTENT_TYPES
1223 ):
1224 for s in item.get("series") or []:
1225 if s.get("asin") == series_asin:
1226 sequence = s.get("sequence")
1227 try:
1228 seq_num = float(sequence) if sequence else 999
1229 except ValueError, TypeError:
1230 seq_num = 999
1231 audiobooks.append((seq_num, self._parse_audiobook(item)))
1232 break
1233 audiobooks.sort(key=lambda x: x[0])
1234 return [book for _, book in audiobooks]
1235
1236 async def deregister(self) -> None:
1237 """Deregister this provider from Audible."""
1238 await asyncio.to_thread(self.client.auth.deregister_device)
1239
1240
1241def _html_to_txt(html_text: str) -> str:
1242 txt = html.unescape(html_text)
1243 tags = re.findall("<[^>]+>", txt)
1244 for tag in tags:
1245 txt = txt.replace(tag, "")
1246 return txt
1247
1248
1249async def audible_get_auth_info(locale: str) -> tuple[str, str, str]:
1250 """
1251 Generate the login URL and auth info for Audible OAuth flow.
1252
1253 :param locale: The locale string (e.g., 'us', 'uk', 'de').
1254 :return: Tuple of (code_verifier, oauth_url, serial).
1255 """
1256 locale_obj = audible.localization.Locale(locale)
1257 code_verifier = await asyncio.to_thread(audible.login.create_code_verifier)
1258 oauth_url, serial = await asyncio.to_thread(
1259 audible.login.build_oauth_url,
1260 country_code=locale_obj.country_code,
1261 domain=locale_obj.domain,
1262 market_place_id=locale_obj.market_place_id,
1263 code_verifier=code_verifier,
1264 with_username=False,
1265 )
1266
1267 return code_verifier.decode(), oauth_url, serial
1268
1269
1270async def audible_custom_login(
1271 code_verifier: str, response_url: str, serial: str, locale: str
1272) -> audible.Authenticator:
1273 """
1274 Complete the authentication using the code_verifier, response_url, and serial.
1275
1276 :param code_verifier: The code verifier string used in OAuth flow.
1277 :param response_url: The response URL containing the authorization code.
1278 :param serial: The device serial number.
1279 :param locale: The locale string.
1280 :return: Audible Authenticator object.
1281 :raises LoginFailed: If authorization code is not found in the URL.
1282 """
1283 logger = logging.getLogger("audible_helper")
1284 auth = audible.Authenticator()
1285 auth.locale = audible.localization.Locale(locale)
1286
1287 response_url_parsed = urlparse(response_url)
1288 parsed_qs = parse_qs(response_url_parsed.query)
1289
1290 # Try multiple parameter names for authorization code
1291 # Audible may use different parameter names depending on the flow
1292 authorization_code = None
1293 for param_name in ["openid.oa2.authorization_code", "authorization_code", "code"]:
1294 if codes := parsed_qs.get(param_name):
1295 authorization_code = codes[0]
1296 logger.debug("Found authorization code in parameter: %s", param_name)
1297 break
1298
1299 if not authorization_code:
1300 available_params = list(parsed_qs.keys())
1301 raise LoginFailed(
1302 f"Authorization code not found in URL. "
1303 f"Expected 'openid.oa2.authorization_code' but found parameters: {available_params}"
1304 )
1305
1306 registration_data = await asyncio.to_thread(
1307 audible.register.register,
1308 authorization_code=authorization_code,
1309 code_verifier=code_verifier.encode(),
1310 domain=auth.locale.domain,
1311 serial=serial,
1312 )
1313 auth._update_attrs(**registration_data)
1314
1315 # Log what auth methods are available after registration
1316 if auth.adp_token and auth.device_private_key:
1317 logger.info("Registration successful with signing auth (stable)")
1318 else:
1319 logger.warning("Registration successful but signing auth not available")
1320
1321 return auth
1322
1323
1324async def check_file_exists(path: str | PathLike[str]) -> bool:
1325 """Async file exists check."""
1326 return await asyncio.to_thread(os.path.exists, path)
1327
1328
1329async def remove_file(path: str | PathLike[str]) -> None:
1330 """Async file delete."""
1331 await asyncio.to_thread(os.remove, path)
1332