diff options
Diffstat (limited to 'duplicate_finder')
| -rw-r--r-- | duplicate_finder/__init__.py | 330 | ||||
| -rw-r--r-- | duplicate_finder/__main__.py | 9 | ||||
| -rw-r--r-- | duplicate_finder/cli.py | 234 | ||||
| -rw-r--r-- | duplicate_finder/hash.py | 174 | ||||
| -rw-r--r-- | duplicate_finder/output.py | 136 | ||||
| -rw-r--r-- | duplicate_finder/scanner.py | 239 | ||||
| -rw-r--r-- | duplicate_finder/types.py | 36 |
7 files changed, 1158 insertions, 0 deletions
diff --git a/duplicate_finder/__init__.py b/duplicate_finder/__init__.py new file mode 100644 index 0000000..b99c0ca --- /dev/null +++ b/duplicate_finder/__init__.py @@ -0,0 +1,330 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +############################################################################### +# DESCRIPTION +# This script scans one or more paths and identifies duplicate files. +# Candidates are grouped by size and file extension before their contents +# are hashed. The output can be rendered as human-readable text, plain text, +# a table, or structured JSON. +# +# NOTES +# This script respects the NO_COLOR standard (https://no-color.org/). +# +# AUTHOR +# Dennis Fink <me+coding@dennisfink.me> +############################################################################### + +"""Provide the command-line interface for finding duplicate files. + +Files are discovered recursively and grouped into candidate sets by size and, +by default, file extension. Paths in each candidate set are also grouped by +filesystem identity so hard links can be handled efficiently. Candidate files +are then hashed to identify matching content. +""" + +import hashlib +import os +from pathlib import Path +from typing import TextIO + +import click_extra as click + +from . import output +from .cli import ByteSizeParamType, FlexibleColorOption, debug, msg +from .hash import group_files_by_hash +from .scanner import group_file_candidates, iter_files_from_paths + +VERSION = "1.0.0" +DESCRIPTION = "Find duplicate files using metadata grouping and content hashes." +DATE_OF_CREATION = "2017-04-23" +DATE_OF_REVISION = "2026-09-20" +AUTHOR = "Dennis Fink <me+coding@dennisfink.me>" +LICENSE = "BSD-3-Clause" + + +def print_version( + ctx: click.Context, param: click.Parameter | None, value: bool +) -> None: + """Print script metadata and exit. + + :param ctx: Click context associated with the currently running command. + :param param: Click parameter that triggered the callback, if available. + :param value: Whether the version option was supplied. + """ + if not value or ctx.resilient_parsing: + return + + click.echo(click.style("Scriptname:", fg="red", bold=True) + f" {ctx.info_name}") + click.echo(click.style("Version:", fg="green", bold=True) + f" {VERSION}") + click.echo(click.style("Description:", fg="yellow", bold=True) + f" {DESCRIPTION}") + click.echo(click.style("Author:", fg="blue", bold=True) + f" {AUTHOR}") + click.echo( + click.style("Date of creation:", fg="magenta", bold=True) + + f" {DATE_OF_CREATION}" + ) + click.echo( + click.style("Date of revision:", fg="cyan", bold=True) + f" {DATE_OF_REVISION}" + ) + click.echo(click.style("License:", fg="red", bold=True) + f" {LICENSE}") + + click.echo("""Copyright (c) 2026 Dennis Fink <me+coding@dennisfink.me>. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this +list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. Neither the name of the copyright holder nor the names of its contributors +may be used to endorse or promote products derived from this software without +specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\" AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.""") + + ctx.exit() + + +@click.command( + context_settings={"help_option_names": ("-h", "--help", "-?")}, params=[] +) +@click.option( + "--read-from", + type=click.File("r", encoding="utf-8"), + metavar="FILE", + help=( + "Read files and directories to scan from FILE, one path per line. " + "Use '-' for standard input. When set, positional paths are ignored." + ), +) +@click.option( + "--follow-symlinks/--no-follow-symlinks", + "follow_symlinks_flag", + default=False, + help="Follow symlinks when scanning directories.", +) +@click.option( + "-H", + "--include-hardlinks/--exclude-hardlinks", + "hardlinks_flag", + default=False, + help=( + "Include hard links in duplicate results, so multiple paths to the same " + "underlying file may be listed as duplicates." + ), +) +@click.option( + "-G", + "--minsize", + type=ByteSizeParamType(), + help="Consider only files >= SIZE bytes (supports suffixes like 10M, 1G, 500K).", +) +@click.option( + "-L", + "--maxsize", + type=ByteSizeParamType(), + help="Consider only files <= SIZE bytes (supports suffixes like 10M, 1G, 500K).", +) +@click.option( + "--include-empty/--exclude-empty", + "include_empty_flag", + default=False, + help="Include files that are 0 bytes in size.", +) +@click.option( + "--include-file-extension/--exclude-file-extension", + "include_file_extension_flag", + default=True, + help=( + "Include file extensions when grouping duplicate candidates. " + "Enabled by default; excluding extensions groups candidates by size only." + ), +) +@click.option( + "--hash", + "hash_func_name", + default="xxh128", + help="Select which hash function to use. Default xxh128.", + type=click.Choice( + sorted(["xxh32", "xxh64", "xxh128"] + list(hashlib.algorithms_available)) + ), +) +@click.option( + "-j", + "--jobs", + default=0, + help="Maximum number of worker threads (0 = use default).", + type=click.IntRange(min=0), +) +@click.option( + "-o", + "--output-format", + default="human", + help=( + "Select the output format for duplicate groups. " + "'human' shows a readable text list, " + "'plain' prints each duplicate group as a space-separated line, " + "'table' prints a column-aligned overview, " + "and 'json' outputs machine-readable structured data." + ), + type=click.Choice(output.OUTPUTS.keys(), case_sensitive=False), +) +@click.option( + "-S", + "--size", + "show_size_flag", + is_flag=True, + default=False, + help="Show size of duplicate files (human/plain/table output only).", +) +@click.option( + "--human-readable/--no-human-readable", "human_readable_flag", default=False +) +@click.option( + "-q", + "--quiet", + "quiet_flag", + is_flag=True, + default=False, + help="Suppress all non-error output.", +) +@click.option( + "-v", + "--verbose", + "verbose_flag", + is_flag=True, + default=False, + help="Enable verbose output.", +) +@click.option("--color", cls=FlexibleColorOption) +@click.no_color_option() +@click.option( + "--version", + callback=print_version, + default=False, + expose_value=False, + help="Show the version and exit.", + is_eager=True, + is_flag=True, +) +@click.argument( + "paths", + type=click.Path( + exists=True, file_okay=True, dir_okay=True, readable=True, path_type=Path + ), + nargs=-1, +) +@click.pass_context +def duplicate_finder( + ctx: click.Context, + paths: tuple[Path] | None = None, + read_from: TextIO | None = None, + follow_symlinks_flag: bool = False, + hardlinks_flag: bool = False, + minsize: int | None = None, + maxsize: int | None = None, + include_empty_flag: bool = False, + include_file_extension_flag: bool = True, + hash_func_name: str = "xxh128", + jobs: int = 0, + output_format: str = "human", + show_size_flag: bool = False, + human_readable_flag: bool = False, + quiet_flag: bool = False, + verbose_flag: bool = False, +) -> None: + """Scan paths and list duplicate files. + + Files are first grouped into candidate sets by size and, by default, + case-insensitive file extension. ``--exclude-file-extension`` groups + candidates by size only. Candidate files are then hashed to identify + matching content. + """ + ctx.ensure_object(dict) + ctx.obj["debug"] = os.environ.get("DEBUG", "").lower() in ("1", "true", "yes") + ctx.obj["quiet"] = quiet_flag + ctx.obj["verbose"] = verbose_flag + + debug("Effective configuration:", err=True) + debug(" paths:", ", ".join(str(p) for p in paths) if paths else ".", err=True) + debug( + " read_from:", + str(read_from.name) if read_from is not None else str(read_from), + err=True, + ) + debug(" follow_symlinks:", str(follow_symlinks_flag), err=True) + debug(" hardlinks:", str(hardlinks_flag), err=True) + debug(" minsize:", str(minsize), err=True) + debug(" maxsize:", str(maxsize), err=True) + debug(" include_empty:", str(include_empty_flag), err=True) + debug(" include_file_extension:", str(include_file_extension_flag), err=True) + debug(" hash:", str(hash_func_name), err=True) + debug(" jobs:", "default" if jobs == 0 else str(jobs), err=True) + debug(" output_format:", str(output_format), err=True) + debug(" size:", str(show_size_flag), err=True) + debug(" human_readable:", str(human_readable_flag), err=True) + debug(" quiet:", str(quiet_flag), err=True) + debug(" verbose:", str(verbose_flag), err=True) + + if read_from is not None: + input_paths = ( + Path(path) for line in read_from if (path := line.rstrip("\r\n")) + ) + else: + input_paths = iter(paths or (Path("."),)) + + candidate_files = group_file_candidates( + iter_files_from_paths(input_paths, follow_symlinks=follow_symlinks_flag), + minsize=minsize, + maxsize=maxsize, + include_empty=include_empty_flag, + include_file_extension=include_file_extension_flag, + include_hardlinks=hardlinks_flag, + ) + debug( + "Candidate grouping complete:", + f"{len(candidate_files)} candidate groups remain for hashing.", + err=True, + ) + + hash_groups = group_files_by_hash( + candidate_files, + include_hardlinks=hardlinks_flag, + hash_name=hash_func_name, + jobs=jobs, + ) + + if ctx.obj["debug"]: + duplicate_paths = sum(len(files) for files in hash_groups.values()) + debug( + "Duplicate detection complete:", + f"{len(hash_groups)} groups containing {duplicate_paths} paths.", + err=True, + ) + + if not hash_groups and output_format != "json": + msg("No duplicates found!", err=True) + ctx.exit(2) + + debug(f"Rendering results using {output_format!r} output.", err=True) + output.OUTPUTS[output_format](hash_groups, show_size_flag, human_readable_flag) + + ctx.exit() + + +if __name__ == "__main__": + duplicate_finder() diff --git a/duplicate_finder/__main__.py b/duplicate_finder/__main__.py new file mode 100644 index 0000000..bb3027a --- /dev/null +++ b/duplicate_finder/__main__.py @@ -0,0 +1,9 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Run the duplicate-file finder when the package is executed as a module.""" + +from . import duplicate_finder + +duplicate_finder() diff --git a/duplicate_finder/cli.py b/duplicate_finder/cli.py new file mode 100644 index 0000000..e1bc4f9 --- /dev/null +++ b/duplicate_finder/cli.py @@ -0,0 +1,234 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Define Click parameter types and styled terminal message helpers.""" + +import re +from collections.abc import Callable +from functools import update_wrapper +from typing import Any, Concatenate + +import click_extra as click +from click_extra.color import ColorOption + +_SIZE_RE = re.compile( + r"^\s*(?P<value>\d+\.?\d*)\s*((?P<unit>[kmgtpe]i?)?)b?\s*$", re.IGNORECASE +) +_SIZE_MULTIPLIERS = { + "": 1, + "k": 1000, + "ki": 1024, + "m": 1000**2, + "mi": 1024**2, + "g": 1000**3, + "gi": 1024**3, + "t": 1000**4, + "ti": 1024**4, + "p": 1000**5, + "pi": 1024**5, + "e": 1000**6, + "ei": 1024**6, +} + + +class ByteSizeParamType(click.ParamType): + """Parse byte sizes with optional decimal or binary unit suffixes.""" + + name = "size" + + def convert( + self, value: Any, param: click.Parameter | None, ctx: click.Context | None + ) -> int: + """Convert a Click parameter value to a size in bytes. + + :param value: Value supplied by Click. + :param param: Parameter currently being converted, if available. + :param ctx: Active Click context, if available. + :returns: The parsed size in bytes. + :raises click.BadParameter: If the value is negative, has an unsupported type, + or cannot be parsed as a byte size. + """ + if isinstance(value, int): + if value < 0: + self.fail("Size must be non-negative.", param, ctx) + return value + + if not isinstance(value, str): + self.fail("Size must be a string like '10M' or an integer.", param, ctx) + + if (match := _SIZE_RE.match(value)) is not None: + groups = match.groupdict() + return round( + float(groups["value"]) * _SIZE_MULTIPLIERS[groups["unit"].lower()] + ) + else: + self.fail(f"Invalid size: {value!r}", param, ctx) + + +class FlexibleColorOption(ColorOption): + """Allow the color option value to be passed with or without ``=``.""" + + _gnu_optional_value = False + + +def pass_obj_silent[T, **P, R]( + f: Callable[Concatenate[T | None, P], R], +) -> Callable[P, R]: + """Pass the current context object to a callback, if available. + + The current Click context is retrieved silently. If no context exists or + its object is not a dictionary, ``None`` is passed instead. + + The context object is inserted as the first positional argument to *f*. + """ + + def new_func(*args: P.args, **kwargs: P.kwargs) -> R: + ctx = click.get_current_context(silent=True) + obj = ctx.obj if ctx is not None else None + return f(obj, *args, **kwargs) + + return update_wrapper(new_func, f) + + +def emit( + prefix: str, color: str, *message: str, enabled: bool = True, err: bool = False +) -> None: + """Render a styled prefix and message to stdout. + + This helper is used by :func:`error`, :func:`msg`, :func:`warn`, + :func:`verbose`, and :func:`debug`. The first message argument is rendered in + bold and subsequent arguments are appended unstyled. + + :param prefix: Prefix displayed before the message. + :param color: Click-compatible color name applied to the prefix. + :param message: One or more message parts to display. + :param enabled: Whether the message should be emitted. + :param err: Write to `stderr` instead of `stdout`. + :raises TypeError: If no message arguments are provided. + """ + if not message: + raise TypeError("emit() missing 1 required positional argument: 'message'") + + if not enabled: + return + + click.echo( + " ".join( + [ + click.style(prefix, fg=color, bold=True), + click.style(message[0], bold=True), + *message[1:], + ] + ), + err=err, + ) + + +def error(*message: str) -> None: + """Print a formatted error message to stdout in red. + + The first argument is rendered in bold and subsequent arguments are appended + unstyled. Error messages are always printed regardless of quiet or verbose + flags. + + :param message: One or more message parts to display. + """ + emit("==> ERROR:", "red", *message, err=True) + + +@pass_obj_silent +def msg(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: + """Print a formatted informational message to stdout in green. + + The message is suppressed when the quiet flag is set. When called outside a + Click context, the message is emitted normally. The first argument is rendered + in bold and subsequent arguments are appended unstyled. + + :param message: One or more message parts to display. + """ + emit( + "==>", + "green", + *message, + enabled=obj is None or not obj.get("quiet", False), + err=err, + ) + + +@pass_obj_silent +def warn(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: + """Print a formatted warning message to stdout in yellow. + + The message is suppressed when the quiet flag is set. When called outside a + Click context, the message is emitted normally. The first argument is rendered + in bold and subsequent arguments are appended unstyled. + + :param message: One or more message parts to display. + """ + emit( + "==>", + "yellow", + *message, + enabled=obj is None or not obj.get("quiet", False), + err=err, + ) + + +@pass_obj_silent +def verbose(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: + """Print a formatted verbose message to stdout in blue. + + The message is printed only when called from a Click context with the verbose + flag set and the quiet flag unset. Calls made outside a Click context are a + no-op. + + :param message: One or more message parts to display. + """ + emit( + "==>", + "blue", + *message, + enabled=( + obj is not None + and obj.get("verbose", False) + and not obj.get("quiet", False) + ), + err=err, + ) + + +@pass_obj_silent +def debug(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None: + """Print a formatted debug message to stdout in magenta. + + The message is printed only when called from a Click context with the debug + flag enabled. Calls made outside a Click context are a no-op. + + :param message: One or more message parts to display. + """ + emit( + "==>", + "magenta", + *message, + enabled=obj is not None and obj.get("debug", False), + err=err, + ) + + +@pass_obj_silent +def is_debug_enabled(obj: dict[str, Any] | None) -> bool: + """Return whether debug output is enabled for the active Click context. + + :returns: ``True`` when a Click context exists and its debug flag is enabled. + """ + return obj is not None and obj.get("debug", False) + + +@pass_obj_silent +def is_quiet_enabled(obj: dict[str, Any] | None) -> bool: + """Return whether quiet output is enabled for the active Click context. + + :returns: ``True`` when a Click context exists and its quiet flag is enabled. + """ + return obj is not None and obj.get("quiet", False) diff --git a/duplicate_finder/hash.py b/duplicate_finder/hash.py new file mode 100644 index 0000000..473ee98 --- /dev/null +++ b/duplicate_finder/hash.py @@ -0,0 +1,174 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Hash candidate files and group paths with identical content.""" + +import concurrent.futures +import hashlib +import sys +from collections import defaultdict +from collections.abc import Callable +from functools import partial +from pathlib import Path +from typing import Any + +import xxhash +from tqdm import tqdm + +from duplicate_finder.cli import debug, is_debug_enabled, is_quiet_enabled + +from .types import FileInfo, FilesByCandidateGroup, FilesByHash, FileSet + + +def compute_hash(path: Path, digest: str | Callable[[], Any]) -> str: + """Compute a content hash using the supplied digest. + + The file is opened in binary mode and streamed through + :func:`hashlib.file_digest`. For extendable-output algorithms whose + ``hexdigest`` method requires an explicit output length, a 32-byte digest + is requested. + + :param path: Path to the file whose contents should be hashed. + :param digest: Hash algorithm name or callable that creates a hash object. + :returns: The hexadecimal digest of the file contents. + """ + with path.open("rb") as f: + digested = hashlib.file_digest(f, digest) + + try: + return digested.hexdigest() + except TypeError: + return digested.hexdigest(length=32) + + +def select_representative_files( + files_by_candidate_group: FilesByCandidateGroup, *, include_hardlinks: bool +) -> dict[FileInfo, FileSet]: + """Select one file to hash for each filesystem identity. + + Candidate groups have already been reduced to groups that can contain + duplicates. Each filesystem identity contributes one representative file. + When hard links are included, that representative stands for every scanned + path belonging to the identity; otherwise it represents only itself. + + :param files_by_candidate_group: Files grouped into duplicate-candidate sets + and then by filesystem identity. + :param include_hardlinks: Whether different paths referring to the same inode + should be treated as duplicate files. + :returns: Representative files mapped to the paths they represent. + """ + representative_files: dict[FileInfo, FileSet] = {} + + for files_by_identity in files_by_candidate_group.values(): + for files in files_by_identity.values(): + representative = next(iter(files)) + representative_files[representative] = ( + files if include_hardlinks else {representative} + ) + + return representative_files + + +def select_hash_function(name: str) -> Callable[[Path], str]: + """Return the file-hashing callable for a named digest algorithm. + + :param name: Name of an xxHash variant or an algorithm available through + :mod:`hashlib`. + :returns: A callable that accepts a file path and returns its hexadecimal digest. + :raises KeyError: If ``name`` is not a supported hash algorithm. + """ + hash_funcs = { + "xxh32": partial(compute_hash, digest=xxhash.xxh32), + "xxh64": partial(compute_hash, digest=xxhash.xxh64), + "xxh128": partial(compute_hash, digest=xxhash.xxh128), + } + for digest in hashlib.algorithms_available: + hash_funcs[digest] = partial(compute_hash, digest=digest) + + return hash_funcs[name] + + +def group_files_by_hash( + files_by_candidate_group: FilesByCandidateGroup, + *, + include_hardlinks: bool = False, + hash_name: str = "xxh128", + jobs: int = 0, +) -> FilesByHash: + """Group duplicate candidates by content hash. + + Only one representative of each filesystem identity is hashed. Hash + computation is parallelized using :class:`ThreadPoolExecutor`, and hash + groups containing only one represented path are discarded before returning. + + :param files_by_candidate_group: Files grouped into duplicate-candidate sets + and then by filesystem identity. Candidate sets are normally based on + file size and extension, or file size alone when extension grouping is + disabled. + :param include_hardlinks: Whether multiple paths referring to the same inode + should be included as separate duplicate paths. + :param hash_name: Name of the hash algorithm to use. This may be an xxHash + variant or any algorithm exposed by :mod:`hashlib`. + :param jobs: Maximum number of worker threads to use for hash computation. + If set to ``0``, the default chosen by :class:`ThreadPoolExecutor` is used. + :returns: A mapping from digest strings to sets of files sharing the same + content hash. Singleton hash groups are omitted. + """ + representative_files = select_representative_files( + files_by_candidate_group, include_hardlinks=include_hardlinks + ) + + if is_debug_enabled(): + represented_paths = sum(len(files) for files in representative_files.values()) + debug( + "Hash candidate selection:", + f"{len(representative_files)} representative files for " + f"{represented_paths} paths.", + err=True, + ) + + if not representative_files: + return {} + + hash_function = select_hash_function(hash_name) + max_workers = None if jobs == 0 else jobs + hash_groups: defaultdict[str, FileSet] = defaultdict(set) + + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_file = { + executor.submit(hash_function, file.path): file + for file in representative_files + } + for future in tqdm( + concurrent.futures.as_completed(future_to_file), + total=len(future_to_file), + desc="Hashing files", + unit="file", + dynamic_ncols=True, + disable=is_quiet_enabled() or not sys.stderr.isatty(), + ): + file = future_to_file[future] + try: + hashsum = future.result() + except FileNotFoundError: + debug(f"Skipping disappeared file: {file.path.absolute()}", err=True) + continue + except PermissionError: + debug(f"Skipping inaccessible file: {file.path.absolute()}", err=True) + continue + + hash_groups[hashsum].update(representative_files[file]) + + duplicate_groups = { + hashsum: files for hashsum, files in hash_groups.items() if len(files) > 1 + } + + debug( + "Hashing complete:", + f"{len(representative_files)} files produced " + f"{len(duplicate_groups)} duplicate hash groups.", + err=True, + ) + + return duplicate_groups diff --git a/duplicate_finder/output.py b/duplicate_finder/output.py new file mode 100644 index 0000000..c3b6307 --- /dev/null +++ b/duplicate_finder/output.py @@ -0,0 +1,136 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Render duplicate-file groups in terminal-friendly output formats.""" + +import shlex +from json import dumps as json_dumps + +import click_extra as click +import humanize +from tabulate import tabulate + +from .types import FilesByHash + +OUTPUTS = {} + + +def register_output(f): + OUTPUTS[f.__name__] = f + return f + + +@register_output +def human( + duplicates: FilesByHash, show_size: bool = False, human_readable: bool = False +) -> None: + """Print duplicate groups in a human-readable format. + + :param duplicates: Hash digests mapped to files sharing each digest. + :param show_size: Whether to display the size of each duplicate group. + :param human_readable: Whether displayed sizes should use human-readable units. + """ + for hash_digest, files in duplicates.items(): + click.secho("Duplicate set ", fg="green", bold=True, nl=False) + click.secho("[", fg="magenta", bold=True, nl=False) + click.secho(hash_digest, fg="cyan", bold=True, nl=False) + click.secho("]", fg="magenta", bold=True, nl=False) + + if show_size: + size = next(iter(files)).size + click.secho(" [", fg="magenta", bold=True, nl=False) + click.secho( + humanize.naturalsize(size) if human_readable else size, + fg="cyan", + bold=True, + nl=False, + ) + click.secho("]", fg="magenta", bold=True, nl=False) + + click.secho(":", fg="green", bold=True) + + for file in files: + click.echo(f" {file.path}") + + +@register_output +def plain( + duplicates: FilesByHash, show_size: bool = False, human_readable: bool = False +) -> None: + """Print duplicate groups as shell-quoted, space-separated fields. + + Each duplicate group is written on a separate line. The hash digest is the + first field, followed by the duplicate paths. If requested, the file size is + appended as the final field. + + :param duplicates: Hash digests mapped to files sharing each digest. + :param show_size: Whether to append the size of each duplicate group. + :param human_readable: Whether displayed sizes should use human-readable units. + """ + for hash_digest, files in duplicates.items(): + fields = [hash_digest] + fields.extend(str(file.path) for file in files) + + if show_size: + size = next(iter(files)).size + fields.append(humanize.naturalsize(size) if human_readable else str(size)) + + click.echo(shlex.join(fields)) + + +@register_output +def table( + duplicates: FilesByHash, show_size: bool = False, human_readable: bool = False +) -> None: + """Print duplicate groups as a column-aligned table. + + :param duplicates: Hash digests mapped to files sharing each digest. + :param show_size: Whether to display the size of each duplicate group. + :param human_readable: Whether displayed sizes should use human-readable units. + """ + rows = [] + for hash_digest, files in duplicates.items(): + for file in files: + if show_size: + rows.append( + ( + click.style(hash_digest, fg="cyan"), + file.path, + humanize.naturalsize(file.size) + if human_readable + else file.size, + ) + ) + else: + rows.append((click.style(hash_digest, fg="cyan"), file.path)) + + click.echo(tabulate(rows, tablefmt="plain")) + + +@register_output +def json( + duplicates: FilesByHash, show_size: bool = False, human_readable: bool = False +) -> None: + """Print duplicate groups as JSON. + + File paths are serialized as absolute strings. ``show_size`` is accepted for + the common output-function interface but does not change the JSON structure. + + :param duplicates: Hash digests mapped to files sharing each digest. + :param show_size: Unused; accepted for consistency with other output formats. + :param human_readable: Whether the JSON should be pretty-printed. + """ + json_config = {"separators": (",", ":"), "indent": 0, "sort_keys": False} + + if human_readable: + json_config["separators"] = (",", ": ") + json_config["indent"] = 2 + json_config["sort_keys"] = True + + serialized_duplicates = { + hash_digest: [str(file.path.absolute()) for file in files] + for hash_digest, files in duplicates.items() + } + + click.echo(json_dumps(serialized_duplicates, **json_config)) 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 diff --git a/duplicate_finder/types.py b/duplicate_finder/types.py new file mode 100644 index 0000000..f2438d8 --- /dev/null +++ b/duplicate_finder/types.py @@ -0,0 +1,36 @@ +# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me> +# +# SPDX-License-Identifier: BSD-3-Clause + +"""Types used throughout the duplicate-file finder.""" + +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path + +type FileIdentity = tuple[int, int] +type FileSize = int +type FileHash = str +type FileExtension = str + + +@dataclass(frozen=True, slots=True) +class FileInfo: + """Store filesystem metadata collected while discovering a file. + + :param path: Path through which the file was discovered. + :param size: File size in bytes. + :param identity: Filesystem identity as ``(device, inode)``. + """ + + path: Path + size: FileSize + identity: FileIdentity + + +type FileSet = set[FileInfo] +type FilesByIdentity = defaultdict[FileIdentity, FileSet] + +type CandidateGroupKey = FileSize | tuple[FileSize, FileExtension] +type FilesByCandidateGroup = dict[CandidateGroupKey, FilesByIdentity] +type FilesByHash = dict[FileHash, FileSet] |