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