diff --git a/find-dupes/find-dupes.py b/find-dupes/find-dupes.py new file mode 100755 index 0000000..acdc09c --- /dev/null +++ b/find-dupes/find-dupes.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +import argparse +import hashlib +import os +import sys +from collections import defaultdict +from pathlib import Path + +CHUNK = 1024 + +G = "\033[32m" +Y = "\033[33m" +C = "\033[36m" +R = "\033[31m" +B = "\033[1m" +Z = "\033[0m" + + +def parse_size(s: str) -> int: + s = s.strip().lower() + mul = 1 + for suf, m in ( + ("kb", 1024), ("k", 1024), + ("mb", 1024**2), ("m", 1024**2), + ("gb", 1024**3), ("g", 1024**3), + ("b", 1), + ): + if s.endswith(suf): + s = s[:-len(suf)].strip() + mul = m + break + return int(float(s) * mul) + + +def human(n: int) -> str: + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if n < 1024 or unit == "TiB": + return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}" + n /= 1024 + + +def prog(stage: str, cur: int, total: int) -> None: + msg = f"{stage:<9} | {cur}/{total} files" + if not hasattr(prog, "_last"): + prog._last = 0 + pad = max(0, prog._last - len(msg)) + sys.stderr.write("\r" + C + msg + Z + (" " * pad)) + sys.stderr.flush() + prog._last = len(msg) + + +def parse_args(): + ap = argparse.ArgumentParser() + ap.add_argument("root", nargs="?", default=".") + ap.add_argument("--min-size", default="1") + ap.add_argument("--max-size", default=None) + ap.add_argument("--ignore", action="append", default=[]) + return ap.parse_args() + + +def is_ignored(p: Path, ignore: list[Path]) -> bool: + try: + rp = p.resolve() + except OSError: + return False + return any(rp == q or q in rp.parents for q in ignore) + + +def file_head_tail(p: Path, n: int = CHUNK): + try: + size = p.stat().st_size + with p.open("rb") as f: + head = f.read(n) + f.seek(max(0, size - n)) + tail = f.read(n) + return head, tail + except OSError: + return None, None + + +def sha256sum(p: Path) -> str | None: + h = hashlib.sha256() + try: + with p.open("rb") as f: + while True: + b = f.read(1024 * 1024) + if not b: + break + h.update(b) + return h.hexdigest() + except OSError: + return None + + +def iter_files(root: Path, ignore: list[Path], min_size: int, max_size: int | None): + for dirpath, dirnames, filenames in os.walk(root): + dp = Path(dirpath) + dirnames[:] = [d for d in dirnames if not (dp / d).is_symlink() and not is_ignored(dp / d, ignore)] + + for name in filenames: + p = dp / name + if p.is_symlink() or is_ignored(p, ignore): + continue + try: + size = p.stat().st_size + except OSError: + continue + if size < min_size: + continue + if max_size is not None and size > max_size: + continue + yield p, size + + +def main(): + args = parse_args() + root = Path(args.root).expanduser().resolve() + min_size = parse_size(args.min_size) + max_size = parse_size(args.max_size) if args.max_size else None + ignore = [Path(p).expanduser().resolve() for p in args.ignore] + + by_size = defaultdict(list) + total = 0 + + for p, size in iter_files(root, ignore, min_size, max_size): + total += 1 + if total % 100 == 0: + prog("Scanning", total, total) + by_size[size].append(p) + + if not total: + print("No files found.") + return + + prog("Scanning", total, total) + sys.stderr.write("\n") + + candidates = {s: ps for s, ps in by_size.items() if len(ps) > 1} + if not candidates: + print("No duplicates found.") + return + + by_ht = defaultdict(list) + ht_total = sum(len(ps) for ps in candidates.values()) + done = 0 + for size, ps in candidates.items(): + for p in ps: + done += 1 + prog("Head/tail", done, ht_total) + h, t = file_head_tail(p) + if h is not None: + by_ht[(size, h, t)].append(p) + sys.stderr.write("\n") + + ht_groups = {k: ps for k, ps in by_ht.items() if len(ps) > 1} + if not ht_groups: + print("No duplicates found.") + return + + by_hash = defaultdict(list) + hash_total = sum(len(ps) for ps in ht_groups.values()) + done = 0 + for ps in ht_groups.values(): + for p in ps: + done += 1 + prog("Hashing", done, hash_total) + h = sha256sum(p) + if h: + by_hash[h].append(p) + sys.stderr.write("\n") + + groups = [] + for h, ps in by_hash.items(): + if len(ps) > 1: + try: + size = ps[0].stat().st_size + except OSError: + size = 0 + groups.append((len(ps), size, h, ps)) + + if not groups: + print("No duplicates found.") + return + + groups.sort(key=lambda x: (-x[0], -x[1])) + + for idx, (count, size, h, ps) in enumerate(groups, 1): + print(f"{B}{Y}Group {idx}{Z} ({count} files, {human(size)}) {C}{h[:12]}{Z}") + for p in sorted(ps): + print(f" {G}{p.resolve()}{Z}") + + +if __name__ == "__main__": + main() +