/
/
/
1"""WebDAV File System Provider for Music Assistant."""
2
3from __future__ import annotations
4
5from dataclasses import asdict
6from pathlib import PurePosixPath
7from typing import TYPE_CHECKING, cast
8from urllib.parse import quote, unquote, urlparse, urlunparse
9
10import aiohttp
11from music_assistant_models.errors import (
12 LoginFailed,
13 MediaNotFoundError,
14 ProviderUnavailableError,
15 SetupFailedError,
16)
17
18from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
19from music_assistant.controllers.tasks.context import update_current_task_progress_text
20from music_assistant.helpers.tags import get_embedded_image
21from music_assistant.providers.filesystem_local import LocalFileSystemProvider
22from music_assistant.providers.filesystem_local.constants import (
23 CONF_ENTRY_CONTENT_TYPE,
24 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
25 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
26 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
27 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
28 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
29 CONF_ENTRY_MISSING_ALBUM_ARTIST,
30 CONF_ENTRY_PROPAGATE_GENRES,
31 SUPPORTED_EXTENSIONS,
32 WALK_EXTENSIONS,
33 content_type_config_entry,
34)
35from music_assistant.providers.filesystem_local.helpers import FileSystemItem, ScanErrors
36
37from .constants import CONF_CONTENT_TYPE, CONF_URL, CONF_VERIFY_SSL
38from .helpers import WebDAVItem, build_webdav_url, webdav_propfind, webdav_test_connection
39
40if TYPE_CHECKING:
41 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
42 from music_assistant_models.provider import ProviderManifest
43
44 from music_assistant.mass import MusicAssistant
45
46
47class WebDAVFileSystemProvider(LocalFileSystemProvider):
48 """WebDAV File System Provider for Music Assistant."""
49
50 # WebDAV servers often struggle with 16 parallel tag-parse GETs
51 _SYNC_CONCURRENCY = 4
52
53 def __init__(
54 self,
55 mass: MusicAssistant,
56 manifest: ProviderManifest,
57 config: ProviderConfig,
58 ) -> None:
59 """Initialize WebDAV FileSystem Provider."""
60 # the base path (WebDAV URL) is resolved from the setup data below, which needs
61 # the initialized instance, so hand the base class a placeholder and set it after
62 super().__init__(mass, manifest, config, base_path="")
63 self.base_url = cast("str", self.get_setup_value(CONF_URL)).rstrip("/")
64 self.base_path = self.base_url
65 self.username = cast("str | None", self.get_setup_value(CONF_USERNAME))
66 self.password = cast("str | None", self.get_setup_value(CONF_PASSWORD))
67 self.verify_ssl = cast("bool", self.get_setup_value(CONF_VERIFY_SSL))
68 self.media_content_type = cast(
69 "str", self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
70 )
71
72 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
73 """Return Config entries to setup this provider."""
74 # connection details and content type are collected by the setup flow; surface the
75 # (immutable) content type read-only so the sync options' depends_on chains resolve
76 content_type = str(
77 self.get_setup_value(CONF_CONTENT_TYPE, CONF_ENTRY_CONTENT_TYPE.default_value)
78 )
79 return (
80 content_type_config_entry(content_type),
81 CONF_ENTRY_MISSING_ALBUM_ARTIST,
82 CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS,
83 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
84 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
85 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
86 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
87 CONF_ENTRY_PROPAGATE_GENRES,
88 )
89
90 @property
91 def instance_name_postfix(self) -> str | None:
92 """Return a (default) instance name postfix for this provider instance."""
93 parsed = urlparse(self.base_url)
94 if parsed.path and parsed.path != "/":
95 return PurePosixPath(parsed.path).name
96 return parsed.netloc
97
98 @property
99 def _auth_header(self) -> str | None:
100 """Return the WebDAV Authorization header value, or None when no credentials are set."""
101 if self.username:
102 return aiohttp.encode_basic_auth(self.username, self.password or "")
103 return None
104
105 @property
106 def _session(self) -> aiohttp.ClientSession:
107 """Get the appropriate HTTP session based on SSL verification setting."""
108 return self.mass.http_session if self.verify_ssl else self.mass.http_session_no_ssl
109
110 async def handle_async_init(self) -> None:
111 """Handle async initialization of the provider."""
112 session = self._session
113 await webdav_test_connection(
114 session,
115 self.base_url,
116 self.username,
117 self.password,
118 timeout=10,
119 )
120 self.write_access = False
121
122 def _build_authenticated_url(self, file_path: str) -> str:
123 """Build authenticated WebDAV URL with properly encoded credentials."""
124 webdav_url = build_webdav_url(self.base_url, file_path)
125 if not (self.username and self.password):
126 return webdav_url
127
128 parsed = urlparse(webdav_url)
129 encoded_username = quote(self.username, safe="")
130 encoded_password = quote(self.password, safe="")
131 netloc = f"{encoded_username}:{encoded_password}@{parsed.netloc}"
132 return urlunparse(
133 (parsed.scheme, netloc, parsed.path, parsed.params, parsed.query, parsed.fragment)
134 )
135
136 def _normalize_path(self, path: str) -> str:
137 """Convert absolute URL to relative path if needed."""
138 if path.startswith("http"):
139 parsed = urlparse(path)
140 base_parsed = urlparse(self.base_url)
141 return parsed.path[len(base_parsed.path) :].strip("/")
142 return path
143
144 async def exists(self, file_path: str) -> bool:
145 """Check if WebDAV resource exists."""
146 if not file_path:
147 return False
148 file_path = self._normalize_path(file_path)
149 webdav_url = build_webdav_url(self.base_url, file_path)
150 session = self._session
151 try:
152 items = await webdav_propfind(
153 session, webdav_url, depth=0, auth_header=self._auth_header
154 )
155 return len(items) > 0 or webdav_url.rstrip("/") == self.base_url.rstrip("/")
156 except LoginFailed, SetupFailedError, ProviderUnavailableError:
157 raise
158 except aiohttp.ClientError:
159 return False
160
161 async def resolve(self, file_path: str) -> FileSystemItem:
162 """Resolve WebDAV path to FileSystemItem."""
163 webdav_url = build_webdav_url(self.base_url, file_path)
164 session = self._session
165
166 items = await webdav_propfind(session, webdav_url, depth=0, auth_header=self._auth_header)
167 if not items:
168 # Handle root directory case
169 if webdav_url.rstrip("/") == self.base_url.rstrip("/"):
170 return FileSystemItem(
171 filename="",
172 relative_path="",
173 absolute_path=self._build_authenticated_url(file_path),
174 is_dir=True,
175 )
176 raise MediaNotFoundError(f"WebDAV resource not found: {file_path}")
177
178 webdav_item = items[0]
179 return FileSystemItem(
180 filename=PurePosixPath(file_path).name or webdav_item.name,
181 relative_path=file_path,
182 absolute_path=self._build_authenticated_url(file_path),
183 is_dir=webdav_item.is_dir,
184 checksum=webdav_item.last_modified or "unknown",
185 file_size=webdav_item.size,
186 metadata_token=webdav_item.etag,
187 )
188
189 async def _scandir(self, path: str, use_cache: bool = True) -> list[FileSystemItem]:
190 """List WebDAV directory contents with caching."""
191 cache_key = f"scandir_{path}"
192 # bypass the cache during sync (edits must be picked up immediately) or when the
193 # caller explicitly asks not to use it (e.g. an on-demand NFO lookup honoring a manual
194 # "Refresh item"); the fresh result is still written back for subsequent browse/exists
195 # calls
196 if use_cache and not self.sync_running:
197 if cached := await self.cache.get(
198 key=cache_key,
199 provider=self.instance_id,
200 category=0,
201 ):
202 return [FileSystemItem(**item) for item in cached]
203
204 path = self._normalize_path(path)
205 webdav_url = build_webdav_url(self.base_url, path)
206 session = self._session
207
208 webdav_items = await webdav_propfind(
209 session, webdav_url, depth=1, auth_header=self._auth_header
210 )
211 filesystem_items = self._convert_webdav_items(webdav_items, path)
212
213 await self.cache.set(
214 key=cache_key,
215 data=[asdict(item) for item in filesystem_items],
216 provider=self.instance_id,
217 category=0,
218 expiration=300,
219 )
220 return filesystem_items
221
222 async def _read_file(self, path: str) -> bytes:
223 """Read file contents over HTTP."""
224 webdav_url = build_webdav_url(self.base_url, path)
225 session = self._session
226 auth_header = self._auth_header
227 headers = {"Authorization": auth_header} if auth_header else None
228 async with session.get(webdav_url, headers=headers) as resp:
229 if resp.status != 200:
230 raise MediaNotFoundError(f"File not found: {path}")
231 return await resp.read()
232
233 def _convert_webdav_items(
234 self,
235 webdav_items: list[WebDAVItem],
236 scan_path: str,
237 ) -> list[FileSystemItem]:
238 """Convert WebDAV items to FileSystemItems."""
239 base_path = urlparse(self.base_url).path.rstrip("/")
240 result: list[FileSystemItem] = []
241
242 for item in webdav_items:
243 # Skip recycle bins
244 if "#recycle" in item.name.lower():
245 continue
246
247 decoded_href = unquote(item.href)
248 if decoded_href.startswith(("http://", "https://")):
249 # Extract the path by hand: urlparse would treat ; ? # in the path as
250 # params/query/fragment and corrupt names containing those characters.
251 after_scheme = decoded_href.split("://", 1)[1]
252 href_path = after_scheme[after_scheme.find("/") :] if "/" in after_scheme else ""
253 else:
254 href_path = decoded_href
255
256 # Calculate relative path
257 if href_path.startswith(base_path):
258 relative_path = href_path[len(base_path) :].strip("/")
259 else:
260 decoded_name = unquote(item.name)
261 relative_path = (
262 str(PurePosixPath(scan_path) / decoded_name) if scan_path else decoded_name
263 )
264
265 # Skip the directory being scanned itself (a depth-1 PROPFIND returns it too).
266 # Comparing on the resolved relative path is reliable even when the name holds
267 # characters a URL parser treats specially (e.g. ; ? #), which would otherwise
268 # make the directory list itself and recurse endlessly.
269 if relative_path == scan_path:
270 continue
271
272 result.append(
273 FileSystemItem(
274 filename=unquote(item.name),
275 relative_path=relative_path,
276 absolute_path=self._build_authenticated_url(relative_path),
277 is_dir=item.is_dir,
278 checksum=item.last_modified or "unknown",
279 file_size=item.size,
280 metadata_token=item.etag,
281 )
282 )
283 return result
284
285 async def resolve_image(self, path: str) -> str | bytes:
286 """Resolve image path to actual image data or URL."""
287 # Check if this is an audio file with embedded image
288 ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
289 if ext in SUPPORTED_EXTENSIONS:
290 # Use authenticated URL for ffmpeg to extract embedded image
291 auth_url = self._build_authenticated_url(path)
292 if img_data := await get_embedded_image(auth_url):
293 return img_data
294 raise MediaNotFoundError(f"No embedded image found: {path}")
295
296 # For actual image files, fetch the raw bytes
297 webdav_url = build_webdav_url(self.base_url, path)
298 session = self._session
299 auth_header = self._auth_header
300 headers = {"Authorization": auth_header} if auth_header else None
301 async with session.get(webdav_url, headers=headers) as resp:
302 if resp.status != 200:
303 raise MediaNotFoundError(f"Image not found: {path}")
304 return await resp.read()
305
306 async def _enumerate_files_for_sync(
307 self,
308 *,
309 file_checksums: dict[str, str],
310 cue_file_checksums: dict[str, set[str]],
311 cur_filenames: set[str],
312 items_to_process: list[tuple[FileSystemItem, str | None]],
313 unchanged_cue_items: list[FileSystemItem],
314 cue_stems: set[str],
315 scan_errors: ScanErrors,
316 metadata_files: list[FileSystemItem],
317 ) -> None:
318 """Walk the WebDAV tree via PROPFIND and populate the sync buckets."""
319 ignore_album_playlists = self.media_content_type == "music" and bool(
320 self.config.get_value(CONF_ENTRY_IGNORE_ALBUM_PLAYLISTS.key)
321 )
322 # mutable counter for the nested coroutine
323 scanned = [0]
324 # guard against directory cycles (e.g. server-side symlink loops) so a single
325 # bad path can never exhaust the recursion limit and abort the whole sync
326 visited: set[str] = set()
327
328 async def _walk(path: str, is_root: bool) -> None:
329 if path in visited:
330 return
331 visited.add(path)
332 try:
333 items = await self._scandir(path)
334 except LoginFailed, SetupFailedError, ProviderUnavailableError:
335 raise
336 except aiohttp.ClientError as err:
337 # a root-level failure aborts the sync right away, subdir failures only
338 # once too many happen in a row, matching the local-filesystem walker
339 if not is_root:
340 self.logger.warning("WebDAV error scanning %s: %s", path, err)
341 scan_errors.record_dir_error(err, is_root=is_root, path=path)
342 return
343 scan_errors.record_dir_read()
344 for item in items:
345 if item.is_dir:
346 await _walk(item.relative_path, is_root=False)
347 if scan_errors.aborted:
348 return
349 continue
350 if item.ext not in WALK_EXTENSIONS:
351 continue
352 scanned[0] += 1
353 if scanned[0] % 500 == 0:
354 update_current_task_progress_text(f"Scanning files: {scanned[0]} found")
355 self._classify_scan_item(
356 item,
357 file_checksums=file_checksums,
358 cue_file_checksums=cue_file_checksums,
359 cur_filenames=cur_filenames,
360 items_to_process=items_to_process,
361 unchanged_cue_items=unchanged_cue_items,
362 cue_stems=cue_stems,
363 ignore_album_playlists=ignore_album_playlists,
364 metadata_files=metadata_files,
365 )
366
367 await _walk("", is_root=True)
368
369 def _get_chapter_path(self, relative_path: str) -> str:
370 """Return authenticated WebDAV URL for a chapter file."""
371 return self._build_authenticated_url(relative_path)
372
373 async def _is_reachable(self) -> bool:
374 """Return whether the WebDAV server can be reached."""
375 # base_path is the server url here, so the parent's directory stat cannot answer this
376 await webdav_test_connection(
377 self._session, self.base_url, self.username, self.password, timeout=10
378 )
379 return True
380