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