#!/usr/bin/env python3 """ Read-only exact-hash audit for scope-triaged files. Purpose: - Compare files routed to 03_PRIVAT / 01_INBOX against the existing authoritative /data/03_PRIVAT and /data/01_INBOX trees. - Detect exact duplicates using full-file SHA-256 only. - Also report cross-scope exact matches. - Never move, rename, delete, overwrite, or modify target files. The only output is a TSV report written to the staging MANIFESTS directory (or another explicitly supplied path). """ from __future__ import annotations import argparse import csv import hashlib import os import sys from collections import defaultdict from pathlib import Path CHUNK = 1024 * 1024 def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(CHUNK), b""): h.update(chunk) return h.hexdigest() def index_tree(root: Path) -> tuple[dict[str, list[Path]], int, list[str]]: by_hash: dict[str, list[Path]] = defaultdict(list) count = 0 errors: list[str] = [] if not root.is_dir(): raise RuntimeError(f"Target root does not exist or is not a directory: {root}") for path in sorted(root.rglob("*")): try: if path.is_symlink(): continue if not path.is_file(): continue digest = sha256_file(path) by_hash[digest].append(path) count += 1 except Exception as exc: errors.append(f"{path}\t{type(exc).__name__}: {exc}") return dict(by_hash), count, errors def read_manifest(path: Path) -> list[dict]: with path.open("r", encoding="utf-8", newline="") as f: rows = list(csv.DictReader(f, delimiter="\t")) required = {"origin", "relative_path", "effective_scope"} missing = required - set(rows[0].keys() if rows else []) if missing: raise RuntimeError( f"Manifest is missing required columns: {', '.join(sorted(missing))}" ) return rows def resolve_candidate(stage: Path, row: dict) -> Path: origin = row["origin"] rel = row["relative_path"] if origin == "SOURCE": return stage / "SOURCE" / rel if origin == "ARCHIVE": # Current non-work audit should not normally contain archive items, # but resolve them defensively if that changes later. return stage / "ARCHIVE_EXTRACTED" / "WORK" / rel raise RuntimeError(f"Unknown origin {origin!r} for {rel!r}") def join_paths(paths: list[Path]) -> str: return " | ".join(str(p) for p in paths) def main() -> int: p = argparse.ArgumentParser() p.add_argument("--stage", required=True, type=Path) p.add_argument("--manifest", required=True, type=Path) p.add_argument("--privat-root", type=Path, default=Path("/data/03_PRIVAT")) p.add_argument("--inbox-root", type=Path, default=Path("/data/01_INBOX")) p.add_argument("--report", required=True, type=Path) args = p.parse_args() stage = args.stage.resolve() manifest = args.manifest.resolve() report = args.report.resolve() rows = read_manifest(manifest) candidates = [ row for row in rows if row["effective_scope"] in {"03_PRIVAT", "01_INBOX"} ] print("=== NON-WORK EXACT-HASH AUDIT ===") print(f"Manifest rows: {len(rows)}") print(f"Candidates: {len(candidates)}") print(f" 03_PRIVAT: {sum(r['effective_scope'] == '03_PRIVAT' for r in candidates)}") print(f" 01_INBOX: {sum(r['effective_scope'] == '01_INBOX' for r in candidates)}") print() print("Hashing authoritative target trees (read-only)...") privat_idx, privat_count, privat_errors = index_tree(args.privat_root) inbox_idx, inbox_count, inbox_errors = index_tree(args.inbox_root) print(f"/data/03_PRIVAT files hashed: {privat_count}") print(f"/data/01_INBOX files hashed: {inbox_count}") print(f"Target hash errors: {len(privat_errors) + len(inbox_errors)}") results = [] candidate_errors = 0 for n, row in enumerate(candidates, 1): candidate = resolve_candidate(stage, row) scope = row["effective_scope"] result = { "effective_scope": scope, "origin": row["origin"], "relative_path": row["relative_path"], "filename": row.get("filename", candidate.name), "candidate_path": str(candidate), "sha256": "", "same_scope_status": "", "same_scope_matches": "", "cross_scope_status": "", "cross_scope_matches": "", "audit_status": "", "error": "", } try: if not candidate.is_file(): raise FileNotFoundError(candidate) digest = sha256_file(candidate) result["sha256"] = digest same_idx = privat_idx if scope == "03_PRIVAT" else inbox_idx cross_idx = inbox_idx if scope == "03_PRIVAT" else privat_idx same = same_idx.get(digest, []) cross = cross_idx.get(digest, []) result["same_scope_status"] = ( "EXACT_DUPLICATE" if same else "NOT_FOUND" ) result["same_scope_matches"] = join_paths(same) result["cross_scope_status"] = ( "EXACT_MATCH_OTHER_SCOPE" if cross else "NOT_FOUND" ) result["cross_scope_matches"] = join_paths(cross) if same: result["audit_status"] = "DUPLICATE_IN_DESTINATION" elif cross: result["audit_status"] = "MATCH_IN_OTHER_SCOPE_REVIEW" else: result["audit_status"] = "UNIQUE_VS_PRIVAT_AND_INBOX" except Exception as exc: candidate_errors += 1 result["audit_status"] = "ERROR" result["error"] = f"{type(exc).__name__}: {exc}" results.append(result) print( f"[{n:>2}/{len(candidates)}] " f"{scope:<9} " f"{result['audit_status']:<29} " f"{result['filename']}" ) report.parent.mkdir(parents=True, exist_ok=True) with report.open("w", encoding="utf-8", newline="") as f: fieldnames = list(results[0].keys()) if results else [] writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t") writer.writeheader() writer.writerows(results) print() print("=== SUMMARY ===") statuses = defaultdict(int) for r in results: statuses[r["audit_status"]] += 1 for key in ( "DUPLICATE_IN_DESTINATION", "MATCH_IN_OTHER_SCOPE_REVIEW", "UNIQUE_VS_PRIVAT_AND_INBOX", "ERROR", ): print(f"{key}: {statuses[key]}") print(f"Target hash errors: {len(privat_errors) + len(inbox_errors)}") print(f"Candidate errors: {candidate_errors}") print(f"Report: {report}") if privat_errors or inbox_errors or candidate_errors: print() print("AUDIT INCOMPLETE: errors occurred; do not proceed to routing.") return 2 print() print("AUDIT COMPLETE: no files were moved, renamed, deleted, or overwritten.") return 0 if __name__ == "__main__": raise SystemExit(main())