music-assistant-server

1"""Several helpers/utils for the Plex Music Provider."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING, cast
7
8import requests
9from plexapi.gdm import GDM
10from plexapi.library import LibrarySection as PlexLibrarySection
11from plexapi.library import MusicSection as PlexMusicSection
12from plexapi.server import PlexServer
13
14if TYPE_CHECKING:
15    from music_assistant.mass import MusicAssistant
16
17
18async def get_libraries(
19    mass: MusicAssistant,
20    auth_token: str | None,
21    local_server_ssl: bool,
22    local_server_ip: str,
23    local_server_port: str,
24    local_server_verify_cert: bool,
25    instance_id: str | None = None,
26) -> list[str]:
27    """
28    Get all music libraries for all plex servers.
29
30    Returns a list of Library names in format ['servername / library name', ...]
31
32    :param mass: MusicAssistant instance.
33    :param auth_token: Authentication token for Plex server.
34    :param local_server_ssl: Whether to use SSL/HTTPS.
35    :param local_server_ip: IP address of the Plex server.
36    :param local_server_port: Port of the Plex server.
37    :param local_server_verify_cert: Whether to verify SSL certificate.
38    :param instance_id: Provider instance ID to use for cache isolation.
39    """
40    cache_key = "plex_libraries"
41
42    def _get_libraries() -> list[str]:
43        # create a listing of available music libraries on all servers
44        all_libraries: list[str] = []
45        session = requests.Session()
46        session.verify = local_server_verify_cert
47        local_server_protocol = "https" if local_server_ssl else "http"
48        plex_server: PlexServer
49        if auth_token is None:
50            plex_server = PlexServer(
51                f"{local_server_protocol}://{local_server_ip}:{local_server_port}"
52            )
53        else:
54            plex_server = PlexServer(
55                f"{local_server_protocol}://{local_server_ip}:{local_server_port}",
56                auth_token,
57                session=session,
58            )
59        for media_section in cast("list[PlexLibrarySection]", plex_server.library.sections()):
60            if media_section.type != PlexMusicSection.TYPE:
61                continue
62            # TODO: figure out what plex uses as stable id and use that instead of names
63            all_libraries.append(f"{plex_server.friendlyName} / {media_section.title}")
64        return all_libraries
65
66    if cache := await mass.cache.get(
67        cache_key, checksum=auth_token, provider=instance_id or local_server_ip
68    ):
69        return cast("list[str]", cache)
70
71    result = await asyncio.to_thread(_get_libraries)
72    # use short expiration for in-memory cache
73    await mass.cache.set(
74        cache_key,
75        result,
76        checksum=auth_token,
77        expiration=3600,
78        provider=instance_id or "default",
79    )
80    return result
81
82
83async def discover_local_servers() -> tuple[str, int] | tuple[None, None]:
84    """Discover all local plex servers on the network."""
85
86    def _discover_local_servers() -> tuple[str, int] | tuple[None, None]:
87        gdm = GDM()
88        gdm.scan()
89        if len(gdm.entries) > 0:
90            entry = gdm.entries[0]
91            data = entry.get("data")
92            local_server_ip = entry.get("from")[0]
93            local_server_port = data.get("Port")
94            return local_server_ip, local_server_port
95        return None, None
96
97    return await asyncio.to_thread(_discover_local_servers)
98