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