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:

  1. Stage 1: Byte-Size Grouping (O(1) metadata lookup) — Two files cannot be identical if their byte counts differ. Grouping by stat().st_size eliminates ~85–95% of files immediately without opening a single file handle.
  2. Stage 2: 4 KB Head/Tail blake2b Sampling — 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 with hashlib.blake2b. This eliminates 99% of remaining false positives (like similarly sized photos or video chunks) in milliseconds.
  3. Stage 3: Streaming Full SHA-256 Verification — Only files that match in both byte size and 8 KB edge signatures undergo a full 1 MB chunked SHA-256 hash across a ThreadPoolExecutor.

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!