#!/usr/bin/env python3 """Extract the replies/ directory into a per-date and per-recipient volume census. LANE: ledger (Extract — data extraction and tables). EVIDENCE on every claim: * Each generated reply is a single Markdown file under /replies/, named -to-[-NN].md where is the check-in owner (e.g. zephyrus, merlin, volt) and NN is an optional per-recipient incident index. * Counted evidence is therefore the live file system listing of that directory; no reply body is read, and nothing is invented. Files that are NOT replies (excluded from the census): *.py probe scripts left by other lanes __pycache__/ compiled bytecode directory README.md repository index, not a response any .md that does not match the -to-[-NN].md template Run: python3 extract_reply_volume_census.py # writes reply-volume-census.tsv, reply-volume-by-recipient.tsv, # reply-volume-census.md (touching this script's directory) python3 extract_reply_volume_census.py --selftest # prints ground-truth counts for a quick cross-check """ import re import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path # wizard-work/ledger/ -> parents[0]=wizard-work, parents[1]=repo root REPO_ROOT = Path(__file__).resolve().parents[2] REPLIES_DIR = REPO_ROOT / "replies" OUT_DIR = REPO_ROOT / "wizard-work" / "ledger" # -to-[ -NN].md (recipient may contain hyphens; no digits) NAME_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})-to-([^0-9]+)(?:-\d+)?$") DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") def ground_truth(): """Derive counts from the live replies/ directory (the evidence source).""" entries = list(REPLIES_DIR.iterdir()) files = [p for p in entries if p.is_file()] non_md = [p.name for p in files if p.suffix.lower() not in (".md",)] non_canon = [] # .md files that do not match the canonical template by_date = Counter() by_recipient = Counter() for p in files: name = p.name if name.endswith(".md"): body = name[:-3] # strip the ".md" extension m = NAME_RE.match(body) if not m: non_canon.append(name) continue date, recipient = m.group(1), m.group(2) if not DATE_RE.match(date): non_canon.append(name) continue by_date[date] += 1 by_recipient[recipient] += 1 return by_date, by_recipient, non_canon, non_md, files def write_tsv(path, rows): with path.open("w", encoding="utf-8") as fh: fh.write("item\tcount\n") for item, count in rows: fh.write(f"{item}\t{count}\n") def write_markdown(path, by_date, by_recipient, non_canon, non_md, files): total = sum(by_date.values()) dates = sorted(by_date) recip = sorted(by_recipient, key=lambda r: (-by_recipient[r], r)) lines = [] lines.append("# Replies volume census") lines.append("") lines.append(f"- Extracted at UTC: `{datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}`") lines.append(f"- Source: `{REPLIES_DIR}` (live file system listing)") lines.append(f"- Reply files matched `-to-[-NN].md`: **{total}**") lines.append(f"- Excluded non-reply files: **{len(non_md) + len(non_canon)}**") lines.append(f" - `.md` not matching the template: **{len(non_canon)}** (`{'`; `'.join(sorted(non_canon))}`)") lines.append(f" - non-`.md` entries: **{len(non_md)}**") lines.append(f"- Distinct dates covered: **{len(dates)}**") lines.append(f"- Distinct recipients: **{len(recip)}**") lines.append("") lines.append("## Evidence on what was counted") lines.append("") lines.append("Every reply is a single Markdown file carrying `-to-` routing and a four-part") lines.append("date header. The directory also holds `README.md` (this index), three probe") lines.append("scripts and a `__pycache__` directory; none is a reply and all are excluded.") lines.append("") lines.append("## Replies by date") lines.append("") lines.append("| date | replies |") lines.append("|---|---|") for d in dates: lines.append(f"| {d} | {by_date[d]} |") lines.append(f"| **TOTAL** | **{total}** |") lines.append("") lines.append("## Replies by recipient") lines.append("") lines.append("| recipient | replies |") lines.append("|---|---|") for r in recip: lines.append(f"| {r} | {by_recipient[r]} |") lines.append(f"| **TOTAL** | **{total}** |") lines.append("") lines.append("## Excluded files (.md not matching template)") lines.append("") if non_canon: for name in sorted(non_canon): lines.append(f"| {name} |") lines.append("") lines.append("## Excluded (all file types)") lines.append("") lines.append("| kind | count |") lines.append("|---|---|") if non_md: for k in sorted(non_md): lines.append(f"| `{k}` | **1** |") lines.append("") lines.append("## Method") lines.append("") lines.append("1. `ls ` enumerates every entry (evidence is the live FS).") lines.append("2. Replies are files whose name matches `YYYY-MM-DD-to-RECEIVER[-NN].md`.") lines.append("3. Per-date and per-recipient counters are aggregated with `collections.Counter`.") lines.append("4. The committed verifier `verify_reply_volume_census.py` re-derives ground truth") lines.append(" from the live directory with independent shell commands and asserts the files") lines.append(" above match.") lines.append("") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines), encoding="utf-8") print(f"wrote {path}") def main(): by_date, by_recipient, non_canon, non_md, files = ground_truth() total = sum(by_date.values()) dates = sorted(by_date) recip = sorted(by_recipient, key=lambda r: (-by_recipient[r], r)) print(f"reply files matched -to-[-NN].md : {total}") print(f"distinct dates : {len(dates)}") print(f"distinct recipients : {len(recip)}") print(f"non-canonical .md excluded : {len(non_canon)}") print(f"non-md entries excluded : {len(non_md)}") OUT_DIR.mkdir(parents=True, exist_ok=True) write_tsv(OUT_DIR / "reply-volume-census.tsv", [(d, by_date[d]) for d in dates] + [("TOTAL", total)]) write_tsv(OUT_DIR / "reply-volume-by-recipient.tsv", [(r, by_recipient[r]) for r in recip] + [("TOTAL", total)]) write_markdown(OUT_DIR / "reply-volume-census.md", by_date, by_recipient, non_canon, non_md, files) if "--selftest" in sys.argv: print(f"TOTAL={total}") print(f"DATES={len(dates)}") print(f"RECIPIENTS={len(recip)}") print("BY_RECIPIENT:") for r in recip: print(f" {r}\t{by_recipient[r]}") if __name__ == "__main__": main()