path: root/duplicate_finder/__init__.py

# 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()