Most Python "file management" tutorials stop at a 30-line shutil.move() loop that sorts files by extension. While great for learning pathlib, that doesn't help when you need to reclaim gigabytes of wasted space across tens of thousands of files without waiting 20 minutes for hashlib to read your entire hard drive.
In this post, we'll build a production-grade Multithreaded Duplicate & Storage Analyzer CLI using pure Python 3.10+ (zero external dependencies) that can scan tens of thousands of files in seconds.
Why Naive Duplicate Finders Are Slow
If you run sha256() on every file in a 500 GB directory, your script is immediately bottlenecked by disk I/O. To avoid reading every byte on the drive, we can use a 3-Stage Hashing Funnel:
-
Stage 1: Byte-Size Grouping (
O(1)metadata lookup) — Two files cannot be identical if their byte counts differ. Grouping bystat().st_sizeeliminates ~85–95% of files immediately without opening a single file handle. -
Stage 2: 4 KB Head/Tail
blake2bSampling — For files that share the exact same size, we read only the first 4 KB and last 4 KB of the file and hash those 8 KB withhashlib.blake2b. This eliminates 99% of remaining false positives (like similarly sized photos or video chunks) in milliseconds. -
Stage 3: Streaming Full
SHA-256Verification — Only files that match in both byte size and 8 KB edge signatures undergo a full 1 MB chunked SHA-256 hash across aThreadPoolExecutor.
On top of that, instead of destructively deleting files (os.remove), every quarantine operation writes to a local JSON transaction log (~/.storage_audit_manifest.json) so you can revert your last batch action with a single undo command.
The Complete Script (disk_audit.py)
Save the following code as disk_audit.py. It requires Python 3.10+ and uses only the standard library:
python
#!/usr/bin/env python3
"""
High-speed disk analyzer, duplicate finder, and reversible file manager.
Zero external dependencies (Python 3.10+ standard library only).
"""
import argparse
import hashlib
import json
import os
import shutil
import sys
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple
MANIFEST_FILE = Path.home() / ".storage_audit_manifest.json"
CHUNK_SIZE = 1024 * 1024 # 1 MB streaming buffer
SAMPLE_SIZE = 4096 # 4 KB fast header/footer sample
@dataclass(frozen=True)
class FileRecord:
path: Path
size: int
mtime: float
def format_bytes(size: int) -> str:
"""Converts raw byte counts into human-readable units."""
for unit in ("B", "KB", "MB", "GB", "TB"):
if size < 1024 or unit == "TB":
return f"{size:.2f} {unit}" if unit != "B" else f"{size} B"
size /= 1024
return f"{size} B"
def fast_sample_hash(path: Path, size: int) -> Optional[str]:
"""Reads only the first and last 4KB of a file to quickly filter non-duplicates."""
try:
hasher = hashlib.blake2b(digest_size=16)
with path.open("rb") as f:
hasher.update(f.read(SAMPLE_SIZE))
if size > SAMPLE_SIZE * 2:
f.seek(-SAMPLE_SIZE, os.SEEK_END)
hasher.update(f.read(SAMPLE_SIZE))
return hasher.hexdigest()
except (OSError, PermissionError):
return None
def full_sha256(path: Path) -> Optional[str]:
"""Computes full SHA-256 digest using memory-efficient 1MB chunks."""
try:
hasher = hashlib.sha256()
with path.open("rb") as f:
while chunk := f.read(CHUNK_SIZE):
hasher.update(chunk)
return hasher.hexdigest()
except (OSError, PermissionError):
return None
def scan_filesystem(root: Path, min_size: int = 1024) -> List[FileRecord]:
"""Recursively yields regular files, skipping symlinks and system directories."""
records: List[FileRecord] = []
skip_dirs = {
".git", "node_modules", ".venv", "__pycache__",
"$RECYCLE.BIN", "System Volume Information"
}
for dirpath, dirnames, filenames in os.walk(root, topdown=True):
# Prune ignored and hidden directories in-place
dirnames[:] = [
d for d in dirnames
if d not in skip_dirs and not d.startswith(".")
]
for fname in filenames:
if fname.startswith("."):
continue
fpath = Path(dirpath) / fname
try:
if fpath.is_symlink():
continue
stat = fpath.stat()
if stat.st_size >= min_size:
records.append(FileRecord(fpath, stat.st_size, stat.st_mtime))
except (OSError, PermissionError):
continue
return records
def find_duplicates(root: Path, min_size: int, workers: int) -> Dict[str, List[FileRecord]]:
"""Three-stage pipeline: Size grouping -> 8KB Blake2b sample -> Full SHA-256."""
t0 = time.perf_counter()
print(f"[*] Scanning '{root}' (min size: {format_bytes(min_size)})...")
files = scan_filesystem(root, min_size)
print(f" Indexed {len(files):,} candidate files in {time.perf_counter() - t0:.2f}s.")
# Stage 1: Group by exact byte size
by_size: Dict[int, List[FileRecord]] = defaultdict(list)
for rec in files:
by_size[rec.size].append(rec)
size_candidates = [
rec for group in by_size.values() if len(group) > 1 for rec in group
]
print(f"[*] Stage 1 (Size match): Narrowed to {len(size_candidates):,} files.")
if not size_candidates:
return {}
# Stage 2: Parallel 4KB head/tail signature
by_sample: Dict[Tuple[int, str], List[FileRecord]] = defaultdict(list)
with ThreadPoolExecutor(max_workers=workers) as pool:
future_map = {
pool.submit(fast_sample_hash, r.path, r.size): r
for r in size_candidates
}
for future in as_completed(future_map):
rec = future_map[future]
sig = future.result()
if sig:
by_sample[(rec.size, sig)].append(rec)
sample_candidates = [
rec for group in by_sample.values() if len(group) > 1 for rec in group
]
print(f"[*] Stage 2 (4KB edge hash): Narrowed to {len(sample_candidates):,} files.")
if not sample_candidates:
return {}
# Stage 3: Full SHA-256 verification
by_full_hash: Dict[str, List[FileRecord]] = defaultdict(list)
with ThreadPoolExecutor(max_workers=workers) as pool:
future_map = {
pool.submit(full_sha256, r.path): r
for r in sample_candidates
}
for future in as_completed(future_map):
rec = future_map[future]
digest = future.result()
if digest:
by_full_hash[digest].append(rec)
confirmed = {
digest: sorted(group, key=lambda r: r.mtime)
for digest, group in by_full_hash.items()
if len(group) > 1
}
return confirmed
def record_transaction(operations: List[Dict[str, str]]) -> None:
"""Saves file moves to a persistent manifest for one-command rollback."""
history = []
if MANIFEST_FILE.exists():
try:
history = json.loads(MANIFEST_FILE.read_text(encoding="utf-8"))
except json.JSONDecodeError:
history = []
history.append({
"timestamp": datetime.now().isoformat(timespec="seconds"),
"operations": operations,
})
MANIFEST_FILE.write_text(json.dumps(history, indent=2), encoding="utf-8")
def rollback_last_batch() -> None:
"""Restores all files moved in the most recent batch operation."""
if not MANIFEST_FILE.exists():
print("[!] No transaction manifest found. Nothing to undo.")
return
history = json.loads(MANIFEST_FILE.read_text(encoding="utf-8"))
if not history:
print("[!] Manifest is empty.")
return
last_batch = history.pop()
ops = last_batch.get("operations", [])
restored = 0
print(f"[*] Rolling back batch from {last_batch['timestamp']} ({len(ops)} files)...")
for op in reversed(ops):
src = Path(op["src"])
dst = Path(op["dst"])
if dst.exists():
src.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(dst), str(src))
restored += 1
else:
print(f" [WARN] Missing quarantined file: {dst}")
MANIFEST_FILE.write_text(json.dumps(history, indent=2), encoding="utf-8")
print(f"[+] Restored {restored}/{len(ops)} files to their original paths.")
def handle_dupes(args: argparse.Namespace) -> None:
root = Path(args.path).expanduser().resolve()
duplicates = find_duplicates(root, args.min_size, args.workers)
if not duplicates:
print("\n[+] Zero exact duplicates found.")
return
reclaimable = sum(
(len(group) - 1) * group[0].size for group in duplicates.values()
)
total_dupe_files = sum(len(group) - 1 for group in duplicates.values())
print(f"\nFound {len(duplicates):,} duplicate sets ({total_dupe_files:,} redundant files).")
print(f"Total reclaimable space: {format_bytes(reclaimable)}\n")
sorted_groups = sorted(
duplicates.items(),
key=lambda item: item[1][0].size * (len(item[1]) - 1),
reverse=True,
)
for idx, (digest, group) in enumerate(sorted_groups[:15], 1):
waste = (len(group) - 1) * group[0].size
print(
f" [{idx}] SHA256: {digest[:12]}... "
f"({format_bytes(group[0].size)} each | Waste: {format_bytes(waste)})"
)
print(f" KEEP (Oldest): {group[0].path}")
for dupe in group[1:]:
print(f" DUPE : {dupe.path}")
if args.quarantine:
q_dir = Path(args.quarantine).expanduser().resolve()
q_dir.mkdir(parents=True, exist_ok=True)
ops = []
for digest, group in sorted_groups:
for dupe in group[1:]:
target = q_dir / f"{digest[:8]}_{dupe.path.name}"
counter = 1
while target.exists():
target = q_dir / f"{digest[:8]}_{counter}_{dupe.path.name}"
counter += 1
shutil.move(str(dupe.path), str(target))
ops.append({"src": str(dupe.path), "dst": str(target)})
record_transaction(ops)
print(f"\n[+] Moved {len(ops)} duplicates to '{q_dir}'.")
print(" Run with 'undo' subcommand at any time to restore them.")
def handle_large(args: argparse.Namespace) -> None:
root = Path(args.path).expanduser().resolve()
files = scan_filesystem(root, min_size=1024 * 1024)
files.sort(key=lambda r: r.size, reverse=True)
print(f"\nTop {min(args.top, len(files))} largest files in '{root}':\n")
for idx, rec in enumerate(files[:args.top], 1):
mod_date = datetime.fromtimestamp(rec.mtime).strftime("%Y-%m-%d")
print(f" {idx:>2}. {format_bytes(rec.size):>10} | {mod_date} | {rec.path}")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="High-speed storage analyzer and duplicate engine."
)
sub = parser.add_subparsers(dest="command", required=True)
# Subcommand: dupes
p_dupes = sub.add_parser(
"dupes", help="Find and optionally quarantine exact duplicate files."
)
p_dupes.add_argument(
"path", nargs="?", default=".", help="Target directory to scan."
)
p_dupes.add_argument(
"--min-size",
type=int,
default=10240,
help="Minimum file size in bytes (default: 10KB).",
)
p_dupes.add_argument(
"--workers",
type=int,
default=8,
help="Thread pool concurrency (default: 8).",
)
p_dupes.add_argument(
"--quarantine",
metavar="DIR",
help="Move duplicate copies into DIR (logged for undo).",
)
p_dupes.set_defaults(func=handle_dupes)
# Subcommand: large
p_large = sub.add_parser(
"large", help="Surface the largest files consuming disk space."
)
p_large.add_argument(
"path", nargs="?", default=".", help="Target directory to scan."
)
p_large.add_argument(
"--top",
type=int,
default=20,
help="Number of files to display (default: 20).",
)
p_large.set_defaults(func=handle_large)
# Subcommand: undo
p_undo = sub.add_parser("undo", help="Revert the last quarantine operation.")
p_undo.set_defaults(func=lambda _: rollback_last_batch())
return parser
if __name__ == "__main__":
cli = build_parser()
parsed_args = cli.parse_args()
try:
parsed_args.func(parsed_args)
except KeyboardInterrupt:
print("\n[!] Operation cancelled by user.")
sys.exit(130)
How to Use It
1. Audit a folder for duplicates (Read-Only)
Scan your ~/Downloads folder for duplicate files larger than 100 KB without modifying a single file:
python disk_audit.py dupes ~/Downloads --min-size 102400
Example Output:
[*] Scanning '/Users/alex/Downloads' (min size: 100.00 KB)...
Indexed 4,812 candidate files in 0.19s.
[*] Stage 1 (Size match): Narrowed to 614 files.
[*] Stage 2 (4KB edge hash): Narrowed to 92 files.
Found 38 duplicate sets (54 redundant files).
Total reclaimable space: 1.42 GB
[1] SHA256: 9f86d081884c... (312.40 MB each | Waste: 624.80 MB)
KEEP (Oldest): /Users/alex/Downloads/ubuntu-iso-backup.iso
DUPE : /Users/alex/Downloads/ubuntu-iso-backup (1).iso
DUPE : /Users/alex/Downloads/old/ubuntu-iso-backup.iso
2. Quarantine Duplicates Safely
Automatically keep the oldest original file in place and move newer redundant copies into a holding folder:
python disk_audit.py dupes ~/Pictures --quarantine ~/Dupe_Quarantine
3. Undo the Last Quarantine Batch
Changed your mind or need files back in their exact nested subdirectories? Run undo:
python disk_audit.py undo
4. Find the Top 25 Space Hogs
Quickly surface the largest individual files across a directory tree:
python disk_audit.py large ~ --top 25
What edge cases or performance bottlenecks do you usually run into when building custom file-scanning scripts in Python? Let me know in the comments below!
Top comments (1)
quarantine with an undo beats deleting outright, that transaction log is the whole safety net.