101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Content-based duplicate file detector.
|
|
|
|
Groups files by size first (cheap), then hashes only size-collisions (SHA-256),
|
|
so identical content is detected regardless of name. Handles symlinks (skips
|
|
by default to avoid loops), empty files (grouped together as one duplicate set),
|
|
and large files (streamed in 1 MiB chunks).
|
|
|
|
Usage:
|
|
dedup.py <dir> [--min-size N] [--follow-symlinks] [--json]
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from collections import defaultdict
|
|
|
|
CHUNK = 1024 * 1024
|
|
|
|
|
|
def sha256_file(path):
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
while True:
|
|
b = f.read(CHUNK)
|
|
if not b:
|
|
break
|
|
h.update(b)
|
|
return h.hexdigest()
|
|
|
|
|
|
def collect(root, follow_symlinks, min_size):
|
|
by_size = defaultdict(list)
|
|
for dirpath, dirnames, filenames in os.walk(root, followlinks=follow_symlinks):
|
|
# prune symlinked dirs unless following
|
|
if not follow_symlinks:
|
|
dirnames[:] = [d for d in dirnames
|
|
if not os.path.islink(os.path.join(dirpath, d))]
|
|
for name in filenames:
|
|
p = os.path.join(dirpath, name)
|
|
if os.path.islink(p) and not follow_symlinks:
|
|
continue
|
|
try:
|
|
st = os.stat(p, follow_symlinks=follow_symlinks)
|
|
except OSError:
|
|
continue
|
|
if not os.path.isfile(p) if False else False:
|
|
pass
|
|
if st.st_size < min_size:
|
|
continue
|
|
by_size[st.st_size].append(p)
|
|
return by_size
|
|
|
|
|
|
def find_duplicates(root, follow_symlinks=False, min_size=1):
|
|
by_size = collect(root, follow_symlinks, min_size)
|
|
groups = []
|
|
for size, paths in by_size.items():
|
|
if len(paths) < 2:
|
|
continue
|
|
by_hash = defaultdict(list)
|
|
for p in paths:
|
|
try:
|
|
by_hash[sha256_file(p)].append(p)
|
|
except OSError:
|
|
continue
|
|
for h, ps in by_hash.items():
|
|
if len(ps) >= 2:
|
|
groups.append({"hash": h, "size": size, "files": sorted(ps)})
|
|
groups.sort(key=lambda g: (-g["size"], g["hash"]))
|
|
return groups
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("root")
|
|
ap.add_argument("--min-size", type=int, default=0)
|
|
ap.add_argument("--follow-symlinks", action="store_true")
|
|
ap.add_argument("--json", action="store_true")
|
|
args = ap.parse_args()
|
|
if not os.path.isdir(args.root):
|
|
print("error: not a directory: " + args.root, file=sys.stderr)
|
|
return 2
|
|
groups = find_duplicates(args.root, args.follow_symlinks, args.min_size)
|
|
if args.json:
|
|
print(json.dumps(groups, indent=2))
|
|
return 0
|
|
if not groups:
|
|
print("no duplicates found")
|
|
return 0
|
|
for i, g in enumerate(groups, 1):
|
|
print("group %d: %d files, %d bytes, sha256=%s" % (i, len(g["files"]), g["size"], g["hash"]))
|
|
for f in g["files"]:
|
|
print(" " + f)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|