diff options
Diffstat (limited to 'duplicate_finder/scanner.py')
| -rw-r--r-- | duplicate_finder/scanner.py | 239 |
1 files changed, 239 insertions, 0 deletions
diff --git a/duplicate_finder/scanner.py b/duplicate_finder/scanner.py new file mode 100644 index 0000000..29f7aef --- /dev/null +++ b/duplicate_finder/scanner.py @@ -0,0 +1,239 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Discover files and group duplicate candidates by metadata and identity.""" + +import sys +from collections import defaultdict +from collections.abc import Iterator +from pathlib import Path + +from tqdm import tqdm + +from .cli import debug, error, is_debug_enabled, is_quiet_enabled +from .types import FileIdentity, FileInfo, FilesByCandidateGroup + + +def should_follow(path: Path, follow_symlinks: bool = False) -> bool: + """Return whether a filesystem entry should be followed. + + Broken symbolic links are skipped and reported when symbolic-link following + is enabled. + + :param path: Filesystem path to inspect. + :param follow_symlinks: Whether symbolic links should be followed. + :returns: ``True`` if the entry should be processed, otherwise ``False``. + """ + if not path.is_symlink(): + return True + + if not follow_symlinks: + return False + + if not path.exists(): + error(f"{path.absolute()} is a broken symbolic link. Skipping.") + return False + + return True + + +def already_visited(path: Path, visited_paths: set[FileIdentity]) -> bool: + """Return whether a filesystem path has already been visited. + + The path is identified by its device and inode so symbolic links to an + already visited directory are recognized as the same filesystem object. + + :param path: Filesystem path to inspect. + :param visited_paths: Filesystem identities that have already been visited. + :returns: ``True`` if the path was already visited, otherwise ``False``. + """ + stat = path.stat() + identity = (stat.st_dev, stat.st_ino) + + if identity in visited_paths: + return True + + visited_paths.add(identity) + return False + + +def iter_files_recursively( + paths: Iterator[Path], + *, + follow_symlinks: bool = False, + visited_paths: set[FileIdentity], +) -> Iterator[FileInfo]: + """Yield regular files and their metadata below the supplied paths. + + Metadata required by later stages is collected when each regular file is + discovered and carried forward with the path so candidate grouping and + hashing do not need to inspect the file again. + + :param paths: File and directory paths to inspect. + :param follow_symlinks: Whether symbolic links to files and directories should + be followed. + :param visited_paths: Filesystem identities of directories already traversed. + :yields: Regular files discovered while scanning the supplied paths, together + with the metadata needed by later stages. + """ + for path in paths: + try: + if not should_follow(path, follow_symlinks): + continue + + if path.is_dir(): + if follow_symlinks and already_visited(path, visited_paths): + debug( + f"Skipping already visited directory: {path.absolute()}", + err=True, + ) + continue + + try: + yield from iter_files_recursively( + path.iterdir(), + follow_symlinks=follow_symlinks, + visited_paths=visited_paths, + ) + except PermissionError: + debug( + f"Skipping inaccessible directory: {path.absolute()}", err=True + ) + continue + elif path.is_file(): + path_stat = path.stat() + yield FileInfo( + path=path, + size=path_stat.st_size, + identity=(path_stat.st_dev, path_stat.st_ino), + ) + else: + error(f"{path} is not a file or directory. Skipping!") + except FileNotFoundError: + debug(f"Skipping disappeared path: {path.absolute()}", err=True) + continue + except PermissionError: + debug(f"Skipping inaccessible path: {path.absolute()}", err=True) + continue + + +def iter_files_from_paths( + paths: Iterator[Path], *, follow_symlinks: bool = False +) -> Iterator[FileInfo]: + """Yield regular files and their metadata below the supplied paths. + + Directories are traversed recursively in depth-first order. Permission + errors are ignored so scanning can continue with the remaining paths. When + symbolic links are followed, directories already visited through another + path are skipped to avoid recursion cycles. + + :param paths: File and directory paths to inspect. + :param follow_symlinks: Whether symbolic links to files and directories + should be followed. + :yields: Regular files discovered while scanning the supplied paths, together + with the metadata needed by later stages. + """ + visited_paths: set[FileIdentity] = set() + + yield from iter_files_recursively( + paths, follow_symlinks=follow_symlinks, visited_paths=visited_paths + ) + + +def group_file_candidates( + files: Iterator[FileInfo], + *, + minsize: int | None = None, + maxsize: int | None = None, + include_empty: bool = True, + include_file_extension: bool = True, + include_hardlinks: bool = False, +) -> FilesByCandidateGroup: + """Group files into candidate sets for duplicate detection. + + Files are grouped by size and, by default, case-insensitive file extension. + Within each candidate group, files are grouped by filesystem identity so + hard links can share a single hash computation. Groups that cannot contain + duplicates are removed before hashing. + + :param files: Files and metadata collected during discovery. + :param minsize: Minimum file size in bytes, or ``None`` for no lower limit. + :param maxsize: Maximum file size in bytes, or ``None`` for no upper limit. + :param include_empty: Whether zero-byte files should be included. + :param include_file_extension: Whether file extensions should be included + in the candidate-group key in addition to file size. + :param include_hardlinks: Whether different paths referring to the same inode + should count as separate duplicate candidates. + :returns: Files grouped into duplicate-candidate sets and then by filesystem + identity. Every returned candidate set can contain at least one duplicate. + """ + files_by_candidate_group = defaultdict(lambda: defaultdict(set)) + accepted = 0 + filtered_by_size = 0 + + for file in tqdm( + files, + desc="Scanning files", + unit="file", + dynamic_ncols=True, + disable=is_quiet_enabled() or not sys.stderr.isatty(), + ): + if ( + (not include_empty and file.size == 0) + or (minsize is not None and file.size < minsize) + or (maxsize is not None and file.size > maxsize) + ): + filtered_by_size += 1 + continue + + group_key = ( + (file.size, file.path.suffix.casefold()) + if include_file_extension + else file.size + ) + files_by_candidate_group[group_key][file.identity].add(file) + accepted += 1 + + candidate_groups: FilesByCandidateGroup = {} + candidate_paths = 0 + + for group_key, files_by_identity in files_by_candidate_group.items(): + candidate_count = ( + sum(len(files) for files in files_by_identity.values()) + if include_hardlinks + else len(files_by_identity) + ) + if candidate_count < 2: + continue + + candidate_groups[group_key] = files_by_identity + candidate_paths += sum(len(files) for files in files_by_identity.values()) + + debug( + "File scan summary:", + f"{accepted} accepted, {filtered_by_size} filtered by size; " + f"{candidate_paths} paths remain in {len(candidate_groups)} candidate groups.", + err=True, + ) + + if is_debug_enabled(): + hardlinked_identity_count = 0 + hardlinked_path_count = 0 + + for files_by_identity in candidate_groups.values(): + for files in files_by_identity.values(): + if len(files) < 2: + continue + hardlinked_identity_count += 1 + hardlinked_path_count += len(files) + + debug( + "Hardlink summary:", + f"{hardlinked_path_count} paths across " + f"{hardlinked_identity_count} multiply referenced identities " + "in candidate groups.", + err=True, + ) + + return candidate_groups |