/
/
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(
49 key=CONF_LOCALE,
50 type=ConfigEntryType.STRING,
51 required=True,
52 default_value="us",
53 value=locale_default,
54 options=[ConfigValueOption(loc) for loc in _LOCALES],
55 ),
56 ],
57 step_id="user",
58 )
59 locale = str(values[CONF_LOCALE])
60
61 errors: dict[str, str] | None = None
62 while True:
63 # a fresh authorize URL (+ PKCE verifier + device serial) per attempt, since the
64 # pasted redirect carries a single-use authorization code
65 code_verifier, login_url, serial = await audible_get_auth_info(locale)
66 values = await session.form(
67 [
68 ConfigEntry(
69 key="auth_link",
70 type=ConfigEntryType.LABEL,
71 translation_params=[login_url],
72 ),
73 ConfigEntry(
74 key=CONF_POST_LOGIN_URL,
75 type=ConfigEntryType.STRING,
76 required=True,
77 ),
78 ],
79 step_id="authenticate",
80 last_step=True,
81 errors=errors,
82 )
83 post_login_url = str(values[CONF_POST_LOGIN_URL])
84 try:
85 auth = await audible_custom_login(code_verifier, post_login_url, serial, locale)
86 if not (auth.adp_token and auth.device_private_key):
87 raise LoginFailed(
88 "Registration succeeded but signing keys were not obtained. Please try again."
89 )
90 except LoginFailed as err:
91 errors = {"base": err.translation_key or str(err)}
92 continue
93 except Exception as err:
94 errors = {"base": str(err)}
95 continue
96 auth_file_path = os.path.join(session.mass.storage_path, f"audible_auth_{uuid4().hex}.json")
97 await asyncio.to_thread(auth.to_file, auth_file_path)
98 collected: dict[str, ConfigValueType] = {
99 CONF_AUTH_FILE: auth_file_path,
100 CONF_LOCALE: locale,
101 }
102 try:
103 await session.finish(collected)
104 return
105 except SetupFlowError as err:
106 # the just-written token file is unusable if the load failed; drop it
107 await remove_file(auth_file_path)
108 errors = {"base": err.translation_key or str(err)}
109