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