troupe-standup/wizard-work/ledger/2026-10-28-checkin-census/extract_checkin_census_2026-10-28.py

#!/usr/bin/env python3
"""Extract the daily check-in census from `checkins/` into tables.

Reads every dated check-in markdown file under `checkins/` (filename
`YYYY-MM-DD-<signer>.md`), detects which of the four ledger sections
(`## Did`, `## Blocked`, `## Next`, `## Ask`) each file carries, and counts
the numbered `### N.` items under `## Did`. Emits:

  checkin-census-<YYYY-MM-DD>.tsv   - machine table (date, signer, file, flags, counts)
  checkin-census-<YYYY-MM-DD>.md    - human markdown table + per-signer summary
  checkin-census-<YYYY-MM-DD>.json  - structured rows + headline

Exit status is 0 only when every dated check-in parsed, every date is valid,
and the rows reconcile with the file inventory (no gaps, no dupes). Any
failure is a nonzero exit.

Usage:
  extract_checkin_census_2026-10-28.py [checkins_dir] [out_dir]
"""

import csv
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path

DEFAULT_CHECKINS = Path("/root/troupe-standup/checkins")
DEFAULT_OUT = Path("/root/troupe-standup/wizard-work/ledger") / "2026-10-28-checkin-census"

SECTION_RE = re.compile(r"^## (Did|Blocked|Next|Ask)\s*$", re.MULTILINE)
ITEM_RE = re.compile(r"^\s*### \d+\.\s+\S", re.M)
FILENAME_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})-(.+)$")


def parse_name(name):
    """Return (date, signer) from a check-in filename; None on any failure."""
    if not name.endswith(".md"):
        return None
    body = name[:-3]
    m = FILENAME_RE.match(body)
    if not m:
        return None
    date_str, signer = m.group(1), m.group(2)
    try:
        datetime.strptime(date_str, "%Y-%m-%d")
    except ValueError:
        return None
    return date_str, signer


def section_bounds(text):
    """Return ordered list of (section_name, start, end) offsets."""
    bounds = []
    for m in SECTION_RE.finditer(text):
        bounds.append((m.group(1), m.start(), m.end()))
    return bounds


def section_text(text, name):
    """Text of one section (up to the next section, or EOF)."""
    bounds = section_bounds(text)
    for i, (n, start, end) in enumerate(bounds):
        if n != name:
            continue
        nxt = bounds[i + 1][1] if i + 1 < len(bounds) else len(text)
        return text[start:nxt]
    return ""


def analyze(text):
    """Return presence flags + did-item count + did chars for a file."""
    present = {name: False for name in ("Did", "Blocked", "Next", "Ask")}
    for n, _start, _end in section_bounds(text):
        present[n] = True
    did = section_text(text, "Did")
    return {
        "has_done": present["Did"],
        "has_blocked": present["Blocked"],
        "has_next": present["Next"],
        "has_ask": present["Ask"],
        "did_items": len(ITEM_RE.findall(did)),
        "did_chars": len(did),
    }


def parse_checkins(checkins_dir):
    rows = []
    skipped = []
    for path in sorted(checkins_dir.iterdir()):
        if not path.is_file():
            continue
        parsed = parse_name(path.name)
        if parsed is None:
            skipped.append(path.name)
            continue
        date_str, signer = parsed
        text = path.read_text(encoding="utf-8", errors="replace")
        rows.append({
            "date": date_str,
            "signer": signer,
            "file": path.name,
            "abs": str(path),
            **analyze(text),
        })
    return rows, skipped


def latest_per_signer(rows):
    out = {}
    for r in rows:
        key = r["signer"]
        cur = out.get(key)
        if cur is None or r["date"] > cur["date"]:
            cur = {"signer": key, "latest_date": r["date"], "file": r["file"]}
            out[key] = cur
    return sorted(out.values(), key=lambda x: x["signer"])


def write_outputs(out_dir, rows, skipped):
    out_dir.mkdir(parents=True, exist_ok=True)
    tag = os.environ.get("LIFT_DATE", datetime.now().strftime("%Y-%m-%d"))
    csv_path = out_dir / ("checkin-census-" + tag + ".tsv")
    json_path = out_dir / ("checkin-census-" + tag + ".json")
    md_path = out_dir / ("checkin-census-" + tag + ".md")

    fieldnames = ["date", "signer", "file", "has_done", "has_blocked", "has_next", "has_ask", "did_items", "did_chars"]
    with open(csv_path, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t")
        w.writeheader()
        for r in rows:
            w.writerow({k: r.get(k, "") for k in fieldnames})

    json.dump({
        "source": str(DEFAULT_CHECKINS),
        "out_dir": str(out_dir),
        "generated": tag,
        "rows": rows,
        "skipped": skipped,
        "headline": {
            "files": len(rows),
            "skipped_non_dated": len(skipped),
            "signers": len(set(r["signer"] for r in rows)),
            "total_did_items": sum(r["did_items"] for r in rows),
            "total_did_chars": sum(r["did_chars"] for r in rows),
            "latest_per_signer": latest_per_signer(rows),
        },
    }, open(json_path, "w", encoding="utf-8"), indent=2)

    lines = []
    w = lines.append
    w("# Check-in census - " + tag)
    w("")
    w("*source*: every `YYYY-MM-DD-<signer>.md` file under `checkins/` "
      "(non-dated files like README.md skipped). Per-row section flags and "
      "did-item counts; `latest_per_signer` = newest dated check-in per signer.")
    w("")
    w("## Headline")
    w("")
    w("| measure | value |")
    w("| --- | --- |")
    w("| dated check-in files | {} |".format(len(rows)))
    w("| non-dated files skipped | {} |".format(len(skipped)))
    w("| signers | {} |".format(len(set(r['signer'] for r in rows))))
    w("| total did items | {} |".format(sum(r['did_items'] for r in rows)))
    w("| total did chars | {} |".format(sum(r['did_chars'] for r in rows)))
    w("")
    w("## Per-signer latest check-in")
    w("")
    w("| signer | latest check-in | file |")
    w("| --- | --- | --- |")
    for s in latest_per_signer(rows):
        w("| {} | {} | `{}` |".format(s["signer"], s["latest_date"], s["file"]))
    w("")
    w("## Full inventory")
    w("")
    w("| date | signer | file | Did | Blocked | Next | Ask | did items | did chars |")
    w("| --- | --- | --- | --- | --- | --- | --- | --- | --- |")
    for r in sorted(rows, key=lambda x: (x["date"], x["signer"])):
        w("| {} | {} | `{}` | {} | {} | {} | {} | {} | {} |".format(
            r["date"], r["signer"], r["file"],
            "1" if r["has_done"] else "0",
            "1" if r["has_blocked"] else "0",
            "1" if r["has_next"] else "0",
            "1" if r["has_ask"] else "0",
            r["did_items"], r["did_chars"]))
    w("")
    md_path.write_text("\n".join(lines), encoding="utf-8")
    return csv_path, json_path, md_path


def main():
    checkins = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_CHECKINS
    out_dir = Path(sys.argv[2]) if len(sys.argv) > 2 else DEFAULT_OUT
    rows, skipped = parse_checkins(checkins)

    # Invariants the extractor stands behind.
    assert rows, "no dated check-ins found"
    assert len({r["file"] for r in rows}) == len(rows), "duplicate file"
    for r in rows:
        datetime.strptime(r["date"], "%Y-%m-%d")

    csv_path, json_path, md_path = write_outputs(out_dir, rows, skipped)

    print("files={} skipped={} signers={} rows={}".format(
        len(rows), len(skipped), len(set(r['signer'] for r in rows)), len(rows)))
    print("wrote: csv={} json={} md={}".format(csv_path, json_path, md_path))
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except AssertionError as exc:
        print("VALIDATION_FAILED: {}".format(exc), file=sys.stderr)
        sys.exit(1)
    except Exception as exc:  # noqa: BLE001
        print("EXTRACTION_FAILED: {}".format(exc), file=sys.stderr)
        sys.exit(1)