music-assistant-server

1"""Some helpers for Filesystem based Musicproviders."""
2
3from __future__ import annotations
4
5import errno
6import hashlib
7import logging
8import os
9import re
10from collections.abc import Iterator
11from dataclasses import dataclass, field
12from pathlib import Path
13
14from music_assistant_models.errors import MediaNotFoundError
15
16from music_assistant.helpers.compare import compare_strings
17from music_assistant.helpers.security import is_safe_path
18
19logger = logging.getLogger(__name__)
20
21# number of consecutive unreadable directories that marks the storage itself as gone
22MAX_CONSECUTIVE_SCAN_ERRORS = 10
23
24# number of example paths kept for the scan summary the user gets to see
25MAX_REPORTED_FAILED_PATHS = 5
26
27IGNORE_DIRS = (
28    "recycle",
29    "Recently-Snaphot",
30    "Recently-Snapshot",
31    "#recycle",
32    "System Volume Information",
33    "lost+found",
34    "@eaDir",
35)
36
37
38@dataclass
39class ScanErrors:
40    """
41    Error state of a single (recursive) scan of a filesystem provider.
42
43    Shared by all directory levels of one scan, so a storage that goes away
44    halfway through is detected after a handful of failures instead of failing
45    once per remaining directory.
46
47    - fatal: The error that ended the scan: the provider root itself is unreadable
48      or too many directories failed in a row. Callers abort the sync and mark
49      the provider unavailable.
50    - failed_dirs: Number of directories that could not be read. A scan with
51      failed directories is incomplete, so callers must not run deletions.
52    - failed_entries: Number of files that could not be read or processed. Those
53      files are missing from the scan result too, so they block deletions as well.
54    - failed_paths: The first few paths that could not be read, named in the summary
55      so the user can find them without turning on debug logging.
56    - consecutive_failures: Directories that failed since the last one read
57      successfully, excluding failures that do not point at unreachable storage.
58    """
59
60    fatal: Exception | None = None
61    failed_dirs: int = 0
62    failed_entries: int = 0
63    consecutive_failures: int = 0
64    failed_paths: list[str] = field(default_factory=list)
65
66    @property
67    def aborted(self) -> bool:
68        """Return True if the scan must be stopped."""
69        return self.fatal is not None
70
71    @property
72    def incomplete(self) -> bool:
73        """Return True if the scan missed content that is still on the storage."""
74        return bool(self.failed_dirs or self.failed_entries)
75
76    def describe(self) -> str:
77        """Return a summary of what this scan could not read. Only meaningful when incomplete."""
78        parts = []
79        if self.failed_dirs:
80            parts.append(f"{self.failed_dirs} folder(s)")
81        if self.failed_entries:
82            parts.append(f"{self.failed_entries} file(s)")
83        summary = f"{' and '.join(parts)} could not be read"
84        if self.failed_paths:
85            summary += f" (e.g. {', '.join(self.failed_paths)})"
86        return summary
87
88    def record_dir_read(self) -> None:
89        """Register a directory that was read successfully."""
90        self.consecutive_failures = 0
91
92    def record_dir_error(
93        self,
94        err: Exception,
95        *,
96        is_root: bool,
97        counts_toward_abort: bool = True,
98        path: str | None = None,
99    ) -> None:
100        """
101        Register a directory that could not be read.
102
103        :param err: The error raised while reading the directory.
104        :param is_root: True if the directory is the provider's root path.
105        :param counts_toward_abort: False for an error that leaves the scan incomplete
106            but says nothing about the storage being reachable, such as a folder that
107            is only permission-denied.
108        :param path: Path of the directory, named in the summary shown to the user.
109        """
110        if is_root:
111            self.fatal = err
112            return
113        self.failed_dirs += 1
114        self._remember_path(path)
115        if not counts_toward_abort:
116            return
117        self.consecutive_failures += 1
118        if self.consecutive_failures >= MAX_CONSECUTIVE_SCAN_ERRORS:
119            self.fatal = err
120
121    def record_entry_error(self, err: Exception, path: str | None = None) -> None:
122        """
123        Register a file that could not be read or processed.
124
125        :param err: The error raised while reading or processing the file.
126        :param path: Path of the file, named in the summary shown to the user.
127        """
128        # a file that disappeared between the listing and the read is a normal race
129        # during a long scan, and it really is gone, so deletions may handle it
130        if getattr(err, "errno", None) == errno.ENOENT:
131            return
132        self.failed_entries += 1
133        self._remember_path(path)
134
135    def _remember_path(self, path: str | None) -> None:
136        """Keep the first few failed paths as examples for the user."""
137        if path and len(self.failed_paths) < MAX_REPORTED_FAILED_PATHS:
138            self.failed_paths.append(path)
139
140
141@dataclass
142class FileSystemItem:
143    """
144    Representation of an item (file or directory) on the filesystem.
145
146    - filename: Name (not path) of the file (or directory).
147    - relative_path: Relative path to the item on this filesystem provider.
148    - absolute_path: Absolute path to this item.
149    - is_dir: Boolean if item is directory (not file).
150    - checksum: Checksum for this path (usually last modified time) None for dir.
151    - file_size : File size in number of bytes or None if unknown (or not a file).
152    - created_at: File creation timestamp (Unix epoch) or None for directories.
153    """
154
155    filename: str
156    relative_path: str
157    absolute_path: str
158    is_dir: bool
159    checksum: str | None = None
160    file_size: int | None = None
161    created_at: int | None = None  # file creation timestamp (Unix epoch)
162
163    @property
164    def ext(self) -> str | None:
165        """Return file extension."""
166        try:
167            # convert to lowercase to make it case insensitive when comparing
168            return self.filename.rsplit(".", 1)[1].lower()
169        except IndexError:
170            return None
171
172    @property
173    def name(self) -> str:
174        """Return file name (without extension)."""
175        return self.filename.rsplit(".", 1)[0]
176
177    @property
178    def parent_name(self) -> str:
179        """Return the name of this item's parent directory."""
180        # derived from the relative path: the absolute path may be a URL on
181        # network/cloud providers (webdav, cloud filesystems)
182        return Path(self.relative_parent_path).name
183
184    @property
185    def relative_parent_path(self) -> str:
186        """Return relative parent path of this item."""
187        return os.path.dirname(self.relative_path)
188
189    @classmethod
190    def from_dir_entry(cls, entry: os.DirEntry[str], base_path: str) -> FileSystemItem:
191        """
192        Create FileSystemItem from os.DirEntry. NOT Async friendly.
193
194        :raises OSError: If the file cannot be stat'd (e.g., invalid filename encoding).
195        """
196        if entry.is_dir(follow_symlinks=False):
197            return cls(
198                filename=entry.name,
199                relative_path=get_relative_path(base_path, entry.path),
200                absolute_path=entry.path,
201                is_dir=True,
202                checksum=None,
203                file_size=None,
204            )
205        # This can raise OSError for files with invalid encoding (e.g., emojis on SMB mounts)
206        # Let the caller handle the exception
207        stat = entry.stat(follow_symlinks=False)
208        # st_birthtime is available on macOS/Windows, st_ctime on Linux
209        # (on Linux st_ctime is metadata change time, not creation time)
210        created_at = int(getattr(stat, "st_birthtime", stat.st_ctime))
211        return cls(
212            filename=entry.name,
213            relative_path=get_relative_path(base_path, entry.path),
214            absolute_path=entry.path,
215            is_dir=False,
216            checksum=str(int(stat.st_mtime)),
217            file_size=stat.st_size,
218            created_at=created_at,
219        )
220
221
222def get_folder_signature(items: list[FileSystemItem]) -> str:
223    """
224    Return an order-independent digest of the given files' paths, mtimes and sizes.
225
226    Intended as a cache checksum: any file added, removed, replaced or retagged changes it.
227
228    :param items: The files to include in the digest.
229    """
230    parts = sorted(f"{x.relative_path}\0{x.checksum}\0{x.file_size}" for x in items)
231    return hashlib.sha256("\0\0".join(parts).encode()).hexdigest()
232
233
234def get_artist_dir(
235    artist_name: str,
236    album_dir: str | None,
237) -> str | None:
238    """Look for (Album)Artist directory in path of a track (or album)."""
239    if not album_dir:
240        return None
241    parentdir = os.path.dirname(album_dir)
242    # account for disc or album sublevel by ignoring (max) 2 levels if needed
243    matched_dir: str | None = None
244    for _ in range(3):
245        dirname = parentdir.rsplit(os.sep)[-1]
246        if compare_strings(artist_name, dirname, False):
247            # literal match
248            # we keep hunting further down to account for the
249            # edge case where the album name has the same name as the artist
250            matched_dir = parentdir
251        parentdir = os.path.dirname(parentdir)
252    return matched_dir
253
254
255def tokenize(input_str: str, delimiters: str) -> list[str]:
256    """Tokenizes the album names or paths."""
257    normalised = re.sub(delimiters, "^^^", input_str)
258    return [x for x in normalised.split("^^^") if x != ""]
259
260
261def _dir_contains_album_name(id3_album_name: str, directory_name: str) -> bool:
262    """
263    Check if a directory name contains an album name.
264
265    This function tokenizes both input strings using different delimiters and
266    checks if the album name is a substring of the directory name.
267
268    First iteration considers the literal dash as one of the separators. The
269    second pass is to catch edge cases where the literal dash is part of the
270    album's name, not an actual separator. For example, an album like 'Aphex
271    Twin - Selected Ambient Works 85-92' would be correctly handled.
272
273    Args:
274        id3_album_name (str): The album name to search for.
275        directory_name (str): The directory name to search in.
276
277    Returns:
278        bool: True if the directory name contains the album name, False otherwise.
279    """
280    for delims in ["[-_ ]", "[_ ]"]:
281        tokenized_album_name = tokenize(id3_album_name, delims)
282        tokenized_dirname = tokenize(directory_name, delims)
283
284        # Exact match, potentially just on the album name
285        # in case artist's name is not included in id3_album_name
286        if all(token in tokenized_dirname for token in tokenized_album_name):
287            return True
288
289        if len(tokenized_album_name) <= len(tokenized_dirname) and compare_strings(
290            "".join(tokenized_album_name),
291            "".join(tokenized_dirname[0 : len(tokenized_album_name)]),
292            False,
293        ):
294            return True
295    return False
296
297
298def get_album_dir(track_dir: str, album_name: str) -> str | None:
299    """Return album/parent directory of a track."""
300    parentdir = track_dir
301    # account for disc sublevel by ignoring 1 level if needed
302    for _ in range(2):
303        dirname = parentdir.rsplit(os.sep)[-1]
304        if compare_strings(album_name, dirname, False):
305            # literal match
306            return parentdir
307        if compare_strings(album_name, dirname.split(" - ")[-1], False):
308            # account for ArtistName - AlbumName format in the directory name
309            return parentdir
310        if compare_strings(album_name, dirname.split(" - ")[-1].split("(")[0], False):
311            # account for ArtistName - AlbumName (Version) format in the directory name
312            return parentdir
313
314        if any(sep in dirname for sep in ["-", " ", "_"]) and album_name:
315            album_chunks = album_name.split(" - ", 1)
316            album_name_includes_artist = len(album_chunks) > 1
317            just_album_name = album_chunks[1] if album_name_includes_artist else None
318
319            # attempt matching using tokenized version of path and album name
320            # with _dir_contains_album_name()
321            if just_album_name and _dir_contains_album_name(just_album_name, dirname):
322                return parentdir
323
324            if _dir_contains_album_name(album_name, dirname):
325                return parentdir
326
327        if compare_strings(album_name.split("(", maxsplit=1)[0], dirname, False):
328            # account for AlbumName (Version) format in the album name
329            return parentdir
330        if compare_strings(album_name.split("(", maxsplit=1)[0], dirname.split(" - ")[-1], False):
331            # account for ArtistName - AlbumName (Version) format
332            return parentdir
333        if len(album_name) > 8 and album_name in dirname:
334            # dirname contains album name
335            # (could potentially lead to false positives, hence the length check)
336            return parentdir
337        parentdir = os.path.dirname(parentdir)
338    return None
339
340
341def get_relative_path(base_path: str, path: str) -> str:
342    """Return the relative path string for a path."""
343    if path.startswith(base_path):
344        path = path.split(base_path)[1]
345    for sep in ("/", "\\"):
346        if path.startswith(sep):
347            path = path[1:]
348    return path
349
350
351def get_absolute_path(base_path: str, path: str) -> str:
352    """
353    Return the absolute path for a path, constrained to base_path.
354
355    :raises MediaNotFoundError: If the resolved path escapes base_path
356        (e.g. via ``../`` traversal or an absolute path outside the base).
357    """
358    absolute_path = path if path.startswith(base_path) else os.path.join(base_path, path)
359    if not is_safe_path(absolute_path, base_path):
360        msg = f"Path is outside the configured base directory: {path}"
361        raise MediaNotFoundError(msg)
362    return absolute_path
363
364
365def recursive_iter(
366    path: str,
367    base_path: str,
368    supported_extensions: set[str],
369    log: logging.Logger,
370    scan_errors: ScanErrors | None = None,
371) -> Iterator[FileSystemItem]:
372    """
373    Recursively traverse directory entries yielding supported files.
374
375    :param path: The directory path to scan.
376    :param base_path: The root base path for constructing relative paths.
377    :param supported_extensions: Set of file extensions to include (lowercase, no dot).
378    :param log: Logger instance to use for warnings/debug messages.
379    :param scan_errors: Optional state object collecting the errors raised during this
380        scan. Callers treat ``fatal`` as "provider unreachable" and abort the sync.
381    """
382    if scan_errors is None:
383        scan_errors = ScanErrors()
384    try:
385        scan_iter = os.scandir(path)
386    except OSError as err:
387        if err.errno == errno.EINVAL:
388            log.warning(
389                "Skipping directory '%s' - unsupported characters in path",
390                path,
391            )
392            return
393        log.warning("Unable to scan directory %s: %s", path, err)
394        _record_dir_failure(scan_errors, err, path=path, base_path=base_path, log=log)
395        return
396    entry_error_logged = False
397    with scan_iter:
398        while True:
399            try:
400                item = next(scan_iter)
401            except StopIteration:
402                scan_errors.record_dir_read()
403                break
404            except OSError as err:
405                log.warning("Error while scanning directory %s: %s", path, err)
406                _record_dir_failure(scan_errors, err, path=path, base_path=base_path, log=log)
407                return
408            if item.name in IGNORE_DIRS or item.name.startswith((".", "_")):
409                continue
410            try:
411                is_dir = item.is_dir(follow_symlinks=False)
412                is_file = item.is_file(follow_symlinks=False)
413            except OSError as err:
414                if err.errno == errno.EINVAL:
415                    log.warning(
416                        "Skipping '%s' - unsupported characters in name",
417                        item.name,
418                    )
419                else:
420                    # the entry may well be a directory, so this can hide a whole subtree
421                    entry_error_logged = _record_entry_failure(
422                        scan_errors,
423                        err,
424                        entry_path=item.path,
425                        base_path=base_path,
426                        log=log,
427                        already_logged=entry_error_logged,
428                    )
429                continue
430            if is_dir:
431                yield from recursive_iter(
432                    item.path,
433                    base_path,
434                    supported_extensions,
435                    log,
436                    scan_errors,
437                )
438                if scan_errors.aborted:
439                    return
440            elif is_file:
441                if "." not in item.name:
442                    continue
443                ext = item.name.rsplit(".", 1)[1].lower()
444                if ext not in supported_extensions:
445                    continue
446                try:
447                    yield FileSystemItem.from_dir_entry(item, base_path)
448                except OSError as err:
449                    if err.errno == errno.EINVAL:
450                        log.warning(
451                            "Skipping '%s' - unsupported characters in name",
452                            item.name,
453                        )
454                    else:
455                        entry_error_logged = _record_entry_failure(
456                            scan_errors,
457                            err,
458                            entry_path=item.path,
459                            base_path=base_path,
460                            log=log,
461                            already_logged=entry_error_logged,
462                        )
463
464
465def sorted_scandir(base_path: str, sub_path: str, sort: bool = False) -> list[FileSystemItem]:
466    """
467    Implement os.scandir that returns (optionally) sorted entries.
468
469    Not async friendly!
470    """
471
472    def nat_key(name: str) -> tuple[int | str, ...]:
473        """Sort key for natural sorting, case insensitive to match the frontend sorting."""
474        return tuple(int(s) if s.isdigit() else s.casefold() for s in re.split(r"(\d+)", name))
475
476    if base_path not in sub_path:
477        sub_path = os.path.join(base_path, sub_path)
478    items: list[FileSystemItem] = []
479    try:
480        entries = os.scandir(sub_path)
481    except OSError as err:
482        if err.errno == errno.EINVAL:
483            logger.warning(
484                "Skipping directory '%s' - unsupported characters in path",
485                sub_path,
486            )
487            return items
488        raise
489    with entries:
490        for entry in entries:
491            try:
492                is_dir = entry.is_dir(follow_symlinks=False)
493                is_file = entry.is_file(follow_symlinks=False)
494            except OSError as err:
495                if err.errno == errno.EINVAL:
496                    logger.warning(
497                        "Skipping '%s' - unsupported characters in name",
498                        entry.name,
499                    )
500                continue
501            if not (is_dir or is_file):
502                continue
503            if entry.name in IGNORE_DIRS or entry.name.startswith("."):
504                continue
505            try:
506                items.append(FileSystemItem.from_dir_entry(entry, base_path))
507            except OSError as err:
508                if err.errno == errno.EINVAL:
509                    logger.warning(
510                        "Skipping '%s' - unsupported characters in name",
511                        entry.name,
512                    )
513                else:
514                    logger.debug("Skipping '%s' due to OS error: %s", entry.name, err)
515                continue
516
517    if sort:
518        return sorted(
519            items,
520            # sort by (natural) name
521            key=lambda x: nat_key(x.name),
522        )
523    return items
524
525
526def _record_entry_failure(
527    scan_errors: ScanErrors,
528    err: OSError,
529    *,
530    entry_path: str,
531    base_path: str,
532    log: logging.Logger,
533    already_logged: bool,
534) -> bool:
535    """Register a directory entry that could not be read and report it once per directory."""
536    # a share that drops mid-listing fails every entry in the directory it was reading,
537    # so only the first one is a warning and the rest are debug to keep the log readable
538    log.log(
539        logging.DEBUG if already_logged else logging.WARNING,
540        "Skipping %s due to OS error: %s",
541        entry_path,
542        err,
543    )
544    scan_errors.record_entry_error(err, get_relative_path(base_path, entry_path))
545    return True
546
547
548def _record_dir_failure(
549    scan_errors: ScanErrors,
550    err: OSError,
551    *,
552    path: str,
553    base_path: str,
554    log: logging.Logger,
555) -> None:
556    """Register a directory that could not be read and report it if the scan gives up."""
557    is_root = path == base_path
558    # a folder we may not read is an ACL problem; the storage itself is still there
559    denied = err.errno in (errno.EACCES, errno.EPERM)
560    scan_errors.record_dir_error(
561        err,
562        is_root=is_root,
563        counts_toward_abort=not denied,
564        path=get_relative_path(base_path, path),
565    )
566    if scan_errors.aborted and not is_root:
567        log.error(
568            "Stopping the scan of %s: %d folders in a row could not be read",
569            base_path,
570            scan_errors.consecutive_failures,
571        )
572