/
/
/
1"""Several helpers/utils for the Plex Music Provider."""
2
3from __future__ import annotations
4
5import asyncio
6import dataclasses
7import json
8import logging
9import re
10from dataclasses import dataclass
11from typing import TYPE_CHECKING, cast
12
13import requests
14from music_assistant_models.enums import ImageType, MediaType, ProviderFeature
15from music_assistant_models.media_items import MediaItemImage, UniqueList
16from plexapi.gdm import GDM
17from plexapi.library import LibrarySection as PlexLibrarySection
18from plexapi.library import MusicSection as PlexMusicSection
19from plexapi.server import PlexServer
20
21from music_assistant.providers.plex.constants import AUTH_TOKEN_UNAUTH
22
23if TYPE_CHECKING:
24 from plexapi.base import PlexObject
25
26 from music_assistant.mass import MusicAssistant
27
28LOGGER = logging.getLogger(__name__)
29
30# Matches the leading timestamp of an LRC lyric line, e.g. "[01:23.45]".
31_LRC_TIMESTAMP_RE = re.compile(r"^\s*\[\d{1,2}:\d{2}(?:[.:]\d{1,3})?\]", re.MULTILINE)
32
33# Library type constants
34CONF_LIBRARY_TYPE = "library_type"
35LIBRARY_TYPE_MUSIC = "music"
36LIBRARY_TYPE_AUDIOBOOKS = "audiobooks"
37LIBRARY_TYPE_PODCASTS = "podcasts"
38
39# Feature sets per library type
40SUPPORTED_FEATURES: set[ProviderFeature] = {
41 ProviderFeature.LIBRARY_ARTISTS,
42 ProviderFeature.LIBRARY_ALBUMS,
43 ProviderFeature.LIBRARY_TRACKS,
44 ProviderFeature.LIBRARY_PLAYLISTS,
45 ProviderFeature.FAVORITE_ALBUMS_EDIT,
46 ProviderFeature.FAVORITE_TRACKS_EDIT,
47 ProviderFeature.BROWSE,
48 ProviderFeature.SEARCH,
49 ProviderFeature.ARTIST_ALBUMS,
50 ProviderFeature.ARTIST_TOPTRACKS,
51 ProviderFeature.SIMILAR_TRACKS,
52 ProviderFeature.RECOMMENDATIONS,
53}
54
55AUDIOBOOK_FEATURES: set[ProviderFeature] = {
56 ProviderFeature.LIBRARY_AUDIOBOOKS,
57 ProviderFeature.BROWSE,
58 ProviderFeature.SEARCH,
59 ProviderFeature.RECOMMENDATIONS,
60}
61
62PODCAST_FEATURES: set[ProviderFeature] = {
63 ProviderFeature.LIBRARY_PODCASTS,
64 ProviderFeature.BROWSE,
65 ProviderFeature.SEARCH,
66 ProviderFeature.RECOMMENDATIONS,
67}
68
69# Mapping of library type to the media types it supports
70LIBRARY_TYPE_TO_MEDIA_TYPES: dict[str, tuple[MediaType, ...]] = {
71 LIBRARY_TYPE_MUSIC: (MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK, MediaType.PLAYLIST),
72 LIBRARY_TYPE_AUDIOBOOKS: (MediaType.AUDIOBOOK,),
73 LIBRARY_TYPE_PODCASTS: (MediaType.PODCAST, MediaType.PODCAST_EPISODE),
74}
75
76
77@dataclass(frozen=True)
78class PlexSectionInfo:
79 """Metadata about a Plex library section for config auto-detection."""
80
81 display_name: str
82 section_title: str
83 server_name: str
84 section_type: str
85 is_tracking_progress: bool
86
87
88def _library_tracks_progress(section: PlexLibrarySection) -> bool:
89 """
90 Check if a music section stores per-track progress (resumable content).
91
92 Uses the ``enableTrackOffsets`` library preference ("Store track progress"
93 advanced setting). When enabled, Plex treats tracks as resumable content,
94 which is characteristic of long-form audio libraries such as audiobooks and podcasts.
95 """
96 try:
97 settings = section.settings()
98 section_title = getattr(section, "title", "<unknown>")
99 LOGGER.debug(
100 "Library '%s' settings: %r",
101 section_title,
102 [{"id": s.id, "value": s.value, "type": getattr(s, "type", "?")} for s in settings],
103 )
104 for setting in settings:
105 if setting.id == "enableTrackOffsets":
106 # Plex may return the value as a bool or string.
107 val = setting.value
108 is_enabled = val is True or (isinstance(val, str) and val.lower() in ("true", "1"))
109 if is_enabled:
110 LOGGER.debug(
111 "Library '%s' flagged as resumable (enableTrackOffsets=%s)",
112 section_title,
113 val,
114 )
115 return True
116 LOGGER.debug(
117 "Library '%s' not flagged as resumable (enableTrackOffsets absent or disabled)",
118 section_title,
119 )
120 except Exception as err:
121 LOGGER.warning(
122 "Failed to read library settings for '%s': %s",
123 getattr(section, "title", "<unknown>"),
124 err,
125 )
126 return False
127
128
129def extract_library_name(conf_value: str) -> str:
130 """
131 Extract the library name from a config value that may include server name.
132
133 Config values from get_config_entries include the server prefix:
134 '<server name> / <library name>'. When typed manually by the user,
135 the value may just be '<library name>'.
136
137 :param conf_value: The raw config value for a library setting.
138 :return: The library name (without server prefix if present).
139 """
140 if " / " in conf_value:
141 return conf_value.split(" / ", 1)[1].strip()
142 return conf_value.strip()
143
144
145async def get_section_info(
146 mass: MusicAssistant,
147 auth_token: str | None,
148 local_server_ssl: bool,
149 local_server_ip: str,
150 local_server_port: str,
151 local_server_verify_cert: bool,
152 instance_id: str | None = None,
153) -> list[PlexSectionInfo]:
154 """
155 Get metadata for all music library sections on the Plex server.
156
157 Returns PlexSectionInfo objects including auto-detection hints for resumable content.
158
159 :param mass: MusicAssistant instance.
160 :param auth_token: Authentication token for Plex server.
161 :param local_server_ssl: Whether to use SSL/HTTPS.
162 :param local_server_ip: IP address of the Plex server.
163 :param local_server_port: Port of the Plex server.
164 :param local_server_verify_cert: Whether to verify SSL certificate.
165 :param instance_id: Provider instance ID to use for cache isolation.
166 """
167 cache_key = "plex_section_info"
168 cache_provider = instance_id or local_server_ip
169
170 def _get_section_info() -> list[PlexSectionInfo]:
171 session = requests.Session()
172 session.verify = local_server_verify_cert
173 local_server_protocol = "https" if local_server_ssl else "http"
174 plex_server: PlexServer
175 plex_url = f"{local_server_protocol}://{local_server_ip}:{local_server_port}"
176 try:
177 if not auth_token or auth_token == AUTH_TOKEN_UNAUTH:
178 # local (unauthenticated) connection, not via plex.tv
179 plex_server = PlexServer(plex_url, session=session)
180 else:
181 plex_server = PlexServer(plex_url, auth_token, session=session)
182 except requests.exceptions.ConnectionError as err:
183 LOGGER.warning(
184 "Could not connect to Plex server at %s:%s: %s",
185 local_server_ip,
186 local_server_port,
187 err,
188 )
189 return []
190 results: list[PlexSectionInfo] = []
191 for media_section in cast("list[PlexLibrarySection]", plex_server.library.sections()):
192 if media_section.type != PlexMusicSection.TYPE:
193 continue
194 results.append(
195 PlexSectionInfo(
196 display_name=f"{plex_server.friendlyName} / {media_section.title}",
197 section_title=media_section.title,
198 server_name=plex_server.friendlyName,
199 section_type=media_section.type,
200 is_tracking_progress=_library_tracks_progress(media_section),
201 )
202 )
203 return results
204
205 if cache := await mass.cache.get(cache_key, checksum=auth_token, provider=cache_provider):
206 if isinstance(cache, list) and cache and all(isinstance(item, dict) for item in cache):
207 try:
208 return [PlexSectionInfo(**item) for item in cache]
209 except TypeError:
210 LOGGER.debug("Discarding corrupt plex_section_info cache entry", exc_info=True)
211 else:
212 LOGGER.debug("Discarding corrupt plex_section_info cache entry: %r", type(cache))
213
214 result = await asyncio.to_thread(_get_section_info)
215 await mass.cache.set(
216 cache_key,
217 [dataclasses.asdict(section) for section in result],
218 checksum=auth_token,
219 expiration=3600,
220 provider=cache_provider,
221 )
222 return result
223
224
225async def discover_local_servers() -> tuple[str, int] | tuple[None, None]:
226 """Discover all local plex servers on the network."""
227
228 def _discover_local_servers() -> tuple[str, int] | tuple[None, None]:
229 gdm = GDM()
230 gdm.scan()
231 if len(gdm.entries) > 0:
232 entry = gdm.entries[0]
233 data = entry.get("data")
234 local_server_ip = entry.get("from")[0]
235 local_server_port = data.get("Port")
236 return local_server_ip, local_server_port
237 return None, None
238
239 return await asyncio.to_thread(_discover_local_servers)
240
241
242def get_thumbnail_images(
243 plex_media: PlexObject,
244 provider_instance_id: str,
245 attrs: tuple[str, ...] = ("thumb", "parentThumb", "grandparentThumb"),
246) -> UniqueList[MediaItemImage] | None:
247 """
248 Get the thumbnail of a Plex object as MA image list, if available.
249
250 :param plex_media: The Plex object to extract the thumbnail from.
251 :param provider_instance_id: The provider instance id to set on the image.
252 :param attrs: Plex attributes to check (in order) for a thumbnail.
253 """
254 if thumb := plex_media.firstAttr(*attrs):
255 return UniqueList(
256 [
257 MediaItemImage(
258 type=ImageType.THUMB,
259 path=thumb,
260 provider=provider_instance_id,
261 remotely_accessible=False,
262 )
263 ]
264 )
265 return None
266
267
268def get_favorite_from_rating(plex_media: PlexObject, threshold: float) -> bool | None:
269 """
270 Derive favorite status from the user rating of a Plex object.
271
272 Returns None if the object has no user rating.
273
274 :param plex_media: The Plex object to read the user rating from.
275 :param threshold: Minimum rating (0.0-10.0) to consider the item a favorite.
276 """
277 rating = getattr(plex_media, "userRating", None)
278 if rating is None:
279 return None
280 return float(rating) >= threshold
281
282
283def get_explicit(plex_media: PlexObject) -> bool | None:
284 """
285 Derive explicit status from a Plex object's content rating.
286
287 Returns True or False based on the content rating, or None when it is unset.
288
289 :param plex_media: The Plex object to read the content rating from.
290 """
291 # contentRating is not typed by plexapi on audio items, so read it from the raw payload.
292 content_rating = plex_media._data.attrib.get("contentRating")
293 if not content_rating or not isinstance(content_rating, str):
294 return None
295 return content_rating.lower() == "explicit"
296
297
298def get_musicbrainz_id(plex_media: PlexObject) -> str | None:
299 """
300 Get the MusicBrainz identifier from a Plex object's guids, if available.
301
302 :param plex_media: The Plex object (artist, album or track) to read from.
303 """
304 # Read guids from the cached payload so partial listing objects don't trigger a reload.
305 for guid in plex_media._data.findall("Guid"):
306 guid_id = str(guid.attrib.get("id") or "")
307 if guid_id.startswith("mbid://"):
308 return guid_id.removeprefix("mbid://")
309 return None
310
311
312def parse_plex_lyrics_payload(content: str) -> tuple[str, bool] | None:
313 """
314 Parse a Plex lyric stream payload into ``(lyrics, is_synced)``.
315
316 Returns the lyrics as an LRC string when synced, plain text when not, or
317 None when no usable lyrics are found.
318
319 :param content: The raw lyric stream body returned by Plex.
320 """
321 if not content or not content.strip():
322 return None
323 # Sniff the payload shape: structured JSON, then timestamped LRC, then plain text.
324 if (parsed := _lyrics_from_plex_json(content)) is not None:
325 return parsed
326 if _LRC_TIMESTAMP_RE.search(content):
327 return content.strip(), True
328 return content.strip(), False
329
330
331def _lyrics_from_plex_json(content: str) -> tuple[str, bool] | None:
332 """
333 Parse Plex structured JSON lyrics into ``(lyrics, is_synced)`` or None.
334
335 :param content: The raw lyric stream body to parse as JSON.
336 """
337 try:
338 data = json.loads(content)
339 except ValueError, TypeError:
340 return None
341 if not isinstance(data, dict):
342 return None
343 container = data.get("MediaContainer")
344 if not isinstance(container, dict):
345 return None
346 lyrics_objs = container.get("Lyrics")
347 if not isinstance(lyrics_objs, list) or not lyrics_objs:
348 return None
349 raw_lines = lyrics_objs[0].get("Line") if isinstance(lyrics_objs[0], dict) else None
350 if not isinstance(raw_lines, list):
351 return None
352 lrc_lines: list[str] = []
353 plain_lines: list[str] = []
354 synced = False
355 for raw_line in raw_lines:
356 spans = raw_line.get("Span") if isinstance(raw_line, dict) else None
357 if not isinstance(spans, list):
358 lrc_lines.append("")
359 plain_lines.append("")
360 continue
361 text = "".join(span.get("text") or "" for span in spans if isinstance(span, dict))
362 plain_lines.append(text)
363 start_offset = spans[0].get("startOffset") if spans and isinstance(spans[0], dict) else None
364 if isinstance(start_offset, int) and not isinstance(start_offset, bool):
365 synced = True
366 lrc_lines.append(f"[{_ms_to_lrc_timestamp(start_offset)}]{text}")
367 else:
368 lrc_lines.append(text)
369 if synced:
370 return "\n".join(lrc_lines), True
371 plain = "\n".join(plain_lines).strip()
372 return (plain, False) if plain else None
373
374
375def _ms_to_lrc_timestamp(milliseconds: int) -> str:
376 """
377 Format a millisecond offset as an LRC ``mm:ss.cc`` timestamp body.
378
379 :param milliseconds: The offset from the start of the track, in milliseconds.
380 """
381 minutes, remainder = divmod(max(milliseconds, 0), 60000)
382 seconds, remainder = divmod(remainder, 1000)
383 return f"{minutes:02d}:{seconds:02d}.{remainder // 10:02d}"
384