/
/
/
1"""
2Setup flow for the Audible provider.
3
4Audible sign-in happens on Amazon's own web page (which handles the password, any
5CAPTCHA and OTP), so Music Assistant cannot receive an OAuth callback. The flow instead
6sends the user to Amazon's authorize URL and asks them to paste the resulting
7"page not found" redirect URL, from which the device is registered. The registration
8tokens are written to a file under the MA storage path; only that file path (and the
9marketplace locale) are persisted as setup data.
10"""
11
12from __future__ import annotations
13
14import asyncio
15import os
16from typing import TYPE_CHECKING
17from uuid import uuid4
18
19from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
20from music_assistant_models.enums import ConfigEntryType
21from music_assistant_models.errors import LoginFailed
22
23from music_assistant.models.setup_flow import SetupFlowError
24
25from . import CONF_AUTH_FILE, CONF_LOCALE
26from .audible_helper import audible_custom_login, audible_get_auth_info, remove_file
27
28if TYPE_CHECKING:
29 from music_assistant_models.config_entries import ConfigValueType
30
31 from music_assistant.models.setup_flow import SetupSession
32
33# form field key of the pasted post-login ("page not found") redirect URL
34CONF_POST_LOGIN_URL = "post_login_url"
35
36_LOCALES = ("us", "ca", "uk", "au", "fr", "de", "jp", "it", "in", "es", "br")
37
38
39async def run_setup(session: SetupSession) -> None:
40 """
41 Run the Audible sign-in flow: marketplace, then authorize + paste redirect URL.
42
43 :param session: The setup session driving the flow.
44 """
45 locale_default = str(session.context.setup_data.get(CONF_LOCALE) or "us")
46 values = await session.form(
47 [
48 ConfigEntry(key="intro", type=ConfigEntryType.LABEL),
49 ConfigEntry(
50 key=CONF_LOCALE,
51 type=ConfigEntryType.STRING,
52 required=True,
53 default_value="us",
54 value=locale_default,
55 options=[ConfigValueOption(loc) for loc in _LOCALES],
56 ),
57 ],
58 step_id="user",
59 )
60 locale = str(values[CONF_LOCALE])
61
62 errors: dict[str, str] | None = None
63 while True:
64 # a fresh authorize URL (+ PKCE verifier + device serial) per attempt, since the
65 # pasted redirect carries a single-use authorization code
66 code_verifier, login_url, serial = await audible_get_auth_info(locale)
67 values = await session.form(
68 [
69 ConfigEntry(
70 key="authorize",
71 type=ConfigEntryType.LABEL,
72 help_link=login_url,
73 ),
74 ConfigEntry(key=CONF_POST_LOGIN_URL, type=ConfigEntryType.STRING, required=True),
75 ],
76 step_id="authenticate",
77 errors=errors,
78 last_step=True,
79 )
80 post_login_url = str(values[CONF_POST_LOGIN_URL])
81 try:
82 auth = await audible_custom_login(code_verifier, post_login_url, serial, locale)
83 if not (auth.adp_token and auth.device_private_key):
84 raise LoginFailed(
85 "Registration succeeded but signing keys were not obtained. Please try again."
86 )
87 except LoginFailed as err:
88 errors = {"base": err.translation_key or str(err)}
89 continue
90 except Exception as err:
91 errors = {"base": str(err)}
92 continue
93 auth_file_path = os.path.join(session.mass.storage_path, f"audible_auth_{uuid4().hex}.json")
94 await asyncio.to_thread(auth.to_file, auth_file_path)
95 collected: dict[str, ConfigValueType] = {
96 CONF_AUTH_FILE: auth_file_path,
97 CONF_LOCALE: locale,
98 }
99 try:
100 await session.finish(collected)
101 return
102 except SetupFlowError as err:
103 # the just-written token file is unusable if the load failed; drop it
104 await remove_file(auth_file_path)
105 errors = {"base": err.translation_key or str(err)}
106