troupe-standup/wizard-work/ledger/2026-10-29-checkin-census/gen.py

#!/usr/bin/env python3
"""Generator: check-in signee census, derived from a PINNED revision.

Extracts every date-tagged check-in file at `<pin>:checkins/`, resolving the
signee handle, the committer, and the headline, then emits a table (CSV/JSON/MD).

Evidence source: `git ls-tree <pin> -- checkins/` (the authoritative file list at
that revision) and `git log -1 <pin> --format=... -- <file>` (the committer of
the last commit touching the file at that revision). Nothing is invented: every
row is derived from a real blob and a real git object, and no host path is read
or written -- the first version hardcoded `/root/troupe-standup` for both ROOT
and OUT and walked `git ls-files`, which is the working tree, not a commit.

The scope is the revision, so this generator is reproducible: run at the pin
that carries the committed tables (`b21bb10`), it rewrites them byte for byte.
It writes nothing without `--write`.

Usage:
  gen.py --pin b21bb10                 # dry run: row counts only
  gen.py --pin b21bb10 --write         # publish into this artifact's directory
  gen.py --pin b21bb10 --outdir DIR --write
Exit: 0 derived/published, 1 the pin cannot be read, 2 bad arguments.
"""
import argparse
import csv
import json
import re
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
DATE = re.compile(r"^checkins/(\d{4}-\d{2}-\d{2})-(.+)$")
INDEX_FILE = "checkins/README.md"
TAG = "2026-10-29"


def repo_root():
    try:
        out = subprocess.run(["git", "-C", str(HERE), "rev-parse", "--show-toplevel"],
                             capture_output=True, text=True, timeout=60)
        if out.returncode == 0 and out.stdout.strip():
            return Path(out.stdout.strip())
    except (OSError, subprocess.SubprocessError):
        pass
    return HERE.parent.parent.parent


def run_git(repo, args, binary=False):
    out = subprocess.run(["git", "-C", str(repo)] + args,
                         capture_output=True, text=not binary, timeout=120)
    if out.returncode != 0:
        err = out.stderr if not binary else (out.stderr or b"").decode("utf-8", "replace")
        raise RuntimeError("git %s failed: %s" % (" ".join(args), (err or "").strip()))
    return out.stdout


def pinned_checkin_files(repo, rev):
    out = run_git(repo, ["ls-tree", "-r", "--name-only", rev, "--", "checkins/"])
    return sorted(f for f in out.splitlines() if f.endswith(".md"))


def signee_from_filename(name):
    m = DATE.match(name)
    token = m.group(2)
    token = token[:-3] if token.endswith(".md") else token
    return re.sub(r"-\d+$", "", token)  # drop -02/-03 revision markers


def headline(repo, rev, path):
    text = run_git(repo, ["show", "%s:%s" % (rev, path)], binary=True)
    for line in text.decode("utf-8", "replace").split("\n"):
        line = line.rstrip("\r")
        if line.startswith("# "):
            return line[2:].strip()
    return ""


def committer(repo, rev, path):
    out = run_git(repo, ["log", "-1", "--format=%an <%ae>", rev, "--", path]).strip()
    return out or "(no committer)"


def derive(repo, rev):
    rows = []
    for f in pinned_checkin_files(repo, rev):
        if f == INDEX_FILE:
            rows.append({
                "file": INDEX_FILE,
                "date": TAG,
                "signee": "troupe-pipe (index)",
                "headline": "Check-ins index",
                "committer": "(index file; no committer)",
                "kind": "index",
            })
            continue
        m = DATE.match(f)
        rows.append({
            "file": f,
            "date": m.group(1),
            "signee": signee_from_filename(f),
            "headline": headline(repo, rev, f),
            "committer": committer(repo, rev, f),
            "kind": "checkin",
        })
    rows.sort(key=lambda r: (r["date"], r["file"]))
    return rows


def write_outputs(out_dir, rows):
    out_dir.mkdir(parents=True, exist_ok=True)
    csv_path = out_dir / "checkin-census.csv"
    with open(csv_path, "w", newline="", encoding="utf-8") as fh:
        w = csv.DictWriter(fh, fieldnames=["file", "date", "signee", "headline",
                                           "committer", "kind"])
        w.writeheader()
        for r in rows:
            w.writerow(r)

    json_path = out_dir / "checkin-census.json"
    with open(json_path, "w", encoding="utf-8") as fh:
        json.dump({
            "source": "git ls-files checkins/ (authoritative tracked list)",
            "total_checkin_rows": len([r for r in rows if r["kind"] == "checkin"]),
            "rows": rows,
        }, fh, indent=2, ensure_ascii=False)

    md_path = out_dir / "checkin-census.md"
    with open(md_path, "w", encoding="utf-8") as fh:
        fh.write("# Check-in signee census (%s)\n\n" % TAG)
        fh.write("Source: `git ls-files checkins/`. Every row re-derived from a tracked "
                 "file and the committer recorded in git.\n\n")
        fh.write("| file | date | signee | headline | committer | kind |\n")
        fh.write("|---|---|---|---|---|---|\n")
        for r in rows:
            fh.write(f"| {r['file']} | {r['date']} | {r['signee']} | {r['headline']} | "
                     f"{r['committer']} | {r['kind']} |\n")
    return csv_path, json_path, md_path


def main(argv=None):
    ap = argparse.ArgumentParser(description="derive the check-in signee census "
                                             "from a pinned git revision")
    ap.add_argument("--pin", required=True, help="revision whose checkins/ is the scope")
    ap.add_argument("--outdir", default=None,
                    help="directory to write into (default: this artifact's directory)")
    ap.add_argument("--write", action="store_true",
                    help="publish; without it the run only derives and reports")
    args = ap.parse_args(argv)

    repo = repo_root()
    try:
        rev = run_git(repo, ["rev-parse", "--verify", args.pin + "^{commit}"]).strip()
        rows = derive(repo, rev)
    except RuntimeError as exc:
        print("VALIDATION_FAILED: %s" % exc, file=sys.stderr)
        return 1
    except (OSError, subprocess.SubprocessError, ValueError, AttributeError) as exc:
        print("CANNOT GRADE: %s: %s" % (type(exc).__name__, exc), file=sys.stderr)
        return 2

    dated = len([r for r in rows if r["kind"] == "checkin"])
    print("pin=%s rows=%d dated=%d index=%d" % (rev[:12], len(rows), dated, len(rows) - dated))
    if not args.write:
        print("dry run: nothing written (pass --write to publish)")
        return 0
    out = Path(args.outdir) if args.outdir else HERE
    for p in write_outputs(out, rows):
        print("wrote:", p)
    return 0


if __name__ == "__main__":
    sys.exit(main())