""" delivery.bundle — Stage 68: package a pilot output directory into a self-describing zip for handoff. A pilot run produces 5-7 HTML files + PDF + an evidence subdir. Mailing the dir or pointing at a URL is fine for tech audiences; for prospects/auditors/customer-success packaging it as ONE file with a manifest is the right shape: bundle_pilot_output("/tmp/pilot_output") # → /tmp/pilot_output.zip with: # manifest.json — generated_at, files: [{path, sha256, bytes}] # The manifest lets the recipient verify the bundle is intact without us needing to ship a separate checksum file. Stdlib only — no external archive deps. """ from __future__ import annotations import hashlib import json import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Optional def _sha256_of_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def bundle_pilot_output(output_dir, bundle_path: Optional[Path] = None) -> Path: """Zip ``output_dir`` recursively + add a manifest.json. Returns the path to the resulting bundle (next to the input dir by default, with ``.zip`` suffix). Raises ``FileNotFoundError`` if ``output_dir`` doesn't exist — callers should ``run_pilot()`` first. """ src = Path(output_dir).resolve() if not src.is_dir(): raise FileNotFoundError( f"pilot output directory does not exist: {src}" ) dst = Path(bundle_path) if bundle_path is not None \ else src.with_suffix(".zip") dst = dst.resolve() # gather every file under src, ordered for deterministic manifests files = sorted([p for p in src.rglob("*") if p.is_file()]) manifest_entries = [] for f in files: body = f.read_bytes() rel = f.relative_to(src).as_posix() manifest_entries.append({ "path": rel, "sha256": _sha256_of_bytes(body), "bytes": len(body), }) manifest = { "generated_at": datetime.now(timezone.utc).isoformat(), "source_dir_name": src.name, "n_files": len(manifest_entries), "total_bytes": sum(e["bytes"] for e in manifest_entries), "files": manifest_entries, } # write the zip — manifest first so unzip tools surface it on top with zipfile.ZipFile(dst, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr("manifest.json", json.dumps(manifest, indent=2, sort_keys=True)) for f in files: rel = f.relative_to(src).as_posix() zf.write(f, arcname=rel) return dst def _cli(argv=None) -> int: """Stage 70: shell-friendly wrapper. Two verbs: python -m delivery.bundle create DIR [--out PATH] python -m delivery.bundle verify ZIP Recipient who got a bundle.zip in an email can run `verify` in one shot — no Python snippet needed. Exit codes: 0 — create succeeded OR verify passed (manifest matches) 1 — verify failed (one or more sha256 mismatches) 2 — bad args / file not found / argparse error """ import argparse import sys parser = argparse.ArgumentParser(prog="python -m delivery.bundle") sub = parser.add_subparsers(dest="action", required=True, metavar="") p_create = sub.add_parser( "create", help="zip a directory + write a manifest.json (Stage 68)", ) p_create.add_argument("source_dir", help="directory to bundle") p_create.add_argument("--out", default=None, help="output path for the zip (default: " "alongside source_dir with .zip suffix)") p_verify = sub.add_parser( "verify", help="re-hash a bundle and compare against its manifest", ) p_verify.add_argument("bundle_path", help="path to a bundle .zip") args = parser.parse_args(argv) if args.action == "create": try: out = bundle_pilot_output( args.source_dir, bundle_path=args.out, ) except FileNotFoundError as e: print(json.dumps({"error": str(e)}), file=sys.stderr) return 2 print(json.dumps({"bundle_path": str(out)}, indent=2)) return 0 # verify try: report = verify_bundle(args.bundle_path) except FileNotFoundError as e: print(json.dumps({"error": str(e)}), file=sys.stderr) return 2 except zipfile.BadZipFile as e: # not a zip — treat as bad-args so the user knows to point # at a real bundle print(json.dumps({ "error": f"{args.bundle_path!r} is not a valid zip: {e}", }), file=sys.stderr) return 2 print(json.dumps(report, indent=2, sort_keys=True)) return 0 if report["ok"] else 1 def verify_bundle(bundle_path) -> dict: """Re-hash every file in ``bundle_path`` and check it matches the manifest's sha256. Returns ``{ok, n_files, mismatches}``. Operationally useful for the recipient: 'hand me a verified bundle' becomes one function call instead of unzip+sha256sum per file.""" path = Path(bundle_path).resolve() mismatches = [] with zipfile.ZipFile(path, "r") as zf: manifest = json.loads(zf.read("manifest.json")) for entry in manifest["files"]: data = zf.read(entry["path"]) actual = _sha256_of_bytes(data) if actual != entry["sha256"]: mismatches.append({ "path": entry["path"], "expected": entry["sha256"], "actual": actual, }) return { "ok": not mismatches, "n_files": manifest["n_files"], "n_mismatches": len(mismatches), "mismatches": mismatches, } if __name__ == "__main__": # pragma: no cover raise SystemExit(_cli())