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