576 lines
18 KiB
Python
Executable File
576 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""CLI wrapper for the treecompare module.
|
|
|
|
The comparison engine lives in treecompare.py. This file intentionally stays
|
|
thin: argument parsing, logging, progress display, and human-oriented reporting.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import textwrap
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Optional
|
|
|
|
# Support the recommended personal module install location.
|
|
PERSONAL_LIB = Path.home() / "lib" / "python"
|
|
if str(PERSONAL_LIB) not in sys.path:
|
|
sys.path.insert(0, str(PERSONAL_LIB))
|
|
|
|
try:
|
|
from treecompare import CompareOptions, CompareResult, compare_paths
|
|
except ImportError as exc: # pragma: no cover - user install failure path
|
|
print(
|
|
"tree-compare: could not import the treecompare module.\n"
|
|
f"Expected it at: {PERSONAL_LIB / 'treecompare.py'}\n"
|
|
"Install it with something like:\n"
|
|
" mkdir -p \"$HOME/lib/python\"\n"
|
|
" cp treecompare.py \"$HOME/lib/python/treecompare.py\"",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(2) from exc
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="tree-compare",
|
|
description="Compare two or more directory trees by relative paths and selected file attributes.",
|
|
)
|
|
parser.add_argument(
|
|
"paths",
|
|
nargs="+",
|
|
help="Directory roots to compare. Provide two or more.",
|
|
)
|
|
|
|
compare = parser.add_argument_group("comparison options")
|
|
compare.add_argument(
|
|
"--ignore-extensions",
|
|
action="store_true",
|
|
help="Compare filenames after removing the final extension from the basename.",
|
|
)
|
|
compare.add_argument(
|
|
"--ignore-case",
|
|
action="store_true",
|
|
help="Compare relative paths case-insensitively.",
|
|
)
|
|
compare.add_argument(
|
|
"--hash",
|
|
dest="use_hash",
|
|
action="store_true",
|
|
help="Hash same-key, same-size regular files and compare hash values. Defaults to md5.",
|
|
)
|
|
compare.add_argument(
|
|
"--hash-algorithm",
|
|
default="md5",
|
|
help="Hashlib algorithm to use with --hash/--check-all. Default: md5.",
|
|
)
|
|
compare.add_argument(
|
|
"--check-permissions",
|
|
action="store_true",
|
|
help="Compare permission bits and uid/gid where available.",
|
|
)
|
|
compare.add_argument(
|
|
"--check-time",
|
|
action="store_true",
|
|
help="Compare modification timestamps.",
|
|
)
|
|
compare.add_argument(
|
|
"--mtime-tolerance",
|
|
type=float,
|
|
default=0.0,
|
|
metavar="SECONDS",
|
|
help="Allowed mtime drift when --check-time is enabled. Default: 0.",
|
|
)
|
|
compare.add_argument(
|
|
"--check-all",
|
|
action="store_true",
|
|
help="Enable --hash, --check-permissions, and --check-time.",
|
|
)
|
|
compare.add_argument(
|
|
"--ignore-symlinks",
|
|
action="store_true",
|
|
help="Skip symlink entries. Symlinked directories are never traversed.",
|
|
)
|
|
|
|
output = parser.add_argument_group("output options")
|
|
output.add_argument(
|
|
"--json",
|
|
dest="json_path",
|
|
metavar="PATH",
|
|
help="Write full comparison results as JSON to PATH.",
|
|
)
|
|
output.add_argument(
|
|
"--include-records",
|
|
action="store_true",
|
|
help="Include all scanned file records in JSON output. Can be large.",
|
|
)
|
|
output.add_argument(
|
|
"--quiet",
|
|
action="store_true",
|
|
help="Only print summary and errors.",
|
|
)
|
|
output.add_argument(
|
|
"--no-progress",
|
|
action="store_true",
|
|
help="Disable stderr progress updates.",
|
|
)
|
|
output.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=200,
|
|
help="Maximum missing/collision/difference entries to print to stdout. Default: 200. Use -1 for all.",
|
|
)
|
|
output.add_argument(
|
|
"--width",
|
|
type=int,
|
|
default=0,
|
|
metavar="COLUMNS",
|
|
help="Wrap human-readable output to this width. Default: detected terminal width.",
|
|
)
|
|
output.add_argument(
|
|
"--no-root-map",
|
|
action="store_true",
|
|
help="Do not print the root index map before the summary.",
|
|
)
|
|
output.add_argument(
|
|
"--no-values",
|
|
action="store_true",
|
|
help="For different files, print only the differing attribute names, not per-root values.",
|
|
)
|
|
output.add_argument(
|
|
"--no-fail-on-diff",
|
|
action="store_true",
|
|
help="Exit 0 even when differences are found. Usage/configuration errors still exit 2.",
|
|
)
|
|
|
|
logging_group = parser.add_argument_group("logging options")
|
|
logging_group.add_argument(
|
|
"--log",
|
|
nargs="?",
|
|
const="tree-compare.log",
|
|
default="tree-compare.log",
|
|
metavar="PATH",
|
|
help="Write a log file. Default: ./tree-compare.log. If provided without PATH, uses ./tree-compare.log.",
|
|
)
|
|
logging_group.add_argument(
|
|
"--no-log",
|
|
action="store_true",
|
|
help="Disable log file creation.",
|
|
)
|
|
|
|
return parser
|
|
|
|
|
|
def clean_root_text(raw: str) -> str:
|
|
"""Normalize user-visible root strings without changing filesystem semantics unnecessarily."""
|
|
text = raw.strip()
|
|
while len(text) > 1 and text.endswith(("/", "\\")):
|
|
candidate = text[:-1]
|
|
# Avoid turning C:\ into C: on Windows-style input.
|
|
if len(candidate) == 2 and candidate[1] == ":":
|
|
break
|
|
text = candidate
|
|
return text
|
|
|
|
|
|
def validate_paths(parser: argparse.ArgumentParser, raw_paths: Iterable[str]) -> list[str]:
|
|
cleaned = [clean_root_text(p) for p in raw_paths]
|
|
if len(cleaned) < 2:
|
|
parser.error("provide two or more directory paths")
|
|
|
|
validated: list[str] = []
|
|
for text in cleaned:
|
|
path = Path(text).expanduser()
|
|
if not path.exists():
|
|
parser.error(f"path does not exist: {text}")
|
|
if path.is_file():
|
|
parser.error(f"path is a file, not a directory: {text}")
|
|
if not path.is_dir():
|
|
parser.error(f"path is not a directory: {text}")
|
|
validated.append(str(path))
|
|
return validated
|
|
|
|
|
|
def setup_logging(log_arg: Optional[str], no_log: bool) -> Optional[Path]:
|
|
if no_log:
|
|
logging.basicConfig(level=logging.CRITICAL + 1)
|
|
return None
|
|
|
|
log_path = Path(log_arg or "tree-compare.log").expanduser()
|
|
logging.basicConfig(
|
|
filename=str(log_path),
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
)
|
|
return log_path
|
|
|
|
|
|
def output_width(requested_width: int = 0) -> int:
|
|
if requested_width and requested_width > 20:
|
|
return requested_width
|
|
return max(50, shutil.get_terminal_size(fallback=(100, 24)).columns)
|
|
|
|
|
|
def compact_path(path_text: str) -> str:
|
|
"""Make paths easier to read by replacing the home directory prefix with ~."""
|
|
text = str(path_text)
|
|
try:
|
|
home = str(Path.home())
|
|
if text == home:
|
|
return "~"
|
|
if text.startswith(home + os.sep):
|
|
return "~" + text[len(home):]
|
|
except Exception:
|
|
pass
|
|
return text
|
|
|
|
|
|
def shorten_middle(text: str, max_len: int) -> str:
|
|
if max_len <= 0 or len(text) <= max_len:
|
|
return text
|
|
if max_len <= 5:
|
|
return text[:max_len]
|
|
left = max_len // 2 - 1
|
|
right = max_len - left - 3
|
|
return text[:left] + "..." + text[-right:]
|
|
|
|
|
|
def print_wrapped(prefix: str, text: Any, *, width: int, subsequent_prefix: Optional[str] = None) -> None:
|
|
body = str(text)
|
|
subsequent = subsequent_prefix if subsequent_prefix is not None else " " * len(prefix)
|
|
print(
|
|
textwrap.fill(
|
|
body,
|
|
width=width,
|
|
initial_indent=prefix,
|
|
subsequent_indent=subsequent,
|
|
break_long_words=False,
|
|
break_on_hyphens=False,
|
|
)
|
|
)
|
|
|
|
|
|
def format_size(value: Any) -> str:
|
|
try:
|
|
size = int(value)
|
|
except Exception:
|
|
return str(value)
|
|
|
|
units = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"]
|
|
n = float(size)
|
|
unit = units[0]
|
|
for unit in units:
|
|
if abs(n) < 1024.0 or unit == units[-1]:
|
|
break
|
|
n /= 1024.0
|
|
|
|
if unit == "B":
|
|
return f"{size:,} B"
|
|
return f"{n:.2f} {unit} ({size:,} bytes)"
|
|
|
|
|
|
def format_mtime_ns(value: Any) -> str:
|
|
try:
|
|
ns = int(value)
|
|
except Exception:
|
|
return str(value)
|
|
dt = datetime.fromtimestamp(ns / 1_000_000_000).astimezone()
|
|
return f"{dt.isoformat(timespec='seconds')} ({ns})"
|
|
|
|
|
|
def format_value(attr: str, value: Any) -> str:
|
|
if attr == "size":
|
|
return format_size(value)
|
|
if attr == "mtime_ns":
|
|
return format_mtime_ns(value)
|
|
if value is None:
|
|
return "<none>"
|
|
return str(value)
|
|
|
|
|
|
def root_labels(result: CompareResult) -> dict[int, str]:
|
|
return {idx: compact_path(root) for idx, root in enumerate(result.roots)}
|
|
|
|
|
|
def make_progress_callback(enabled: bool, *, width: int):
|
|
if not enabled:
|
|
return None
|
|
|
|
last_print = {"time": 0.0}
|
|
|
|
def progress(stage: str, current: int, total: int, message: str) -> None:
|
|
now = time.monotonic()
|
|
should_print = current == total or now - last_print["time"] >= 0.5
|
|
if not should_print:
|
|
return
|
|
last_print["time"] = now
|
|
msg = shorten_middle(compact_path(message), max(10, width - 20))
|
|
if total > 0:
|
|
line = f"{stage}: {current}/{total} {msg}"
|
|
else:
|
|
line = f"{stage}: {current} {msg}"
|
|
print("\r" + line[:width].ljust(width), end="", file=sys.stderr, flush=True)
|
|
if current == total:
|
|
print(file=sys.stderr, flush=True)
|
|
|
|
return progress
|
|
|
|
|
|
def print_root_map(result: CompareResult, *, width: int) -> None:
|
|
labels = root_labels(result)
|
|
print("Roots:")
|
|
for idx in range(len(result.roots)):
|
|
print_wrapped(f" [{idx}] ", labels[idx], width=width)
|
|
|
|
|
|
def print_summary(result: CompareResult) -> None:
|
|
summary = result.summary()
|
|
print("Summary:")
|
|
for key, value in summary.items():
|
|
print(f" {key}: {value}")
|
|
|
|
|
|
def print_missing_entry(item: Any, labels: dict[int, str], *, width: int) -> None:
|
|
"""Verbose, entry-oriented missing output. Kept for debugging/manual reuse."""
|
|
print_wrapped(" - ", item.key, width=width, subsequent_prefix=" ")
|
|
|
|
print(" present in:")
|
|
for root_i in item.present_in:
|
|
rel_path = item.paths_by_root.get(root_i, item.key)
|
|
print_wrapped(
|
|
f" [{root_i}] ",
|
|
f"{labels[root_i]} :: {rel_path}",
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
print(" missing from:")
|
|
for root_i in item.missing_from:
|
|
print_wrapped(
|
|
f" [{root_i}] ",
|
|
labels[root_i],
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
|
|
def print_missing_grouped_by_root(result: CompareResult, labels: dict[int, str], *, width: int, limit: int) -> int:
|
|
"""Print missing files grouped by the root they are absent from.
|
|
|
|
This is the normal human-facing layout because it answers the operational
|
|
question directly: "what does this root lack?"
|
|
|
|
Returns the number of missing item/root pairs printed. One MissingEntry may
|
|
count more than once when comparing three or more roots.
|
|
"""
|
|
if not result.missing or limit == 0:
|
|
return 0
|
|
|
|
printed = 0
|
|
root_count = len(result.roots)
|
|
|
|
for missing_root in range(root_count):
|
|
items = [item for item in result.missing if missing_root in item.missing_from]
|
|
if not items:
|
|
continue
|
|
if limit > 0 and printed >= limit:
|
|
break
|
|
|
|
print()
|
|
print_wrapped(
|
|
f"Missing in [{missing_root}] ",
|
|
f"{labels[missing_root]}:",
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
for item in items:
|
|
if limit > 0 and printed >= limit:
|
|
break
|
|
|
|
present = tuple(root_i for root_i in item.present_in if root_i != missing_root)
|
|
suffix = ""
|
|
if root_count > 2 and present:
|
|
suffix = " (present in " + ", ".join(f"[{root_i}]" for root_i in present) + ")"
|
|
|
|
print_wrapped(" - ", f"{item.key}{suffix}", width=width, subsequent_prefix=" ")
|
|
|
|
# If normalization changed the visible path, show the on-disk names.
|
|
# Example: --ignore-case or --ignore-extensions may make the compare
|
|
# key differ from the actual relative path in the source root.
|
|
alternate_paths = []
|
|
for root_i in present:
|
|
rel_path = item.paths_by_root.get(root_i, item.key)
|
|
if rel_path != item.key:
|
|
alternate_paths.append((root_i, rel_path))
|
|
|
|
if alternate_paths:
|
|
print(" present as:")
|
|
for root_i, rel_path in alternate_paths:
|
|
print_wrapped(
|
|
f" [{root_i}] ",
|
|
rel_path,
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
printed += 1
|
|
|
|
return printed
|
|
|
|
|
|
def print_collision_entry(item: Any, labels: dict[int, str], *, width: int) -> None:
|
|
print_wrapped(" - ", item.key, width=width, subsequent_prefix=" ")
|
|
print_wrapped(
|
|
" root: ",
|
|
f"[{item.root_index}] {labels[item.root_index]}",
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
print(" colliding paths:")
|
|
for rel_path in item.rel_paths:
|
|
print_wrapped(" - ", rel_path, width=width, subsequent_prefix=" ")
|
|
|
|
|
|
def print_difference_entry(item: Any, labels: dict[int, str], *, width: int, show_values: bool) -> None:
|
|
attrs = ", ".join(item.differences.keys())
|
|
print_wrapped(" - ", item.key, width=width, subsequent_prefix=" ")
|
|
print_wrapped(" differs: ", attrs, width=width, subsequent_prefix=" ")
|
|
|
|
# When --ignore-case or --ignore-extensions is used, the compare key may not
|
|
# be an exact on-disk path. Showing per-root paths prevents ambiguity.
|
|
records_by_root = getattr(item, "records_by_root", {})
|
|
if records_by_root and any(record.rel_path != item.key for record in records_by_root.values()):
|
|
print(" paths:")
|
|
for root_i, record in sorted(records_by_root.items()):
|
|
print_wrapped(
|
|
f" [{root_i}] ",
|
|
f"{labels[root_i]} :: {record.rel_path}",
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
if not show_values:
|
|
return
|
|
|
|
for attr, values in item.differences.items():
|
|
print(f" {attr}:")
|
|
for root_i in sorted(values):
|
|
print_wrapped(
|
|
f" [{root_i}] ",
|
|
f"{format_value(attr, values[root_i])}",
|
|
width=width,
|
|
subsequent_prefix=" ",
|
|
)
|
|
|
|
|
|
def print_limited_results(result: CompareResult, limit: int, *, width: int, show_values: bool) -> None:
|
|
if limit == 0:
|
|
return
|
|
|
|
labels = root_labels(result)
|
|
printed = 0
|
|
|
|
def room() -> bool:
|
|
return limit < 0 or printed < limit
|
|
|
|
if result.missing and room():
|
|
printed += print_missing_grouped_by_root(result, labels, width=width, limit=(-1 if limit < 0 else limit - printed))
|
|
|
|
if result.collisions and room():
|
|
print("\nKey collisions:")
|
|
for item in result.collisions:
|
|
if not room():
|
|
break
|
|
print_collision_entry(item, labels, width=width)
|
|
printed += 1
|
|
|
|
if result.differences and room():
|
|
print("\nDifferent:")
|
|
for item in result.differences:
|
|
if not room():
|
|
break
|
|
print_difference_entry(item, labels, width=width, show_values=show_values)
|
|
printed += 1
|
|
|
|
total_entries = len(result.missing) + len(result.collisions) + len(result.differences)
|
|
if limit > 0 and total_entries > printed:
|
|
print(f"\n... {total_entries - printed} more entries not printed. Use --json for full output or --limit -1 to print all.")
|
|
|
|
|
|
def main(argv: Optional[list[str]] = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
width = output_width(args.width)
|
|
|
|
paths = validate_paths(parser, args.paths)
|
|
log_path = setup_logging(args.log, args.no_log)
|
|
|
|
options = CompareOptions(
|
|
ignore_extensions=args.ignore_extensions,
|
|
ignore_case=args.ignore_case,
|
|
use_hash=args.use_hash,
|
|
hash_algorithm=args.hash_algorithm,
|
|
check_permissions=args.check_permissions,
|
|
check_time=args.check_time,
|
|
check_all=args.check_all,
|
|
ignore_symlinks=args.ignore_symlinks,
|
|
mtime_tolerance=args.mtime_tolerance,
|
|
).normalized()
|
|
|
|
logging.info("starting comparison")
|
|
logging.info("paths: %s", paths)
|
|
logging.info("options: %s", options)
|
|
|
|
if log_path and not args.quiet:
|
|
print(f"Log: {log_path}", file=sys.stderr)
|
|
|
|
try:
|
|
result = compare_paths(
|
|
paths,
|
|
options=options,
|
|
progress_callback=make_progress_callback(not args.no_progress and not args.quiet, width=width),
|
|
)
|
|
except Exception as exc:
|
|
logging.exception("comparison failed")
|
|
print(f"tree-compare: error: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
logging.info("summary: %s", result.summary())
|
|
logging.info(
|
|
"result counts: missing=%d collisions=%d differences=%d",
|
|
len(result.missing),
|
|
len(result.collisions),
|
|
len(result.differences),
|
|
)
|
|
|
|
if not args.quiet and not args.no_root_map:
|
|
print_root_map(result, width=width)
|
|
print_summary(result)
|
|
if not args.quiet:
|
|
print_limited_results(result, args.limit, width=width, show_values=not args.no_values)
|
|
|
|
if args.json_path:
|
|
json_path = Path(args.json_path).expanduser()
|
|
with json_path.open("w", encoding="utf-8") as handle:
|
|
json.dump(result.to_dict(include_records=args.include_records), handle, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
print(f"\nWrote JSON: {json_path}")
|
|
logging.info("wrote JSON: %s", json_path)
|
|
|
|
if result.has_differences and not args.no_fail_on_diff:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|