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

#!/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()