troupe-standup/wizard-work/ledger/verify_ledger_census.py

#!/usr/bin/env python3
# Ledger — independent re-derivation + comparison gate for the census.
#
# Re-walks the live tree with os.stat, classifies each entry with its own copy
# of the classification rules, and compares against the persisted
# ledger-census-2026-10-28.{md,csv,json}. Exit 0 iff they agree field-for-field.
# This is a genuine cross-derivation (a second implementation + a fresh walk),
# not a self-referential equality.
import os
import sys
import csv
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path("/root/troupe-standup/wizard-work/ledger")
MD = ROOT / "ledger-census-2026-10-28.md"
CSV = ROOT / "ledger-census-2026-10-28.csv"
JSON = ROOT / "ledger-census-2026-10-28.json"
EXCLUDE_DIRS = {"__pycache__", ".git"}
EXCLUDED_NAMES = {
    "build_ledger_census.py", "verify_ledger_census.py",
    "ledger-census-2026-10-28.md", "ledger-census-2026-10-28.csv",
    "ledger-census-2026-10-28.json",
}
RULES = [
    ("script", {".py", ".sh"}),
    ("config", {".yaml", ".yml", ".json", ".toml", ".env", ".lock", ".ini", ".pem"}),
    ("doc", {".md", ".txt"}),
    ("data", {".csv", ".tsv", ".xml", ".db", ".sqlite"}),
    ("media", {".mp3", ".wav", ".png", ".jpg", ".jpeg", ".etag"}),
]


def classify(name):
    suffix = os.path.splitext(name)[1].lower()
    for cat, exts in RULES:
        if suffix in exts:
            return cat
    return "other"


def rederive():
    files = []
    dirs = []
    for dirpath, dirnames, filenames in os.walk(ROOT):
        dirnames.sort()
        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDE_DIRS)
        for fn in sorted(filenames):
            p = Path(dirpath) / fn
            if p.name in EXCLUDED_NAMES:
                continue
            st = p.stat()
            files.append({"type": "file", "rel": p.relative_to(ROOT).as_posix(),
                          "size": st.st_size,
                          "mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
                          "cat": classify(fn)})
        for dn in dirnames:
            dirs.append({"type": "dir", "rel": os.path.relpath(str(Path(dirpath) / dn), str(ROOT)),
                         "size": 0, "mtime": "", "cat": "dir"})
    total_bytes = sum(f["size"] for f in files)
    cat_counts = Counter(f["cat"] for f in files)
    return files, dirs, len(files), len(dirs), total_bytes, cat_counts


def build_text(files, dirs, total_bytes, cat_counts):
    lines = []
    w = lines.append
    w = lines.append
    w("# wizard-work/ledger census — 2026-10-28")
    w("")
    w("*source*: os.walk + os.stat over `{0}` (no symlink following)".format(ROOT))
    w("*classification*: extension-based per CATEGORY_RULES; row count = files + dirs")
    w("")
    w("## Headline")
    w("")
    w("| measure | value |")
    w("| --- | --- |")
    w("| files | {} |".format(len(files)))
    w("| subdirectories | {} |".format(len(dirs)))
    w("| total entries | {} |".format(len(files) + len(dirs)))
    w("| total bytes | {:,} |".format(total_bytes))
    w("| categories | {} |".format(len(cat_counts)))
    w("")
    w("## Category rollup")
    w("")
    w("| category | files | bytes |")
    w("| --- | --- | --- |")
    for cat in sorted(cat_counts):
        b = sum(f["size"] for f in files if f["cat"] == cat)
        w("| {} | {} | {:,} |".format(cat, cat_counts[cat], b))
    w("")
    w("## File inventory")
    w("")
    w("| category | relpath | size | mtime(UTC) |")
    w("| --- | --- | --- | --- |")
    for f in sorted(files, key=lambda x: (x["cat"], x["rel"])):
        w("| {} | `{}` | {} | {} |".format(f["cat"], f["rel"], f["size"], f["mtime"]))
    w("")
    return "\n".join(lines)


def build_csv(files):
    rows = [["type", "relpath", "size_bytes", "mtime_utc", "category"]]
    for f in sorted(files, key=lambda x: (x["cat"], x["rel"])):
        rows.append([f["type"], f["rel"], str(f["size"]), f["mtime"], f["cat"]])
    # simple join; catalogue names contain no commas.
    return "\n".join(",".join(r) for r in rows)


def main():
    errors = []

    # Independent re-derivation.
    files, dirs, n_files, n_dirs, total_bytes, cat_counts = rederive()

    # Compare JSON (structured).
    try:
        persisted = json.loads(JSON.read_text(encoding="utf-8"))
    except Exception as e:
        errors.append("JSON read/parse failed: {!r}".format(e))
        persisted = None
    if persisted is not None:
        expected = {
            "source": str(ROOT),
            "classification_rules": {c: sorted(e) for c, e in RULES},
            "excluded_dirs": sorted(EXCLUDE_DIRS),
            "excluded_names": sorted(EXCLUDED_NAMES),
            "counts": {"files": n_files, "dirs": n_dirs, "total_entries": n_files + n_dirs,
                       "total_bytes": total_bytes},
            "category_counts": {c: cat_counts.get(c, 0) for c, _ in RULES + [("other", set())]},
            "files": files,
            "dirs": dirs,
        }
        # category_counts key order noise: normalise by comparing as dicts.
        if persisted.get("counts") != expected["counts"]:
            errors.append("JSON counts mismatch)")
        if persisted.get("category_counts") != expected["category_counts"]:
            errors.append("JSON category_counts mismatch")
        if persisted.get("files") != expected["files"]:
            errors.append("JSON files mismatch")
        if persisted.get("dirs") != expected["dirs"]:
            errors.append("JSON dirs mismatch")

    # Compare CSV.
    expected_csv = build_csv(files) + "\n"
    try:
        persisted_csv = CSV.read_text(encoding="utf-8")
    except Exception as e:
        errors.append("CSV read failed: {!r}".format(e))
        persisted_csv = None
    if persisted_csv != expected_csv:
        errors.append("CSV content mismatch")

    # Compare markdown.
    expected_md = build_text(files, dirs, total_bytes, cat_counts)
    try:
        persisted_md = MD.read_text(encoding="utf-8")
    except Exception as e:
        errors.append("MD read failed: {!r}".format(e))
        persisted_md = None
    if persisted_md != expected_md:
        errors.append("MD content mismatch")

    if errors:
        print("VERIFY FAIL:")
        for e in errors:
            print(" -", e)
        return 1
    print("VERIFY OK — files={} dirs={} bytes={:,} cats={}".format(
        n_files, n_dirs, total_bytes, dict(sorted(cat_counts.items()))))
    return 0


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