/
/
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 contextlib import suppress
17from typing import TYPE_CHECKING
18from uuid import uuid4
19
20from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
21from music_assistant_models.enums import ConfigEntryType
22from music_assistant_models.errors import LoginFailed
23
24from music_assistant.models.setup_flow import SetupFlowError, StepExpiredError
25
26from . import CONF_AUTH_FILE, CONF_LOCALE
27from .audible_helper import audible_custom_login, audible_get_auth_info, remove_file
28
29if TYPE_CHECKING:
30 from music_assistant_models.config_entries import ConfigValueType
31
32 from music_assistant.models.setup_flow import SetupSession
33
34# form field key of the pasted post-login ("page not found") redirect URL
35CONF_POST_LOGIN_URL = "post_login_url"
36
37_LOCALES = ("us", "ca", "uk", "au", "fr", "de", "jp", "it", "in", "es", "br")
38
39
40async def run_setup(session: SetupSession) -> None:
41 """
42 Run the Audible sign-in flow: marketplace, then authorize + paste redirect URL.
43
44 :param session: The setup session driving the flow.
45 """
46 locale_default = str(session.context.setup_data.get(CONF_LOCALE) or "us")
47 values = await session.form(
48 [
49 ConfigEntry(key="intro", type=ConfigEntryType.LABEL),
50 ConfigEntry(
51 key=CONF_LOCALE,
52 type=ConfigEntryType.STRING,
53 required=True,
54 default_value="us",
55 value=locale_default,
56 options=[ConfigValueOption(loc) for loc in _LOCALES],
57 ),
58 ],
59 step_id="user",
60 )
61 locale = str(values[CONF_LOCALE])
62
63 errors: dict[str, str] | None = None
64 while True:
65 # a fresh authorize URL (+ PKCE verifier + device serial) per attempt, since the
66 # pasted redirect carries a single-use authorization code
67 code_verifier, login_url, serial = await audible_get_auth_info(locale)
68 with suppress(StepExpiredError, OSError):
69 await session.external_until(
70 asyncio.Future(),
71 login_url,
72 step_id="authorize",
73 expires_in=30,
74 )
75 values = await session.form(
76 [
77 ConfigEntry(
78 key=CONF_POST_LOGIN_URL,
79 type=ConfigEntryType.STRING,
80 required=True,
81 ),
82 ],
83 step_id="authenticate",
84 last_step=True,
85 errors=errors,
86 )
87 post_login_url = str(values[CONF_POST_LOGIN_URL])
88 try:
89 auth = await audible_custom_login(code_verifier, post_login_url, serial, locale)
90 if not (auth.adp_token and auth.device_private_key):
91 raise LoginFailed(
92 "Registration succeeded but signing keys were not obtained. Please try again."
93 )
94 except LoginFailed as err:
95 errors = {"base": err.translation_key or str(err)}
96 continue
97 except Exception as err:
98 errors = {"base": str(err)}
99 continue
100 auth_file_path = os.path.join(session.mass.storage_path, f"audible_auth_{uuid4().hex}.json")
101 await asyncio.to_thread(auth.to_file, auth_file_path)
102 collected: dict[str, ConfigValueType] = {
103 CONF_AUTH_FILE: auth_file_path,
104 CONF_LOCALE: locale,
105 }
106 try:
107 await session.finish(collected)
108 return
109 except SetupFlowError as err:
110 # the just-written token file is unusable if the load failed; drop it
111 await remove_file(auth_file_path)
112 errors = {"base": err.translation_key or str(err)}
113