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