scripts/duplicate_file_report.py — Duplicate File and Photo Report

#!/usr/bin/env python3
"""Find duplicate local files without changing scanned content.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import argparse
import errno
import hashlib
import json
import os
import stat
import sys
import unicodedata
from pathlib import Path
from typing import Any, Callable, Sequence

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
SF_DATALESS = 0x40000000
HASH_CHUNK = 64 * 1024
DIRECTORY_APIS = (os.open in os.supports_dir_fd and os.stat in os.supports_dir_fd
                  and os.scandir in os.supports_fd)
PACKAGE_SUFFIXES = (
    ".app", ".bundle", ".framework", ".library", ".photoslibrary",
    ".pkg", ".plugin", ".scptd", ".xcodeproj",
)


class ScanError(Exception):
    """Expected input or output error."""


class SkillArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise ScanError(message)


def require_platform() -> None:
    if (os.name != "posix" or sys.platform not in ("darwin", "linux") or not hasattr(os, "O_DIRECTORY") or
            not hasattr(os, "O_NOFOLLOW") or not DIRECTORY_APIS):
        raise ScanError("Supported systems: macOS/Linux with POSIX directory descriptors and O_NOFOLLOW; Windows is unsupported")


def validated_path(path: str) -> str:
    if not path or "\0" in path or ".." in path.split("/"):
        raise ScanError("Path must not contain parent traversal (..) or NUL")
    return os.path.abspath(os.path.expanduser(path))


def identity(descriptor: int) -> tuple[int, int]:
    info = os.fstat(descriptor)
    return info.st_dev, info.st_ino


def open_directory(path: str, expected: dict[str, tuple[int, int]] | None = None) -> int:
    """Walk every component without following symlinks, checking pinned identities."""
    descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    current = "/"
    if expected is not None:
        actual = identity(descriptor)
        if "/" in expected and expected["/"] != actual:
            os.close(descriptor)
            raise ScanError("Root directory identity changed")
        expected["/"] = actual
    try:
        for name in Path(path).parts[1:]:
            try:
                child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
            except OSError as exc:
                if exc.errno in {errno.ENOTDIR, errno.ELOOP}:
                    try:
                        component = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
                    except OSError:
                        detail = "is a symlink or not a directory"
                    else:
                        detail = "is a symlink" if stat.S_ISLNK(component.st_mode) else "is not a directory"
                    hint = "; use the physical path (for example /private/tmp)" if "symlink" in detail else ""
                    raise ScanError(f"a path component {detail}{hint}") from exc
                raise
            os.close(descriptor)
            descriptor = child
            current = os.path.join(current, name)
            if expected is not None:
                actual = identity(descriptor)
                if current in expected and expected[current] != actual:
                    raise ScanError("Directory identity changed; refusing access")
                expected[current] = actual
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def output_plan(path: str) -> dict[str, Any]:
    destination = validated_path(path)
    parent = os.path.dirname(destination)
    ancestors: dict[str, tuple[int, int]] = {}
    descriptor = open_directory(parent, ancestors)
    try:
        try:
            os.stat(os.path.basename(destination), dir_fd=descriptor, follow_symlinks=False)
        except FileNotFoundError:
            pass
        else:
            raise ScanError("Report destination already exists; choose a new path")
    finally:
        os.close(descriptor)
    return {"destination": destination, "resolved_destination": str(Path(destination).resolve()),
            "directory": ancestors.get(parent), "parent": ancestors.get(os.path.dirname(parent)),
            "ancestors": ancestors}


def display_text(value: object) -> str:
    """Keep untrusted values on one printable line."""
    pieces = []
    for character in str(value):
        point = ord(character)
        if character in ("\u2028", "\u2029") or unicodedata.category(character).startswith("C"):
            pieces.append(f"\\u{point:04x}" if point <= 0xFFFF else f"\\U{point:08x}")
        else:
            pieces.append(character)
    return "".join(pieces)


def file_hash(path: str, size: int, partial: bool, expected: dict[str, tuple[int, int]] | None = None) -> str:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
    directory = open_directory(os.path.dirname(path), expected)
    try:
        descriptor = os.open(os.path.basename(path), flags, dir_fd=directory)
    finally:
        os.close(directory)
    try:
        with os.fdopen(descriptor, "rb") as stream:
            descriptor = -1
            before = os.fstat(stream.fileno())
            if not stat.S_ISREG(before.st_mode) or getattr(before, "st_flags", 0) & SF_DATALESS:
                raise OSError("file became special or not downloaded")
            if before.st_size != size:
                raise OSError("file changed during scan")
            digest = hashlib.sha256()
            if partial:
                digest.update(stream.read(HASH_CHUNK))
                if size > HASH_CHUNK:
                    stream.seek(size - HASH_CHUNK)
                    digest.update(stream.read(HASH_CHUNK))
            else:
                remaining = size
                while remaining:
                    block = stream.read(min(1024 * 1024, remaining))
                    if not block:
                        raise OSError("file shrank during scan")
                    digest.update(block)
                    remaining -= len(block)
                if stream.read(1):
                    raise OSError("file grew during scan")
            after = os.fstat(stream.fileno())
            if (before.st_size, before.st_mtime_ns, before.st_ino) != (
                after.st_size, after.st_mtime_ns, after.st_ino
            ):
                raise OSError("file changed during scan")
            return digest.hexdigest()
    finally:
        if descriptor >= 0:
            os.close(descriptor)


def scan_paths(
    roots: Sequence[str], min_size: int = 1, include_hidden: bool = False,
    include_packages: bool = False,
    stat_func: Callable[..., os.stat_result] | None = None,
) -> dict[str, Any]:
    require_platform()
    if min_size < 0:
        raise ScanError("--min-size must be nonnegative")
    normalized_roots = list(dict.fromkeys(validated_path(root) for root in roots))
    pinned: dict[str, tuple[int, int]] = {}
    result: dict[str, Any] = {
        "roots": normalized_roots, "min_size": min_size,
        "include_hidden": include_hidden, "include_packages": include_packages,
        "scanned_files": 0, "not_downloaded": 0, "unreadable": 0,
        "skipped": [], "incomplete": False, "groups": [], "reclaimable_bytes": 0,
    }
    by_size: dict[int, list[dict[str, Any]]] = {}
    visited_paths: set[str] = set()

    def skip(path: str, reason: str, incomplete: bool = False) -> None:
        result["skipped"].append({"path": path, "reason": reason})
        if reason == "not_downloaded":
            result["not_downloaded"] += 1
        elif reason == "unreadable":
            result["unreadable"] += 1
        if incomplete:
            result["incomplete"] = True

    stack = list(reversed(normalized_roots))
    while stack:
        path = stack.pop()
        if path in visited_paths:
            continue
        visited_paths.add(path)
        try:
            if stat_func is not None:
                info = stat_func(path, follow_symlinks=False)
            else:
                directory = open_directory(os.path.dirname(path), pinned)
                try:
                    info = os.stat(os.path.basename(path) or ".", dir_fd=directory, follow_symlinks=False)
                finally:
                    os.close(directory)
        except ScanError:
            if path in normalized_roots:
                raise
            skip(path, "unreadable", True)
            continue
        except OSError:
            skip(path, "unreadable", True)
            continue
        if stat.S_ISLNK(info.st_mode):
            skip(path, "symlink")
            continue
        if stat.S_ISDIR(info.st_mode):
            name = os.path.basename(path)
            if not include_hidden and name.startswith("."):
                skip(path, "hidden_directory")
                continue
            if not include_packages and name.lower().endswith(PACKAGE_SUFFIXES):
                skip(path, "package_directory")
                continue
            try:
                directory = open_directory(path, pinned)
                try:
                    with os.scandir(directory) as entries:
                        children = sorted((os.path.join(path, entry.name) for entry in entries), reverse=True)
                finally:
                    os.close(directory)
                stack.extend(children)
            except ScanError:
                if path in normalized_roots:
                    raise
                skip(path, "unreadable", True)
            except OSError:
                skip(path, "unreadable", True)
            continue
        if not stat.S_ISREG(info.st_mode):
            skip(path, "special_file")
            continue
        if not include_hidden and os.path.basename(path).startswith("."):
            skip(path, "hidden_file")
            continue
        if getattr(info, "st_flags", 0) & SF_DATALESS:
            skip(path, "not_downloaded", True)
            continue
        if info.st_size < min_size:
            continue
        result["scanned_files"] += 1
        by_size.setdefault(info.st_size, []).append({
            "path": path, "size": info.st_size, "mtime_ns": info.st_mtime_ns,
        })

    for size, candidates in sorted(by_size.items()):
        if len(candidates) < 2:
            continue
        partial_groups: dict[str, list[dict[str, Any]]] = {}
        for item in candidates:
            try:
                key = file_hash(item["path"], size, True, pinned)
            except (OSError, ScanError):
                skip(item["path"], "unreadable", True)
                continue
            partial_groups.setdefault(key, []).append(item)
        for partial_group in partial_groups.values():
            if len(partial_group) < 2:
                continue
            full_groups: dict[str, list[dict[str, Any]]] = {}
            for item in partial_group:
                try:
                    key = file_hash(item["path"], size, False, pinned)
                except (OSError, ScanError):
                    skip(item["path"], "unreadable", True)
                    continue
                full_groups.setdefault(key, []).append(item)
            for digest, files in full_groups.items():
                if len(files) < 2:
                    continue
                files.sort(key=lambda item: item["path"])
                reclaimed = size * (len(files) - 1)
                result["groups"].append({
                    "size": size, "sha256": digest,
                    "reclaimable_bytes": reclaimed, "files": files,
                })
                result["reclaimable_bytes"] += reclaimed
    result["groups"].sort(key=lambda group: (-group["reclaimable_bytes"], group["files"][0]["path"]))
    result["skipped"].sort(key=lambda item: (item["path"], item["reason"]))
    return result


def report_bytes(result: dict[str, Any]) -> bytes:
    return (json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True) + "\n").encode("utf-8")


def approval_digest(path: str, content: bytes, plan: dict[str, Any] | None = None) -> str:
    plan = output_plan(path) if plan is None else plan
    digest = hashlib.sha256()
    digest.update(b"duplicate-file-and-photo-report-v2\0")
    digest.update(json.dumps(plan, sort_keys=True).encode("utf-8"))
    digest.update(b"\0")
    digest.update(content)
    return digest.hexdigest()


def create_report(path: str, content: bytes, plan: dict[str, Any] | None = None) -> None:
    require_platform()
    plan = output_plan(path) if plan is None else plan
    destination = validated_path(path)
    if destination != plan["destination"]:
        raise ScanError("Report destination changed")
    descriptor = open_directory(os.path.dirname(destination), dict(plan["ancestors"]))
    try:
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW
        output = os.open(os.path.basename(destination), flags, 0o600, dir_fd=descriptor)
        with os.fdopen(output, "wb") as stream:
            stream.write(content)
    finally:
        os.close(descriptor)


def build_parser() -> SkillArgumentParser:
    parser = SkillArgumentParser(
        description="Report duplicate local files without deleting or moving them.",
        epilog=SIGNATURE,
    )
    subcommands = parser.add_subparsers(dest="command", required=True)
    scan = subcommands.add_parser("scan", help="scan local roots", epilog=SIGNATURE)
    scan.add_argument("roots", metavar="ROOT", nargs="+")
    scan.add_argument("--min-size", type=int, default=1, metavar="BYTES",
                      help="minimum file size in bytes (default: 1; zero-byte files are skipped)")
    scan.add_argument("--include-hidden", action="store_true")
    scan.add_argument("--include-packages", action="store_true")
    scan.add_argument("--json", action="store_true")
    scan.add_argument("--report", metavar="PATH")
    scan.add_argument("--dry-run", action="store_true")
    scan.add_argument("--approve", metavar="DIGEST")
    scan.add_argument("--allow-partial", action="store_true")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    try:
        require_platform()
        args = build_parser().parse_args(argv)
        if args.report:
            validated_path(args.report)
        if args.dry_run and not args.report:
            raise ScanError("--dry-run requires --report")
        if args.approve and not args.report:
            raise ScanError("--approve requires --report")
        if args.dry_run and args.approve:
            raise ScanError("--dry-run and --approve cannot be combined")
        result = scan_paths(args.roots, args.min_size, args.include_hidden, args.include_packages)
        content = report_bytes(result)
        if args.report:
            plan = output_plan(args.report)
            digest = approval_digest(args.report, content, plan)
            if not args.dry_run:
                if result["incomplete"] and not args.allow_partial:
                    raise ScanError("incomplete scan; report create requires --allow-partial")
                if not args.approve or args.approve != digest:
                    raise ScanError("report approval missing or different; run --dry-run with the same options")
                create_report(args.report, content, plan)
        if args.json:
            sys.stdout.write(content.decode("utf-8"))
        else:
            print(f"Scanned files: {result['scanned_files']}; duplicate groups: {len(result['groups'])}; reclaimable bytes: {result['reclaimable_bytes']}")
            for group in result["groups"]:
                print(f"Group: {len(group['files'])} copies, {group['size']} bytes each, {group['reclaimable_bytes']} reclaimable bytes")
                for item in group["files"]:
                    print(f"  {display_text(item['path'])} (mtime_ns={item['mtime_ns']})")
            if result["incomplete"]:
                print(f"INCOMPLETE: {result['not_downloaded']} not downloaded; {result['unreadable']} unreadable. Skipped paths:")
                for item in result["skipped"]:
                    if item["reason"] in ("not_downloaded", "unreadable"):
                        print(f"  {display_text(item['path'])}: {item['reason']}")
        if args.dry_run:
            print(f"Report dry run; destination: {display_text(plan['resolved_destination'])}; no directories created; approval digest: {digest}", file=sys.stderr)
        if result["incomplete"] and not args.allow_partial:
            return 2
        return 1 if result["groups"] else 0
    except (ScanError, OSError, ValueError) as exc:
        print(f"Error: {display_text(exc)}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing