music-assistant-server

1"""
2Setup flow for the Spotify Connect plugin.
3
4The flow starts with an explicit backend choice: Spotify Soloist (Spotify's
5official headless client, guarded by a ToS warning/consent step and a personal
6API key) or the community go-librespot daemon. Both branches end with the shared
7target-player / device-name step. Reconfigure preselects the stored backend and
8re-runs the branch steps; switching away from soloist clears the soloist
9secrets, but only once the new setup finishes successfully.
10"""
11
12from __future__ import annotations
13
14from typing import TYPE_CHECKING, Any
15
16from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
17from music_assistant_models.enums import ConfigEntryType
18
19from music_assistant.helpers.config_entries import create_player_selector
20from music_assistant.models.setup_flow import SetupFlowError
21
22from . import (
23    BACKEND_GO_LIBRESPOT,
24    BACKEND_SOLOIST,
25    CONF_API_KEY,
26    CONF_BACKEND,
27    CONF_MASS_PLAYER_ID,
28    CONF_PUBLISH_NAME,
29    CONF_SOLOIST_CONSENT,
30    DEFAULT_PUBLISH_NAME,
31    PLAYER_ID_AUTO,
32)
33from .soloist import UnsupportedPlatformError, verify_platform_supported
34
35if TYPE_CHECKING:
36    from music_assistant.models.setup_flow import SetupSession
37
38# Minimum plausible length of a pasted Soloist API key: anything shorter is a
39# partial paste. No further format rules are applied locally — Spotify rejects
40# an invalid key when soloist authenticates.
41MIN_API_KEY_LENGTH = 16
42
43
44async def run_setup(session: SetupSession) -> None:
45    """
46    Configure the Spotify Connect backend, target player and device name.
47
48    :param session: The setup session driving the flow.
49    """
50    setup_data = dict(session.context.setup_data)
51    stored_backend = str(
52        setup_data.get(CONF_BACKEND) or session.context.values.get(CONF_BACKEND) or ""
53    )
54    selected = stored_backend or BACKEND_SOLOIST
55    choice_errors: dict[str, str] | None = None
56    while True:
57        selected = await _choose_backend(session, selected, choice_errors)
58        choice_errors = None
59        # rebuild from the stored setup data each round so a refused soloist
60        # attempt does not leak partial values into a later selection
61        collected = dict(setup_data)
62        collected[CONF_BACKEND] = selected
63        if selected == BACKEND_SOLOIST:
64            if not await _run_soloist_steps(session, collected):
65                # consent refused: back to the backend choice with a clear error
66                choice_errors = {"base": "soloist_consent_required"}
67                continue
68        elif stored_backend == BACKEND_SOLOIST:
69            # switching away from soloist: overwrite the soloist secrets; they
70            # only reach the stored setup_data when finish() succeeds, so an
71            # aborted or failed switch keeps them intact
72            collected[CONF_API_KEY] = ""
73            collected[CONF_SOLOIST_CONSENT] = False
74        await _finish_with_player_and_name(session, collected)
75        return
76
77
78async def _choose_backend(
79    session: SetupSession, preselect: str, errors: dict[str, str] | None
80) -> str:
81    """
82    Show the backend choice step until a usable backend is selected.
83
84    :param session: The setup session driving the flow.
85    :param preselect: Backend to preselect (the stored or previously chosen one).
86    :param errors: Optional errors to display on the first render.
87    """
88    while True:
89        values = await session.form(
90            [
91                ConfigEntry(
92                    key=CONF_BACKEND,
93                    type=ConfigEntryType.STRING,
94                    required=True,
95                    default_value=BACKEND_SOLOIST,
96                    value=preselect,
97                    options=[
98                        ConfigValueOption(BACKEND_SOLOIST),
99                        ConfigValueOption(BACKEND_GO_LIBRESPOT),
100                    ],
101                    expanded_options=True,
102                ),
103            ],
104            step_id="backend",
105            errors=errors,
106        )
107        selected = str(values[CONF_BACKEND])
108        if selected == BACKEND_SOLOIST:
109            try:
110                verify_platform_supported()
111            except UnsupportedPlatformError:
112                errors = {"base": "soloist_unsupported_platform"}
113                preselect = BACKEND_GO_LIBRESPOT
114                continue
115        return selected
116
117
118async def _run_soloist_steps(session: SetupSession, collected: dict[str, Any]) -> bool:
119    """
120    Run the soloist branch: the ToS warning/consent step, then the API key step.
121
122    :param session: The setup session driving the flow.
123    :param collected: The values collected so far; updated in place.
124    :return: True when the branch completed, False when consent was refused.
125    """
126    if not await _ask_consent(session, bool(collected.get(CONF_SOLOIST_CONSENT))):
127        return False
128    collected[CONF_SOLOIST_CONSENT] = True
129    await _ask_api_key(session, collected)
130    return True
131
132
133async def _ask_consent(session: SetupSession, prefill: bool) -> bool:
134    """
135    Show the soloist warning/consent step and return whether consent was given.
136
137    :param session: The setup session driving the flow.
138    :param prefill: Whether consent was already given on an earlier run.
139    """
140    values = await session.form(
141        [
142            ConfigEntry(
143                key=CONF_SOLOIST_CONSENT,
144                type=ConfigEntryType.BOOLEAN,
145                required=False,
146                default_value=False,
147                value=prefill,
148            ),
149        ],
150        step_id="soloist_terms",
151    )
152    return bool(values.get(CONF_SOLOIST_CONSENT))
153
154
155async def _ask_api_key(session: SetupSession, collected: dict[str, Any]) -> None:
156    """
157    Collect the Soloist API key.
158
159    An already stored key (reconfigure) is kept when the field is left empty;
160    it is never shown back to the user.
161
162    :param session: The setup session driving the flow.
163    :param collected: The values collected so far; updated in place.
164    """
165    has_stored_key = bool(collected.get(CONF_API_KEY))
166    errors: dict[str, str] | None = None
167    while True:
168        entries = [
169            ConfigEntry(
170                key=CONF_API_KEY,
171                type=ConfigEntryType.SECURE_STRING,
172                required=not has_stored_key,
173            ),
174        ]
175        if has_stored_key:
176            entries.insert(0, ConfigEntry(key="soloist_api_key_hint", type=ConfigEntryType.LABEL))
177        values = await session.form(entries, step_id="soloist_api_key", errors=errors)
178        api_key = str(values.get(CONF_API_KEY) or "").strip()
179        if api_key or not has_stored_key:
180            if len(api_key) < MIN_API_KEY_LENGTH:
181                errors = {CONF_API_KEY: "soloist_api_key_invalid"}
182                continue
183            collected[CONF_API_KEY] = api_key
184        return
185
186
187async def _finish_with_player_and_name(session: SetupSession, collected: dict[str, Any]) -> None:
188    """
189    Run the shared target-player / device-name step and finish the flow.
190
191    :param session: The setup session driving the flow.
192    :param collected: The values collected by the earlier steps.
193    """
194    errors: dict[str, str] | None = None
195    while True:
196        prefill: dict[str, Any] = {**session.context.values, **collected}
197        publish_name = str(prefill.get(CONF_PUBLISH_NAME) or DEFAULT_PUBLISH_NAME)
198        values = await session.form(
199            [
200                create_player_selector(
201                    session.mass,
202                    CONF_MASS_PLAYER_ID,
203                    prefill.get(CONF_MASS_PLAYER_ID),
204                    PLAYER_ID_AUTO,
205                ),
206                ConfigEntry(
207                    key=CONF_PUBLISH_NAME,
208                    type=ConfigEntryType.STRING,
209                    required=True,
210                    default_value=publish_name,
211                    value=publish_name,
212                ),
213            ],
214            step_id="user",
215            errors=errors,
216            last_step=True,
217        )
218        collected.update(values)
219        try:
220            await session.finish(collected)
221            return
222        except SetupFlowError as err:
223            errors = {"base": err.translation_key or str(err)}
224