When managing deployments, verifying backups, or auditing system configurations across environments, developers and DevOps engineers frequently need to com
When managing deployments, verifying backups, or auditing system configurations across environments, developers and DevOps engineers frequently need to compare two directory trees. While commands like diff -r or rsync -n exist on Unix systems, they lack cross-platform consistency, cannot easily produce machine-readable JSON reports for CI/CD pipelines, and often hit performance or memory bottlenecks on large directory trees.
This script solves the problem of structural and content auditing by recursively scanning two directories, computing SHA-256 hashes for file integrity, and identifying added, removed, modified, and identical files. It outputs structured JSON or human-readable text summaries without requiring third-party dependencies.
The script operates in three distinct phases:
--src) and destination (--dst) directories. For every file encountered, it records relative paths, byte sizes, modification timestamps, and calculates cryptographic hashes (default SHA-256) using chunked streaming to keep memory usage low.The architecture relies on standard design patterns: a data transfer object pattern using dataclasses for tracking file metadata, and a functional pipeline approach for traversal, comparison, and reporting.
#!/usr/bin/env python3
"""
Directory Tree Comparator and Diff Reporter
Recursively compares two directory trees for structural and content differences.
Generates machine-readable JSON reports or human-readable summaries using only
the Python standard library.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import fnmatch
import hashlib
import json
import os
import sys
from pathlib import Path
from typing import Any, NamedTuple
# Environment variable overrides with sensible defaults
DEFAULT_HASH_ALGO = os.environ.get("DIR_DIFF_HASH_ALGO", "sha256")
DEFAULT_CHUNK_SIZE = int(os.environ.get("DIR_DIFF_CHUNK_SIZE", "65536"))
DEFAULT_IGNORE = os.environ.get("DIR_DIFF_IGNORE", ".git,*.pyc,__pycache__,.DS_Store")
class FileEntry(NamedTuple):
"""Represents file metadata and state within a directory tree."""
rel_path: str
size: int
mtime: float
sha256: str
def compute_file_hash(
file_path: Path, algo: str = "sha256", chunk_size: int = 65536
) -> str:
"""Computes the cryptographic hash of a file using chunked reads to save memory."""
try:
hasher = hashlib.new(algo)
except ValueError:
sys.exit(f"Error: Unsupported hash algorithm '{algo}'.")
try:
with open(file_path, "rb") as f:
while chunk := f.read(chunk_size):
hasher.update(chunk)
return hasher.hexdigest()
except (PermissionError, OSError) as e:
# If a file cannot be read, return an error token rather than crashing
return f"ERROR_UNREADABLE: {e}"
def should_ignore(rel_path: str, ignore_patterns: list[str]) -> bool:
"""Checks if a relative path matches any defined ignore pattern."""
path_parts = rel_path.split(os.sep)
for pattern in ignore_patterns:
pattern = pattern.strip()
if not pattern:
continue
# Check against full relative path or any single path component (directory name)
if fnmatch.fnmatch(rel_path, pattern) or any(
fnmatch.fnmatch(part, pattern) for part in path_parts
):
return True
return False
def scan_directory(
root: Path,
ignore_patterns: list[str],
hash_algo: str,
chunk_size: int,
) -> dict[str, FileEntry]:
"""
Recursively scans a directory tree and returns a mapping of relative paths
to FileEntry objects containing metadata and content hashes.
"""
entries: dict[str, FileEntry] = {}
if not root.exists():
raise FileNotFoundError(f"Target directory does not exist: {root}")
if not root.is_dir():
raise NotADirectoryError(f"Target path is not a directory: {root}")
for current_dir, _, files in os.walk(root):
for file_name in files:
full_path = Path(current_dir) / file_name
# Resolve relative path to ensure consistent key comparison
rel_path = str(full_path.relative_to(root))
if should_ignore(rel_path, ignore_patterns):
continue
try:
stat = full_path.stat()
file_hash = compute_file_hash(full_path, hash_algo, chunk_size)
entries[rel_path] = FileEntry(
rel_path=rel_path,
size=stat.st_size,
mtime=stat.st_mtime,
sha256=file_hash,
)
except OSError as e:
print(
f"Warning: Skipping unreadable file '{full_path}': {e}",
file=sys.stderr,
)
return entries
def compare_trees(
src_entries: dict[str, FileEntry],
dst_entries: dict[str, FileEntry],
) -> dict[str, Any]:
"""Compares two scanned directory entry maps and categorizes differences."""
src_keys = set(src_entries.keys())
dst_keys = set(dst_entries.keys())
only_in_src = sorted(src_keys - dst_keys)
only_in_dst = sorted(dst_keys - src_keys)
common_keys = src_keys & dst_keys
modified: list[dict[str, Any]] = []
identical: list[str] = []
for key in sorted(common_keys):
src_file = src_entries[key]
dst_file = dst_entries[key]
reasons: list[str] = []
if src_file.size != dst_file.size:
reasons.append("size_mismatch")
if src_file.sha256 != dst_file.sha256:
reasons.append("hash_mismatch")
if reasons:
modified.append(
{
"path": key,
"reasons": reasons,
"src_size": src_file.size,
"dst_size": dst_file.size,
"src_hash": src_file.sha256,
"dst_hash": dst_file.sha256,
"src_mtime": datetime.fromtimestamp(
src_file.mtime, tz=timezone.utc
).isoformat(),
"dst_mtime": datetime.fromtimestamp(
dst_file.mtime, tz=timezone.utc
).isoformat(),
}
)
else:
identical.append(key)
return {
"summary": {
"total_src_files": len(src_entries),
"total_dst_files": len(dst_entries),
"only_in_src_count": len(only_in_src),
"only_in_dst_count": len(only_in_dst),
"modified_count": len(modified),
"identical_count": len(identical),
},
"only_in_src": only_in_src,
"only_in_dst": only_in_dst,
"modified": modified,
"identical": identical,
}
def render_text_report(diff_results: dict[str, Any], src_path: Path, dst_path: Path) -> str:
"""Generates a human-readable text report from the diff dictionary."""
summary = diff_results["summary"]
lines = [
"=" * 60,
" DIRECTORY TREE DIFF REPORT",
"=" * 60,
f" Source (src) : {src_path.resolve()}",
f" Destination (dst): {dst_path.resolve()}",
"-" * 60,
f" Files only in SRC : {summary['only_in_src_count']}",
f" Files only in DST : {summary['only_in_dst_count']}",
f" Modified files : {summary['modified_count']}",
f" Identical files : {summary['identical_count']}",
"=" * 60,
]
if diff_results["only_in_src"]:
lines.append("\n[+] ONLY IN SOURCE (Removed in Destination):")
lines.extend(f" - {path}" for path in diff_results["only_in_src"])
if diff_results["only_in_dst"]:
lines.append("\n[+] ONLY IN DESTINATION (Added in Destination):")
lines.extend(f" + {path}" for path in diff_results["only_in_dst"])
if diff_results["modified"]:
lines.append("\n[*] MODIFIED FILES:")
for mod in diff_results["modified"]:
reasons = ", ".join(mod["reasons"])
lines.append(f" * {mod['path']} [{reasons}]")
return "\n".join(lines)
def build_parser() -> argparse.ArgumentParser:
"""Constructs the command-line argument parser."""
parser = argparse.ArgumentParser(
description="Compare two directory trees and report structural/content differences."
)
parser.add_argument(
"--src",
required=True,
type=Path,
help="Path to the source directory.",
)
parser.add_argument(
"--dst",
required=True,
type=Path,
help="Path to the destination directory.",
)
parser.add_argument(
"--output",
"-o",
type=Path,
default=None,
help="Path to output file. If omitted, prints text report to standard output.",
)
parser.add_argument(
"--format",
choices=["json", "text"],
default="json",
help="Format of the output report file (default: json).",
)
parser.add_argument(
"--ignore",
default=DEFAULT_IGNORE,
help=f"Comma-separated list of glob patterns to ignore (default: '{DEFAULT_IGNORE}').",
)
parser.add_argument(
"--hash-algo",
default=DEFAULT_HASH_ALGO,
help=f"Cryptographic hash algorithm to use (default: '{DEFAULT_HASH_ALGO}').",
)
return parser
def main() -> None:
"""Main execution entry point."""
parser = build_parser()
args = parser.parse_args()
ignore_patterns = [p.strip() for p in args.ignore.split(",") if p.strip()]
print(f"Scanning Source: {args.src} ...", file=sys.stderr)
try:
src_entries = scan_directory(
args.src, ignore_patterns, args.hash_algo, DEFAULT_CHUNK_SIZE
)
except (FileNotFoundError, NotADirectoryError) as e:
sys.exit(f"Source error: {e}")
print(f"Scanning Destination: {args.dst} ...", file=sys.stderr)
try:
dst_entries = scan_directory(
args.dst, ignore_patterns, args.hash_algo, DEFAULT_CHUNK_SIZE
)
except (FileNotFoundError, NotADirectoryError) as e:
sys.exit(f"Destination error: {e}")
print("Comparing trees...", file=sys.stderr)
diff_data = compare_trees(src_entries, dst_entries)
# Format result payload
if args.format == "text" and not args.output:
report = render_text_report(diff_data, args.src, args.dst)
print(report)
elif args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
with open(args.output, "w", encoding="utf-8") as f:
if args.format == "json":
json.dump(diff_data, f, indent=2)
else:
f.write(render_text_report(diff_data, args.src, args.dst))
print(f"Report written to: {args.output}", file=sys.stderr)
else:
# Default JSON console dump if format is json and no output file provided
print(json.dumps(diff_data, indent=2))
if __name__ == "__main__":
main()
The script can be configured via command-line arguments or environment variables. Command-line flags always take precedence over environment variable defaults.
| Variable | Default Value | Description |
|---|---|---|
DIR_DIFF_HASH_ALGO |
sha256 |
Hash algorithm supported by hashlib (e.g., sha256, md5, sha1). |
DIR_DIFF_CHUNK_SIZE |
65536 (64KB) |
Read buffer size in bytes for reading files during hashing. |
DIR_DIFF_IGNORE |
.git,*.pyc,__pycache__,.DS_Store |
Comma-separated glob patterns to exclude from comparison. |
Compare a release directory against a build directory and save the result to a JSON file:
python3 dir_diff.py --src ./v1.0.0 --dst ./v1.0.1 --output diff_report.json
Output in diff_report.json:
{
"summary": {
"total_src_files": 3,
"total_dst_files": 3,
"only_in_src_count": 1,
"only_in_dst_count": 1,
"modified_count": 1,
"identical_count": 1
},
"only_in_src": [
"deprecated.py"
],
"only_in_dst": [
"features/new_module.py"
],
"modified": [
{
"path": "config.json",
"reasons": [
"size_mismatch",
"hash_mismatch"
],
"src_size": 128,
"dst_size": 142,
"src_hash": "a1b2c3...",
"dst_hash": "e5f6g7...",
"src_mtime": "2023-10-01T12:00:00+00:00",
"dst_mtime": "2023-10-02T09:30:00+00:00"
}
],
"identical": [
"main.py"
]
}
Print a human-readable text report directly to stdout while excluding custom build files:
python3 dir_diff.py --src ./src_dir --dst ./dst_dir --format text --ignore "*.log,tmp/*,*.tmp"
Output:
Scanning Source: ./src_dir ...
Scanning Destination: ./dst_dir ...
Comparing trees...
============================================================
DIRECTORY TREE DIFF REPORT
============================================================
Source (src) : /home/user/src_dir
Destination (dst): /home/user/dst_dir
------------------------------------------------------------
Files only in SRC : 1
Files only in DST : 1
Modified files : 1
Identical files : 1
============================================================
[+] ONLY IN SOURCE (Removed in Destination):
- deprecated.py
[+] ONLY IN DESTINATION (Added in Destination):
+ features/new_module.py
[*] MODIFIED FILES:
* config.json [size_mismatch, hash_mismatch]
Pre-configure defaults via environment variables inside a Docker or CI environment:
export DIR_DIFF_HASH_ALGO="md5"
export DIR_DIFF_IGNORE="*.bak,node_modules"
python3 dir_diff.py --src ./deployment/old --dst ./deployment/new -o /tmp/audit.json
--src or --dst points to a non-existent path or file instead of a directory, the script handles FileNotFoundError and NotADirectoryError gracefully, outputting a clear error message to stderr and exiting with a non-zero exit code.scan_directory function catches OSError and PermissionError, prints a warning to stderr, and stores an explicit ERROR_UNREADABLE string inside the hash field.--hash-algo or DIR_DIFF_HASH_ALGO, the script catches ValueError from hashlib.new() and exits cleanly with a descriptive message.--output, parent directories are automatically created using path.parent.mkdir(parents=True, exist_ok=True) to prevent FileNotFoundError during output writing.concurrent.futures.ProcessPoolExecutor in scan_directory to process file hashes concurrently, significantly speeding up execution on multi-gigabyte directories.--fast mode flag that skips cryptographic hashing entirely and compares files solely by file size and modification timestamp (mtime).--sync flag that accepts actions (e.g., --sync=copy-missing) to turn the script into a bidirectional synchronization utility using shutil.copy2.