delivery/bundle.py

"""
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}]
  #     <every file from the pilot dir, at its original relative path>

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="<action>")

    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())