/
/
/
1"""Tidal music provider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, cast
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
9from music_assistant_models.enums import ConfigEntryType
10
11from .auth_manager import ManualAuthenticationHelper, TidalAuthManager
12from .constants import (
13 CONF_ACTION_CLEAR_AUTH,
14 CONF_ACTION_COMPLETE_PKCE_LOGIN,
15 CONF_ACTION_START_PKCE_LOGIN,
16 CONF_AUTH_TOKEN,
17 CONF_EXPIRY_TIME,
18 CONF_OOPS_URL,
19 CONF_QUALITY,
20 CONF_REFRESH_TOKEN,
21 CONF_TEMP_SESSION,
22 CONF_USER_ID,
23 LABEL_COMPLETE_PKCE_LOGIN,
24 LABEL_OOPS_URL,
25 LABEL_START_PKCE_LOGIN,
26)
27from .provider import TidalProvider
28
29if TYPE_CHECKING:
30 from music_assistant_models.config_entries import ProviderConfig
31 from music_assistant_models.provider import ProviderManifest
32
33 from music_assistant.mass import MusicAssistant
34 from music_assistant.models import ProviderInstanceType
35
36
37async def setup(
38 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
39) -> ProviderInstanceType:
40 """Initialize provider(instance) with given configuration."""
41 return TidalProvider(mass, manifest, config)
42
43
44async def get_config_entries(
45 mass: MusicAssistant,
46 instance_id: str | None = None, # noqa: ARG001
47 action: str | None = None,
48 values: dict[str, ConfigValueType] | None = None,
49) -> tuple[ConfigEntry, ...]:
50 """
51 Return configuration entries required to set up the Tidal provider.
52
53 Parameters:
54 mass (MusicAssistant): The MusicAssistant instance.
55 instance_id (str | None): Optional instance identifier for the provider.
56 action (str | None): Optional action to perform (e.g., start or complete PKCE login).
57 values (dict[str, ConfigValueType] | None): Dictionary of current configuration values.
58
59 Returns:
60 tuple[ConfigEntry, ...]: Tuple of ConfigEntry objects representing the configuration steps.
61
62 The function handles authentication actions and returns the appropriate configuration entries
63 based on the current state and provided values.
64 """
65 assert values is not None
66
67 if action == CONF_ACTION_START_PKCE_LOGIN:
68 async with ManualAuthenticationHelper(
69 mass, cast("str", values["session_id"])
70 ) as auth_helper:
71 quality = str(values.get(CONF_QUALITY))
72 base64_session = await TidalAuthManager.generate_auth_url(auth_helper, quality)
73 values[CONF_TEMP_SESSION] = base64_session
74 # Tidal is using the ManualAuthenticationHelper just to send the user to an URL
75 # there is no actual oauth callback happening, instead the user is redirected
76 # to a non-existent page and needs to copy the URL from the browser and paste it
77 # we simply wait here to allow the user to start the auth
78 await asyncio.sleep(15)
79
80 if action == CONF_ACTION_COMPLETE_PKCE_LOGIN:
81 quality = str(values.get(CONF_QUALITY))
82 pkce_url = str(values.get(CONF_OOPS_URL))
83 base64_session = str(values.get(CONF_TEMP_SESSION))
84 auth_data = await TidalAuthManager.process_pkce_login(
85 mass.http_session, base64_session, pkce_url
86 )
87 values[CONF_AUTH_TOKEN] = auth_data["access_token"]
88 values[CONF_REFRESH_TOKEN] = auth_data["refresh_token"]
89 values[CONF_EXPIRY_TIME] = auth_data["expires_at"]
90 values[CONF_USER_ID] = auth_data["userId"]
91 values[CONF_TEMP_SESSION] = ""
92
93 if action == CONF_ACTION_CLEAR_AUTH:
94 values[CONF_AUTH_TOKEN] = None
95 values[CONF_REFRESH_TOKEN] = None
96 values[CONF_EXPIRY_TIME] = None
97 values[CONF_USER_ID] = None
98
99 if values.get(CONF_AUTH_TOKEN):
100 auth_entries: tuple[ConfigEntry, ...] = (
101 ConfigEntry(
102 key="label_ok",
103 type=ConfigEntryType.LABEL,
104 label="You are authenticated with Tidal",
105 ),
106 ConfigEntry(
107 key=CONF_ACTION_CLEAR_AUTH,
108 type=ConfigEntryType.ACTION,
109 label="Reset authentication",
110 description="Reset the authentication for Tidal",
111 action=CONF_ACTION_CLEAR_AUTH,
112 value=None,
113 ),
114 ConfigEntry(
115 key=CONF_QUALITY,
116 type=ConfigEntryType.STRING,
117 label="Quality setting for Tidal:",
118 description="High = 16bit 44.1kHz\n\nMax = Up to 24bit 192kHz",
119 options=[
120 ConfigValueOption("High", "LOSSLESS"),
121 ConfigValueOption("Max", "HI_RES_LOSSLESS"),
122 ],
123 default_value="HI_RES_LOSSLESS",
124 ),
125 )
126 else:
127 auth_entries = (
128 ConfigEntry(
129 key=CONF_QUALITY,
130 type=ConfigEntryType.STRING,
131 label="Quality setting for Tidal:",
132 required=True,
133 description="High = 16bit 44.1kHz\n\nMax = Up to 24bit 192kHz",
134 options=[
135 ConfigValueOption("High", "LOSSLESS"),
136 ConfigValueOption("Max", "HI_RES_LOSSLESS"),
137 ],
138 default_value="HI_RES_LOSSLESS",
139 ),
140 ConfigEntry(
141 key=LABEL_START_PKCE_LOGIN,
142 type=ConfigEntryType.LABEL,
143 label="The button below will redirect you to Tidal.com to authenticate.\n\n"
144 " After authenticating, you will be redirected to a page that prominently displays"
145 " 'Page Not Found' at the top. That is normal, you need to copy that URL from the "
146 "address bar and come back here",
147 hidden=action == CONF_ACTION_START_PKCE_LOGIN,
148 ),
149 ConfigEntry(
150 key=CONF_ACTION_START_PKCE_LOGIN,
151 type=ConfigEntryType.ACTION,
152 label="Starts the auth process via PKCE on Tidal.com",
153 description="This button will redirect you to Tidal.com to authenticate."
154 " After authenticating, you will be redirected to a page that prominently displays"
155 " 'Page Not Found' at the top.",
156 action=CONF_ACTION_START_PKCE_LOGIN,
157 depends_on=CONF_QUALITY,
158 action_label="Starts the auth process via PKCE on Tidal.com",
159 value=cast("str", values.get(CONF_TEMP_SESSION)) if values else None,
160 hidden=action == CONF_ACTION_START_PKCE_LOGIN,
161 ),
162 ConfigEntry(
163 key=CONF_TEMP_SESSION,
164 type=ConfigEntryType.STRING,
165 label="Temporary session for Tidal",
166 hidden=True,
167 required=False,
168 value=cast("str", values.get(CONF_TEMP_SESSION)) if values else None,
169 ),
170 ConfigEntry(
171 key=LABEL_OOPS_URL,
172 type=ConfigEntryType.LABEL,
173 label="Copy the URL from the 'Page Not Found' page that you were previously"
174 " redirected to and paste it in the field below",
175 hidden=action != CONF_ACTION_START_PKCE_LOGIN,
176 ),
177 ConfigEntry(
178 key=CONF_OOPS_URL,
179 type=ConfigEntryType.STRING,
180 label="Oops URL from Tidal redirect",
181 description="This field should be filled manually by you after authenticating on"
182 " Tidal.com and being redirected to a page that prominently displays"
183 " 'Page Not Found' at the top.",
184 depends_on=CONF_ACTION_START_PKCE_LOGIN,
185 value=cast("str", values.get(CONF_OOPS_URL)) if values else None,
186 hidden=action != CONF_ACTION_START_PKCE_LOGIN,
187 ),
188 ConfigEntry(
189 key=LABEL_COMPLETE_PKCE_LOGIN,
190 type=ConfigEntryType.LABEL,
191 label="After pasting the URL in the field above, click the button below to complete"
192 " the process.",
193 hidden=action != CONF_ACTION_START_PKCE_LOGIN,
194 ),
195 ConfigEntry(
196 key=CONF_ACTION_COMPLETE_PKCE_LOGIN,
197 type=ConfigEntryType.ACTION,
198 label="Complete the auth process via PKCE on Tidal.com",
199 description="Click this after adding the 'Page Not Found' URL above, this will"
200 " complete the authentication process.",
201 action=CONF_ACTION_COMPLETE_PKCE_LOGIN,
202 depends_on=CONF_OOPS_URL,
203 action_label="Complete the auth process via PKCE on Tidal.com",
204 value=None,
205 hidden=action != CONF_ACTION_START_PKCE_LOGIN,
206 ),
207 )
208
209 # return the auth_data config entry
210 return (
211 *auth_entries,
212 ConfigEntry(
213 key=CONF_AUTH_TOKEN,
214 type=ConfigEntryType.SECURE_STRING,
215 label="Authentication token for Tidal",
216 description="You need to link Music Assistant to your Tidal account.",
217 hidden=True,
218 value=cast("str", values.get(CONF_AUTH_TOKEN)) if values else None,
219 ),
220 ConfigEntry(
221 key=CONF_REFRESH_TOKEN,
222 type=ConfigEntryType.SECURE_STRING,
223 label="Refresh token for Tidal",
224 description="You need to link Music Assistant to your Tidal account.",
225 hidden=True,
226 value=cast("str", values.get(CONF_REFRESH_TOKEN)) if values else None,
227 ),
228 ConfigEntry(
229 key=CONF_EXPIRY_TIME,
230 type=ConfigEntryType.STRING,
231 label="Expiry time of auth token for Tidal",
232 hidden=True,
233 value=cast("str", values.get(CONF_EXPIRY_TIME)) if values else None,
234 ),
235 ConfigEntry(
236 key=CONF_USER_ID,
237 type=ConfigEntryType.STRING,
238 label="Your Tidal User ID",
239 description="This is your unique Tidal user ID.",
240 hidden=True,
241 value=cast("str", values.get(CONF_USER_ID)) if values else None,
242 ),
243 )
244