/
/
1"""
2Setup flow for the Tidal music provider.
3
4Tidal uses OAuth PKCE, but its authorize page redirects to Tidal's own
5"Page Not Found" page instead of a callback Music Assistant can receive. The flow
6therefore shows the authorize link and a field for the user to paste that redirected
7URL, then exchanges it for tokens. The PKCE ``code_verifier`` lives only as a local
8for the duration of the flow (it used to be smuggled through a hidden config value).
9"""
10
11from __future__ import annotations
12
13import json
14from typing import TYPE_CHECKING
15
16from music_assistant_models.config_entries import ConfigEntry
17from music_assistant_models.enums import ConfigEntryType
18from music_assistant_models.errors import LoginFailed
19
20from music_assistant.models.setup_flow import SetupFlowError
21
22from .auth_manager import TidalAuthManager
23from .constants import (
24 CONF_AUTH_TOKEN,
25 CONF_EXPIRY_TIME,
26 CONF_OOPS_URL,
27 CONF_QUALITY,
28 CONF_REFRESH_TOKEN,
29 CONF_USER_ID,
30)
31
32if TYPE_CHECKING:
33 from music_assistant_models.config_entries import ConfigValueType
34
35 from music_assistant.models.setup_flow import SetupSession
36
37DEFAULT_QUALITY = "HI_RES_LOSSLESS"
38
39
40async def run_setup(session: SetupSession) -> None:
41 """
42 Run the Tidal PKCE login flow: show the authorize link, exchange the pasted URL.
43
44 :param session: The setup session driving the flow.
45 """
46 # quality stays a genuine option; it is only carried through the exchange, so a
47 # (reconfigure) prefill or the default is enough here - the user edits it later
48 quality = str(session.context.values.get(CONF_QUALITY) or DEFAULT_QUALITY)
49 # the PKCE verifier now lives only in this local auth_params blob
50 authorize_url, auth_params = TidalAuthManager.build_pkce_login(quality)
51
52 errors: dict[str, str] | None = None
53 while True:
54 values = await session.form(
55 [
56 ConfigEntry(
57 key="auth_instructions",
58 type=ConfigEntryType.LABEL,
59 help_link=authorize_url,
60 ),
61 ConfigEntry(key=CONF_OOPS_URL, type=ConfigEntryType.STRING, required=True),
62 ],
63 step_id="user",
64 errors=errors,
65 last_step=True,
66 )
67 oops_url = str(values[CONF_OOPS_URL])
68 try:
69 auth_data = await TidalAuthManager.process_pkce_login(
70 session.mass.http_session, json.dumps(auth_params), oops_url
71 )
72 except LoginFailed as err:
73 errors = {"base": err.translation_key or str(err)}
74 continue
75 collected: dict[str, ConfigValueType] = {
76 CONF_AUTH_TOKEN: auth_data["access_token"],
77 CONF_REFRESH_TOKEN: auth_data["refresh_token"],
78 CONF_EXPIRY_TIME: auth_data["expires_at"],
79 CONF_USER_ID: str(auth_data["userId"]),
80 }
81 try:
82 await session.finish(collected)
83 return
84 except SetupFlowError as err:
85 errors = {"base": err.translation_key or str(err)}
86