/
/
/
1"""
2Setup flow for the Tidal music provider.
3
4Tidal is linked with the OAuth device flow. It has no browser callback (Tidal's
5authorize page cannot redirect back to Music Assistant), so completion is detected by
6polling the token endpoint. The single setup step shows an "Open" button for the
7``link.tidal.com`` verification URL (code pre-filled) plus a waiting spinner, and
8auto-completes the moment the poll reports approval. A code that expires before
9approval is minted afresh and the step re-shown.
10"""
11
12from __future__ import annotations
13
14from typing import TYPE_CHECKING
15
16from music_assistant_models.errors import LoginFailed
17
18from music_assistant.models.setup_flow import AbortFlow, StepExpiredError
19
20from .auth_manager import TidalAuthManager
21from .constants import (
22 CONF_AUTH_TOKEN,
23 CONF_EXPIRY_TIME,
24 CONF_REFRESH_TOKEN,
25 CONF_USER_ID,
26)
27
28if TYPE_CHECKING:
29 from music_assistant_models.config_entries import ConfigValueType
30
31 from music_assistant.models.setup_flow import SetupSession
32
33
34async def run_setup(session: SetupSession) -> None:
35 """
36 Run the Tidal device-flow login: open the link, poll until approved, persist tokens.
37
38 :param session: The setup session driving the flow.
39 """
40 http_session = session.mass.http_session
41 while True:
42 device = await TidalAuthManager.start_device_login(http_session)
43 # a single "Open link.tidal.com" step (code pre-filled) that auto-completes when
44 # the poll reports approval. The step's countdown owns expiry, raising
45 # StepExpiredError so we mint and show a fresh code.
46 try:
47 auth_data = await session.external_until(
48 TidalAuthManager.poll_device_login(http_session, device),
49 url=_verification_url(device),
50 step_id="device_login",
51 expires_in=float(device["expiresIn"]),
52 # shown on the step so the code can also be typed at link.tidal.com
53 # from a phone, rather than only carried by the Open button's url
54 translation_params=[str(device["userCode"])],
55 )
56 except StepExpiredError:
57 continue
58 except LoginFailed as err:
59 raise AbortFlow("login_failed") from err
60
61 collected: dict[str, ConfigValueType] = {
62 CONF_AUTH_TOKEN: auth_data["access_token"],
63 CONF_REFRESH_TOKEN: auth_data["refresh_token"],
64 CONF_EXPIRY_TIME: auth_data["expires_at"],
65 CONF_USER_ID: str(auth_data["userId"]),
66 }
67 await session.finish(collected)
68 return
69
70
71def _verification_url(device: dict[str, str]) -> str:
72 """Return the full (scheme-prefixed) verification URL with the code pre-filled."""
73 url = str(device.get("verificationUriComplete") or device["verificationUri"])
74 return url if url.startswith("http") else f"https://{url}"
75