From b751174b4ed7c23786ae54a1bbaf332f653faa81 Mon Sep 17 00:00:00 2001 From: Dennis Fink Date: Sun, 20 Sep 2026 20:04:13 +0200 Subject: feat: release duplicate-finder 1.0.0 Introduce the first public release of duplicate-finder, a command-line tool for detecting duplicate files through metadata grouping and content hashing. Support recursive path scanning, optional symlink traversal, hard-link handling, size and extension-based candidate grouping, and parallel content hashing with xxHash or hashlib algorithms. Provide human, plain, table, and JSON output formats together with size filtering, path input from files or stdin, progress reporting, diagnostic output, and configurable terminal colors. Add Python packaging for Python 3.14, project documentation, dependency locking, development tooling, and BSD-3-Clause licensing. --- duplicate_finder/__init__.py | 330 +++++++++++++++++++++++++++++++++++++++++++ duplicate_finder/__main__.py | 9 ++ duplicate_finder/cli.py | 234 ++++++++++++++++++++++++++++++ duplicate_finder/hash.py | 174 +++++++++++++++++++++++ duplicate_finder/output.py | 136 ++++++++++++++++++ duplicate_finder/scanner.py | 239 +++++++++++++++++++++++++++++++ duplicate_finder/types.py | 36 +++++ 7 files changed, 1158 insertions(+) create mode 100644 duplicate_finder/__init__.py create mode 100644 duplicate_finder/__main__.py create mode 100644 duplicate_finder/cli.py create mode 100644 duplicate_finder/hash.py create mode 100644 duplicate_finder/output.py create mode 100644 duplicate_finder/scanner.py create mode 100644 duplicate_finder/types.py (limited to 'duplicate_finder') 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 +# +# 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 +############################################################################### + +"""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 " +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 . + +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 +# +# 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 +# +# 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\d+\.?\d*)\s*((?P[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 +# +# 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 +# +# 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 +# +# 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 +# +# 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] -- cgit v1.3.1