101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""verify_nas.py — NAS mirror consistency verifier (Phase D2).
|
|
|
|
Idempotent, safe to run anytime. After a sync to the NAS mirror it asserts, for
|
|
each project directory under PROJECT_BASE:
|
|
- the mirror path exists
|
|
- the mirror matches local (rsync --dry-run -i reports no changes)
|
|
|
|
Exit code 0 = all projects verified; 1 = NAS missing/not mounted or a project
|
|
differs from its mirror.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from scripts.nas_lifecycle import DEFAULT_NAS_ROOT, RSYNC_EXCLUDES
|
|
from umbraaudio.config import list_projects
|
|
|
|
|
|
def build_cmd(local_root: Path, nas_root: Path) -> list[str]:
|
|
cmd = ["rsync", "-an", "-i", "--delete"]
|
|
for exc in RSYNC_EXCLUDES:
|
|
cmd += ["--exclude", exc]
|
|
cmd += [f"{local_root}/", f"{nas_root}/"]
|
|
return cmd
|
|
|
|
|
|
def verify_nas(local_root: Path = Path("/Volumes/1TB_APL_Storage/UmbraAudio"),
|
|
nas_root: Path = DEFAULT_NAS_ROOT) -> int:
|
|
projects = list_projects()
|
|
if not projects:
|
|
print("No projects found; nothing to verify.")
|
|
return 0
|
|
|
|
if not nas_root.exists():
|
|
print(f"[FAIL] NAS mirror not present: {nas_root}")
|
|
return 1
|
|
|
|
# Mirror must not be empty (nas_mounted check in nas_lifecycle requires it).
|
|
if not nas_root.is_dir() or not any(nas_root.iterdir()):
|
|
print(f"[FAIL] NAS mirror is empty/unmounted: {nas_root}")
|
|
return 1
|
|
|
|
print(f"Verifying NAS mirror {nas_root} against {local_root}\n")
|
|
problems: list[str] = []
|
|
|
|
for p in projects:
|
|
slug = str(p["slug"])
|
|
mirror = nas_root / slug
|
|
if not mirror.exists():
|
|
problems.append(f"{slug}: missing on NAS ({mirror})")
|
|
continue
|
|
print(f" {slug}: project dir present")
|
|
|
|
cmd = build_cmd(local_root, nas_root)
|
|
try:
|
|
result = subprocess.run(
|
|
cmd, capture_output=True, text=True, errors="replace", timeout=600
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
print("[FAIL] rsync dry-run timed out (600s)")
|
|
return 1
|
|
changes = [ln for ln in result.stdout.splitlines() if ln.strip()]
|
|
|
|
if result.returncode != 0:
|
|
print(f"[FAIL] rsync dry-run errored:\n{result.stderr[-500:]}")
|
|
return 1
|
|
|
|
if changes:
|
|
print(f"\n[FAIL] {len(changes)} difference(s) between local and NAS mirror:")
|
|
for ln in changes[:50]:
|
|
print(" ", ln)
|
|
if len(changes) > 50:
|
|
print(f" ... and {len(changes) - 50} more")
|
|
problems.append("local/NAS out of sync")
|
|
else:
|
|
print("\nNAS mirror matches local (no changes).")
|
|
|
|
print(f"\n{'FAILED' if problems else 'PASSED'} ({len(problems)} issue(s))")
|
|
if problems:
|
|
for pr in problems:
|
|
print(" [x]", pr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
default_local = Path("/Volumes/1TB_APL_Storage/UmbraAudio")
|
|
local_root = Path(sys.argv[1]) if len(sys.argv) > 1 else default_local
|
|
nas_root = Path(sys.argv[2]) if len(sys.argv) > 2 else DEFAULT_NAS_ROOT
|
|
return verify_nas(local_root, nas_root)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|