music-assistant-server

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