File Deduplicator (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: File Deduplicator

Write a program that finds and manages duplicate files across directories using content-based hashing, supporting multiple deduplication strategies and detailed reporting.

FUNCTIONAL REQUIREMENTS:
- Accept one or more directory paths as command-line arguments
- Find duplicate files by comparing SHA-256 content hashes, using a two-phase approach: first compare file sizes to narrow candidates, then hash only size-matched files
- Support configurable minimum file size via --min-size flag (default: 1 byte) to skip tiny files
- Support file type filtering via --include and --exclude flags with glob patterns
- Group duplicates into sets showing all copies with their full paths, sizes, and modification dates
- Support multiple deduplication actions via --action flag: report (default, just list duplicates), delete (remove duplicates keeping the oldest/newest based on --keep flag), hardlink (replace duplicates with hard links to save space), symlink (replace with symbolic links)
- Support a --dry-run flag to preview what would be done without actually modifying files
- Scan directories recursively by default, with --no-recursive flag to disable
- Display a progress bar during scanning showing files processed and duplicates found so far
- Print summary to console: total files scanned, total unique files, duplicate sets found, total wasted space, space that would be recovered
- Save the full deduplication report as JSON with --output flag (default: dedup_report.json)
- If no directories are given, create a sample directory with intentional duplicates (exact copies, files with same content but different names, and unique files), run deduplication analysis, and display the results
- Handle errors: permission denied, broken symlinks, files modified during scan, and cross-filesystem hard links

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# File Deduplicator (Python)

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Install
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run
```bash
python src/main.py dir1 dir2
python src/main.py dir1 --action delete --keep oldest --dry-run
python src/main.py dir1 --action hardlink --keep newest --include "*.txt" --exclude "*.log"
```

Output is written to `dedup_report.json` by default (override with `--output`).

# No external dependencies required

#!/usr/bin/env python3
import argparse
import fnmatch
import hashlib
import json
import os
import shutil
import sys
from pathlib import Path


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def should_include(rel: str, includes: list[str], excludes: list[str]) -> bool:
    inc = (not includes) or any(fnmatch.fnmatch(rel, p) for p in includes)
    exc = any(fnmatch.fnmatch(rel, p) for p in excludes)
    return inc and not exc


def walk_dirs(dirs: list[Path], recursive: bool):
    for d in dirs:
        if not d.exists():
            print(f"Warning: missing directory {d}", file=sys.stderr)
            continue
        if recursive:
            for p in d.rglob("*"):
                if p.is_file():
                    yield p.resolve()
        else:
            for p in d.glob("*"):
                if p.is_file():
                    yield p.resolve()


def scan_files(dirs: list[Path], min_size: int, includes: list[str], excludes: list[str], recursive: bool):
    by_size: dict[int, list[dict]] = {}
    files = list(walk_dirs(dirs, recursive))
    for i, p in enumerate(files, start=1):
        try:
            st = p.stat()
            if st.st_size < min_size:
                continue
            rel = p.name
            if not should_include(rel, includes, excludes):
                continue
            by_size.setdefault(st.st_size, []).append({"path": str(p), "size": st.st_size, "mtime": st.st_mtime})
            if i % 250 == 0 or i == len(files):
                print(f"\rScanned {i}/{len(files)} files...", end="", flush=True)
        except Exception as exc:
            print(f"Warning: {p}: {exc}", file=sys.stderr)
    print()

    dup_sets = []
    for size, group in by_size.items():
        if len(group) < 2:
            continue
        by_hash: dict[str, list[dict]] = {}
        for item in group:
            try:
                h = sha256(Path(item["path"]))
                by_hash.setdefault(h, []).append({**item, "hash": h})
            except Exception as exc:
                print(f"Warning: hash failed {item['path']}: {exc}", file=sys.stderr)
        for arr in by_hash.values():
            if len(arr) > 1:
                dup_sets.append({"size": size, "hash": arr[0]["hash"], "files": arr})

    return {"files_scanned": len(files), "duplicate_sets": dup_sets}


def pick_keeper(files: list[dict], keep: str) -> dict:
    sorted_files = sorted(files, key=lambda f: f["mtime"])
    return sorted_files[0] if keep == "oldest" else sorted_files[-1]


def apply_action(dup_sets: list[dict], action: str, keep: str, dry_run: bool):
    actions = []
    reclaimed = 0
    for ds in dup_sets:
        keeper = pick_keeper(ds["files"], keep)
        for f in ds["files"]:
            if f["path"] == keeper["path"]:
                continue
            row = {"type": action, "source": keeper["path"], "target": f["path"], "status": "planned"}
            if action == "report":
                actions.append(row)
                continue
            if dry_run:
                row["status"] = "dry-run"
                actions.append(row)
                reclaimed += f["size"]
                continue
            try:
                target = Path(f["path"])
                source = Path(keeper["path"])
                if action == "delete":
                    target.unlink(missing_ok=True)
                elif action == "hardlink":
                    target.unlink(missing_ok=True)
                    os.link(source, target)
                elif action == "symlink":
                    target.unlink(missing_ok=True)
                    target.symlink_to(os.path.relpath(source, target.parent))
                row["status"] = "done"
                reclaimed += f["size"]
            except Exception as exc:
                row["status"] = "failed"
                row["error"] = str(exc)
            actions.append(row)
    return {"actions": actions, "reclaimed": reclaimed}


def make_sample() -> list[Path]:
    root = Path("sample_dedup_data").resolve()
    if root.exists():
        shutil.rmtree(root)
    (root / "a").mkdir(parents=True)
    (root / "b").mkdir(parents=True)
    (root / "a" / "x1.txt").write_text("hello duplicate\n", encoding="utf-8")
    (root / "a" / "x2.txt").write_text("hello duplicate\n", encoding="utf-8")
    (root / "b" / "x3.txt").write_text("hello duplicate\n", encoding="utf-8")
    (root / "b" / "unique.txt").write_text("unique\n", encoding="utf-8")
    return [root]


def main() -> int:
    p = argparse.ArgumentParser(description="File deduplicator")
    p.add_argument("dirs", nargs="*", help="Directories to scan")
    p.add_argument("--min-size", type=int, default=1)
    p.add_argument("--include", action="append", default=[])
    p.add_argument("--exclude", action="append", default=[])
    p.add_argument("--action", choices=["report", "delete", "hardlink", "symlink"], default="report")
    p.add_argument("--keep", choices=["oldest", "newest"], default="oldest")
    p.add_argument("--dry-run", action="store_true")
    p.add_argument("--no-recursive", action="store_true")
    p.add_argument("--output", default="dedup_report.json")
    args = p.parse_args()

    dirs = [Path(d) for d in args.dirs] if args.dirs else make_sample()

    try:
        scan = scan_files(dirs, args.min_size, args.include, args.exclude, not args.no_recursive)
        actions = apply_action(scan["duplicate_sets"], args.action, args.keep, args.dry_run)

        total_dup_files = sum(len(ds["files"]) for ds in scan["duplicate_sets"])
        wasted = sum(ds["size"] * (len(ds["files"]) - 1) for ds in scan["duplicate_sets"])

        report = {
            "summary": {
                "totalFilesScanned": scan["files_scanned"],
                "duplicateSetsFound": len(scan["duplicate_sets"]),
                "totalWastedSpace": wasted,
                "recoverableSpace": actions["reclaimed"] if actions["reclaimed"] else wasted,
                "totalUniqueFilesEstimate": scan["files_scanned"] - (total_dup_files - len(scan["duplicate_sets"])),
            },
            "config": vars(args),
            "duplicateSets": scan["duplicate_sets"],
            "actions": actions["actions"],
        }

        print(f"Duplicate sets: {report['summary']['duplicateSetsFound']}")
        print(f"Wasted space: {report['summary']['totalWastedSpace']} bytes")
        print(f"Recoverable: {report['summary']['recoverableSpace']} bytes")

        Path(args.output).write_text(json.dumps(report, indent=2), encoding="utf-8")
        return 0
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


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