ledger: corrected workqueue census v2 (96 rows, independent CHECK)

wizard-work/ledger/workqueue-census-v2/extract_workqueue.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Corrected, complete census of every numbered work-queue (Q) item.
+
+Extract lane: data extraction and tables. One row per "## Q<n> ..." heading in
+WORK-QUEUE.md, built from the source with a real parser and verified by an
+INDEPENDENT re-derivation in CHECK (the old census's CHECK re-imported its own
+extractor, so it could never fail; that is the fault this version removes).
+
+Output (all in this directory):
+  workqueue-census.tsv     one row per Q heading, tab-separated
+  workqueue-census.md      markdown table of the same rows
+  extract_manifest.json    schema + counts + provenance (extracted_at_utc = NOW)
+  README.md                documentation for this artifact
+  CHECK                  real verification (independent re-derivation)
+
+Conventions
+-----------
+* A row is created for every line matching '## <Qnum> ...' where the '<Qnum>' is
+  'Q' followed by digits (continuation labels such as 'closure', 'section 9
+  follow-on', 'verified', 'claimed' are kept verbatim in the title).
+* q        : the label 'Q<num>' (dupes kept as separate rows).
+* title    : verbatim heading text after '## <Qnum>' (trimmed).
+* status   : the first '**Status:' line in the block, whitespace collapsed,
+             truncated at the first sentence end, single trailing period;
+             empty when no such marker precedes the next heading.
+* done_implies : the first '**Done means:' line in the block, trimmed; empty
+             when no such marker precedes the next heading.
+"""
+
+import json
+import os
+import re
+from datetime import datetime, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
+SRC = os.path.join(REPO_ROOT, "WORK-QUEUE.md")
+OUT = HERE
+
+# '## Q' + digits, then anything (heading title begins). Captures Q1..Q98
+# headings, including continuation/closure labels that follow the number.
+HEADING_RE = re.compile(r"^##\s+Q(\d+)(.*)$")
+
+STATUS_RE = re.compile(r"^\*\*Status:\s*(.*)$")
+DONE_RE = re.compile(r"^\*\*Done means:\s*(.*)$")
+
+
+def _sentence(s):
+    s = re.sub(r"\s+", " ", s).strip()
+    m = re.search(r"\.\s", s)
+    if m:
+        s = s[: m.start()] + "."
+    if s.endswith("."):
+        s = s[: -1]
+    return s + "."
+
+
+def split_blocks(lines):
+    """Return list of dicts: idx (heading line number), qid, title, body.
+
+    A 'block' is the heading line plus the following lines up to (but not
+    including) the next '## Q<n>' heading line.
+    """
+    blocks = []
+    current = None
+    for i, line in enumerate(lines):
+        m = HEADING_RE.match(line)
+        if m:
+            if current is not None:
+                blocks.append(current)
+            current = {
+                "line": i,
+                "qid": "Q" + m.group(1),
+                "title": m.group(2).strip(),
+                "body": [],
+            }
+        else:
+            if current is not None:
+                current["body"].append((i, line))
+    if current is not None:
+        blocks.append(current)
+    return blocks
+
+
+def derive(block):
+    status = ""
+    done_implies = ""
+    for _, line in block["body"]:
+        if status == "":
+            sm = STATUS_RE.match(line)
+            if sm:
+                status = _sentence(sm.group(1))
+        if done_implies == "":
+            dm = DONE_RE.match(line)
+            if dm:
+                done_implies = dm.group(1).strip()
+    return status, done_implies
+
+
+def extract(lines):
+    rows = []
+    for block in split_blocks(lines):
+        status, done_implies = derive(block)
+        rows.append(
+            {
+                "q": block["qid"],
+                "title": block["title"],
+                "status": status,
+                "done_implies": done_implies,
+                "line": block["line"] + 1,
+            }
+        )
+    return rows
+
+
+def main():
+    with open(SRC, encoding="utf-8") as fh:
+        lines = fh.read().splitlines()
+
+    rows = extract(lines)
+
+    tsv_path = os.path.join(OUT, "workqueue-census.tsv")
+    md_path = os.path.join(OUT, "workqueue-census.md")
+    json_path = os.path.join(OUT, "extract_manifest.json")
+
+    with open(tsv_path, "w", encoding="utf-8") as fh:
+        fh.write("q\ttitle\tstatus\tdone_implies\n")
+        for r in rows:
+            fh.write(
+                f"{r['q']}\t{r['title']}\t{r['status']}\t{r['done_implies']}\n"
+            )
+
+    distinct = sorted({r["q"] for r in rows})
+    reopened = sorted(q for q in distinct if sum(1 for r in rows if r["q"] == q) > 1)
+
+    with open(md_path, "w", encoding="utf-8") as fh:
+        fh.write("# Work queue census v2 (extract lane)\n\n")
+        fh.write(
+            "One row per '## Q<n> ...' heading in WORK-QUEUE.md. Extracted from "
+            f"{len(rows)} headings; {len(distinct)} distinct q labels; "
+            f"{len(reopened)} reopened ({', '.join(reopened)}). "
+            "extracted_at_utc is the measurement moment, not a constant.\n\n"
+        )
+        fh.write("| q | title | status | done_implies |\n")
+        fh.write("|---|---|---|---|\n")
+        for r in rows:
+            fh.write(f"| {r['q']} | {r['title']} | {r['status']} | {r['done_implies']} |\n")
+
+    manifest = {
+        "schema": "workqueue-census-v2",
+        "source": os.path.abspath(SRC),
+        "rows": len(rows),
+        "columns": ["q", "title", "status", "done_implies"],
+        "distinct_q_labels": len(distinct),
+        "reopened_q_labels": reopened,
+        "extracted_at_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "generated_by": "extract_workqueue.py",
+        "verification": "CHECK (independent re-derivation, does not import extractor)",
+    }
+    with open(json_path, "w", encoding="utf-8") as fh:
+        json.dump(manifest, fh, indent=2)
+
+    print(f"rows: {len(rows)}")
+    print(f"distinct labels: {len(distinct)}; reopened: {reopened}")
+    print(f"written: {tsv_path}, {md_path}, {json_path}")
+
+
+if __name__ == "__main__":
+    main()