/
/
/
1"""Setup flow for the Last.fm (and Libre.fm) scrobble provider."""
2
3from __future__ import annotations
4
5import asyncio
6from dataclasses import replace
7from typing import TYPE_CHECKING
8
9import pylast
10from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
11from music_assistant_models.enums import ConfigEntryType
12from music_assistant_models.errors import SetupFailedError
13
14from music_assistant.models.setup_flow import SetupFlowError
15from music_assistant.providers.lastfm_scrobble import (
16 CONF_API_KEY,
17 CONF_API_SECRET,
18 CONF_PROVIDER,
19 CONF_SESSION_KEY,
20 CONF_USERNAME,
21 _NetworkType,
22 _resolve_credentials,
23 get_network,
24)
25
26if TYPE_CHECKING:
27 from music_assistant.models.setup_flow import SetupSession
28
29# Last.fm ships with a built-in app (key/secret from app_vars); only Libre.fm - or a
30# custom Last.fm developer app - needs the advanced key/secret fields below
31CONF_ENTRY_PROVIDER = ConfigEntry(
32 key=CONF_PROVIDER,
33 type=ConfigEntryType.STRING,
34 required=True,
35 options=[
36 ConfigValueOption(_NetworkType.LASTFM.value),
37 ConfigValueOption(_NetworkType.LIBREFM.value),
38 ],
39 default_value=_NetworkType.LASTFM.value,
40)
41CONF_ENTRY_API_KEY = ConfigEntry(
42 key=CONF_API_KEY, type=ConfigEntryType.SECURE_STRING, required=False, advanced=True
43)
44CONF_ENTRY_API_SECRET = ConfigEntry(
45 key=CONF_API_SECRET, type=ConfigEntryType.SECURE_STRING, required=False, advanced=True
46)
47
48
49async def run_setup(session: SetupSession) -> None:
50 """
51 Run the Last.fm (or Libre.fm) scrobbler setup flow.
52
53 Collects the network and, when overriding the built-in Last.fm app, its API
54 credentials, then completes the web-auth handshake and persists the resulting
55 (permanent) session key.
56
57 :param session: The setup session driving the flow.
58 """
59 network_default = str(session.context.setup_data.get(CONF_PROVIDER, _NetworkType.LASTFM.value))
60 errors: dict[str, str] | None = None
61 while True:
62 values = await session.form(
63 [
64 replace(CONF_ENTRY_PROVIDER, value=network_default),
65 CONF_ENTRY_API_KEY,
66 CONF_ENTRY_API_SECRET,
67 ],
68 step_id="user",
69 errors=errors,
70 )
71 network_default = str(values[CONF_PROVIDER])
72 network_type = _NetworkType(network_default)
73 try:
74 api_key, api_secret = _resolve_credentials(values, network_type)
75 except SetupFailedError as err:
76 errors = {"base": str(err)}
77 continue
78
79 network = await asyncio.to_thread(
80 get_network, {**values, CONF_API_KEY: api_key, CONF_API_SECRET: api_secret}
81 )
82 # pylast only implements desktop auth; build the web-auth URL (with our own
83 # dynamic callback) ourselves - the session key exchange below doesn't care
84 # how the token was obtained
85 url = f"{network.homepage}/api/auth/?api_key={network.api_key}&cb={session.callback_url}"
86 params = await session.external(url, step_id="auth")
87 if not (token := params.get("token")):
88 errors = {"base": "auth_failed"}
89 continue
90
91 try:
92 # blocking IO (and MD5 request signing, handled internally by pylast); the
93 # url argument is unused by pylast here, only the token is
94 session_key, username = await asyncio.to_thread(
95 pylast.SessionKeyGenerator(network).get_web_auth_session_key_username, url, token
96 )
97 except pylast.PyLastError as err:
98 errors = {"base": str(err)}
99 continue
100
101 finish_values: dict[str, ConfigValueType] = {
102 CONF_PROVIDER: network_type.value,
103 CONF_SESSION_KEY: session_key,
104 CONF_USERNAME: username,
105 }
106 # only persist a custom key/secret when the user actually entered one; the
107 # built-in Last.fm credentials are re-resolved from app_vars at runtime
108 if values.get(CONF_API_KEY) and values.get(CONF_API_SECRET):
109 finish_values[CONF_API_KEY] = api_key
110 finish_values[CONF_API_SECRET] = api_secret
111 try:
112 await session.finish(finish_values)
113 return
114 except SetupFlowError as err:
115 errors = {"base": err.translation_key or str(err)}
116