230 lines
7.1 KiB
Python
230 lines
7.1 KiB
Python
#!/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())
|