music-assistant-server

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