# SPDX-FileCopyrightText: 2026 Dennis Fink # # 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