music-assistant-server

1"""Setup flow for the Plex music provider."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import ConfigEntryType
10from plexapi.myplex import MyPlexPinLogin
11
12from music_assistant.models.setup_flow import AbortFlow, SetupFlowError
13
14from .constants import (
15    AUTH_TOKEN_UNAUTH,
16    CONF_AUTH_TOKEN,
17    CONF_LIBRARY_ID,
18    CONF_LOCAL_SERVER_IP,
19    CONF_LOCAL_SERVER_PORT,
20    CONF_LOCAL_SERVER_SSL,
21    CONF_LOCAL_SERVER_VERIFY_CERT,
22)
23from .helpers import (
24    CONF_LIBRARY_TYPE,
25    LIBRARY_TYPE_AUDIOBOOKS,
26    LIBRARY_TYPE_MUSIC,
27    LIBRARY_TYPE_PODCASTS,
28    discover_local_servers,
29    get_section_info,
30)
31
32if TYPE_CHECKING:
33    from music_assistant_models.config_entries import ConfigValueType
34
35    from music_assistant.mass import MusicAssistant
36    from music_assistant.models.setup_flow import SetupSession
37
38    from .helpers import PlexSectionInfo
39
40AUTH_METHOD_MYPLEX = "myplex"
41AUTH_METHOD_LOCAL = "local"
42
43
44async def run_setup(session: SetupSession) -> None:
45    """
46    Run the Plex setup flow.
47
48    Collects the local server details, authenticates (MyPlex OAuth or a tokenless local
49    connection), lets the user pick the music/audiobook/podcast library, and persists the
50    connection + selection as setup data.
51
52    :param session: The setup session driving the flow.
53    """
54    setup_data = dict(session.context.setup_data)
55    discovered = await _discover_servers(session)
56    # server details first (matches the pre-flow UI order)
57    server = await _collect_server(session, setup_data, discovered)
58    # authenticate once: MyPlex OAuth is expensive to repeat on a retry
59    token = await _authenticate(session)
60    errors: dict[str, str] | None = None
61    while True:
62        if errors is not None:
63            # a previous attempt failed: let the user correct the server details
64            server = await _collect_server(session, setup_data, discovered, errors=errors)
65        sections = await session.progress_until(
66            get_section_info(
67                session.mass,
68                token,
69                bool(server[CONF_LOCAL_SERVER_SSL]),
70                str(server[CONF_LOCAL_SERVER_IP]),
71                str(server[CONF_LOCAL_SERVER_PORT]),
72                bool(server[CONF_LOCAL_SERVER_VERIFY_CERT]),
73                session.context.instance_id,
74            ),
75            step_id="loading_libraries",
76            text="loading_libraries",
77            expires_in=60,
78        )
79        if not sections:
80            errors = {"base": "no_libraries"}
81            continue
82        library_id, library_type = await _collect_library(
83            session, sections, setup_data, session.context.instance_id
84        )
85        try:
86            await session.finish(
87                {
88                    CONF_AUTH_TOKEN: token,
89                    **server,
90                    CONF_LIBRARY_ID: library_id,
91                    CONF_LIBRARY_TYPE: library_type,
92                }
93            )
94            return
95        except SetupFlowError as err:
96            errors = {"base": err.translation_key or str(err)}
97
98
99async def _discover_servers(session: SetupSession) -> tuple[str | None, int | None]:
100    """Run a best-effort GDM discovery to prefill the server details."""
101    try:
102        return await session.progress_until(
103            discover_local_servers(),
104            step_id="discovering",
105            text="discovering_servers",
106            expires_in=15,
107        )
108    except Exception:
109        # discovery is only a convenience prefill; never let it break the flow
110        return None, None
111
112
113async def _collect_server(
114    session: SetupSession,
115    setup_data: dict[str, ConfigValueType],
116    discovered: tuple[str | None, int | None],
117    errors: dict[str, str] | None = None,
118) -> dict[str, ConfigValueType]:
119    """Show the server-details form and return the collected connection settings."""
120    ip_default = setup_data.get(CONF_LOCAL_SERVER_IP) or discovered[0]
121    port_default = setup_data.get(CONF_LOCAL_SERVER_PORT) or discovered[1] or 32400
122    values = await session.form(
123        [
124            ConfigEntry(
125                key=CONF_LOCAL_SERVER_IP,
126                type=ConfigEntryType.STRING,
127                required=True,
128                value=ip_default,
129            ),
130            ConfigEntry(
131                key=CONF_LOCAL_SERVER_PORT,
132                type=ConfigEntryType.INTEGER,
133                required=True,
134                default_value=32400,
135                value=port_default,
136            ),
137            ConfigEntry(
138                key=CONF_LOCAL_SERVER_SSL,
139                type=ConfigEntryType.BOOLEAN,
140                required=True,
141                default_value=False,
142                value=setup_data.get(CONF_LOCAL_SERVER_SSL),
143            ),
144            ConfigEntry(
145                key=CONF_LOCAL_SERVER_VERIFY_CERT,
146                type=ConfigEntryType.BOOLEAN,
147                required=True,
148                default_value=True,
149                depends_on=CONF_LOCAL_SERVER_SSL,
150                advanced=True,
151                value=setup_data.get(CONF_LOCAL_SERVER_VERIFY_CERT),
152            ),
153        ],
154        step_id="server",
155        errors=errors,
156    )
157    return {
158        CONF_LOCAL_SERVER_IP: str(values[CONF_LOCAL_SERVER_IP]),
159        CONF_LOCAL_SERVER_PORT: int(values[CONF_LOCAL_SERVER_PORT]),  # type: ignore[arg-type]
160        CONF_LOCAL_SERVER_SSL: bool(values[CONF_LOCAL_SERVER_SSL]),
161        CONF_LOCAL_SERVER_VERIFY_CERT: bool(values[CONF_LOCAL_SERVER_VERIFY_CERT]),
162    }
163
164
165async def _authenticate(session: SetupSession) -> str:
166    """Pick an auth method and return the Plex auth token (or the local sentinel)."""
167    method = (
168        await session.form(
169            [
170                ConfigEntry(
171                    key="auth_method",
172                    type=ConfigEntryType.STRING,
173                    required=True,
174                    default_value=AUTH_METHOD_MYPLEX,
175                    options=[
176                        ConfigValueOption(value=AUTH_METHOD_MYPLEX),
177                        ConfigValueOption(value=AUTH_METHOD_LOCAL),
178                    ],
179                )
180            ],
181            step_id="auth_method",
182        )
183    )["auth_method"]
184    if method == AUTH_METHOD_LOCAL:
185        return AUTH_TOKEN_UNAUTH
186    plex_auth = MyPlexPinLogin(headers={"X-Plex-Product": "Music Assistant"}, oauth=True)
187    await asyncio.to_thread(plex_auth._getCode)
188    auth_url = plex_auth.oauthUrl(session.callback_url)
189    # the callback is only a "user came back" signal; the token is fetched by polling
190    await session.external(auth_url, step_id="myplex_auth", expires_in=600)
191    token = await session.progress_until(
192        _poll_myplex_token(plex_auth),
193        step_id="finalizing_auth",
194        text="finalizing_auth",
195        expires_in=45,
196    )
197    if not token:
198        raise AbortFlow("myplex_auth_failed")
199    return token
200
201
202async def _poll_myplex_token(plex_auth: MyPlexPinLogin) -> str | None:
203    """Poll plex.tv until the PIN is linked and the auth token becomes available."""
204    for attempt in range(8):
205        if await asyncio.to_thread(plex_auth.checkLogin):
206            break
207        await asyncio.sleep(min(0.5 * (2**attempt), 5))
208    return str(plex_auth.token) if plex_auth.token else None
209
210
211async def _collect_library(
212    session: SetupSession,
213    sections: list[PlexSectionInfo],
214    setup_data: dict[str, ConfigValueType],
215    instance_id: str | None,
216) -> tuple[str, str]:
217    """Show the library-selection form (with smart defaults) and return (library_id, type)."""
218    used_libraries, has_audiobook_provider = _claimed_libraries(session.mass, instance_id)
219    prefilled_library = setup_data.get(CONF_LIBRARY_ID)
220    default_library, suggested_type = _default_library_selection(
221        sections,
222        used_libraries,
223        has_audiobook_provider,
224        str(prefilled_library) if prefilled_library else None,
225    )
226    # a stored library type (reconfigure) wins over the freshly derived suggestion
227    default_type = str(setup_data.get(CONF_LIBRARY_TYPE) or suggested_type)
228    values = await session.form(
229        [
230            ConfigEntry(
231                key=CONF_LIBRARY_ID,
232                type=ConfigEntryType.STRING,
233                required=True,
234                options=[
235                    ConfigValueOption(title=s.display_name, value=s.display_name) for s in sections
236                ],
237                default_value=default_library,
238                value=default_library,
239            ),
240            ConfigEntry(
241                key=CONF_LIBRARY_TYPE,
242                type=ConfigEntryType.STRING,
243                required=True,
244                options=[
245                    ConfigValueOption(value=LIBRARY_TYPE_MUSIC),
246                    ConfigValueOption(value=LIBRARY_TYPE_AUDIOBOOKS),
247                    ConfigValueOption(value=LIBRARY_TYPE_PODCASTS),
248                ],
249                default_value=default_type,
250                value=default_type,
251            ),
252        ],
253        step_id="library",
254        last_step=True,
255    )
256    return str(values[CONF_LIBRARY_ID]), str(values[CONF_LIBRARY_TYPE])
257
258
259def _claimed_libraries(mass: MusicAssistant, instance_id: str | None) -> tuple[set[str], bool]:
260    """
261    Return the libraries already claimed by other plex instances (and if an audiobook one exists).
262
263    Reads the raw stored provider config directly to avoid recursively triggering
264    get_config_entries for every plex instance.
265
266    :param mass: The MusicAssistant instance.
267    :param instance_id: The instance being configured (excluded from the claim scan).
268    """
269    used_libraries: set[str] = set()
270    has_audiobook_provider = False
271    for prov_id, prov_conf in mass.config.get("providers", {}).items():
272        if prov_conf.get("domain") != "plex" or prov_id == instance_id:
273            continue
274        prov_setup = prov_conf.get("setup_data", {})
275        prov_values = prov_conf.get("values", {})
276        if lib_val := (prov_setup.get(CONF_LIBRARY_ID) or prov_values.get(CONF_LIBRARY_ID)):
277            used_libraries.add(str(lib_val))
278        if LIBRARY_TYPE_AUDIOBOOKS in (
279            prov_setup.get(CONF_LIBRARY_TYPE),
280            prov_values.get(CONF_LIBRARY_TYPE),
281        ):
282            has_audiobook_provider = True
283    return used_libraries, has_audiobook_provider
284
285
286def _default_library_selection(
287    sections: list[PlexSectionInfo],
288    used_libraries: set[str],
289    has_audiobook_provider: bool,
290    prefilled_library: str | None,
291) -> tuple[str, str]:
292    """
293    Derive the default (library display name, library type) for the selection form.
294
295    :param sections: All music sections discovered on the server.
296    :param used_libraries: Libraries already claimed by other plex instances.
297    :param has_audiobook_provider: Whether another plex instance already serves audiobooks.
298    :param prefilled_library: A previously selected library (reconfigure) to keep, if any.
299    """
300    available = [s for s in sections if s.display_name not in used_libraries] or sections
301    if prefilled_library:
302        default_library = prefilled_library
303    else:
304        # non-tracking libraries first (music), then alphabetically
305        default_library = sorted(available, key=lambda s: (s.is_tracking_progress, s.display_name))[
306            0
307        ].display_name
308    selected = next((s for s in sections if s.display_name == default_library), None)
309    if selected and selected.is_tracking_progress:
310        default_type = LIBRARY_TYPE_PODCASTS if has_audiobook_provider else LIBRARY_TYPE_AUDIOBOOKS
311    else:
312        default_type = LIBRARY_TYPE_MUSIC
313    return default_library, default_type
314