skills/diligence/scripts/export_register.py

#!/usr/bin/env python3
# ruff: noqa: E501 -- lawyer-facing register prompts remain readable inline.
"""Export the further-enquiries register as register.csv plus register.json.

Usage:
    python3 export_register.py --findings findings.json --gaps gap-report.json \
        --families families.json --manifest manifest.json \
        [--framework framework.json] --out register.csv

One row per underlying ask: a missing document (gap entries of type
index-missing or referenced-absent), an unresolved finding (the clarification
its unresolved_when rule implies; --framework supplies the rule text, else the
finding's own characterization stands in), or a cross-document inconsistency
(findings on one issue within one family with conflicting present and absent
statuses). Rows deduplicate on (category, ask); source_refs merge. Ordering is
deterministic by (category, register_id). register.json is written beside the
CSV with the same rows.
"""

import argparse
import csv
import hashlib
import json
import os
import sys

MISSING_GAP_TYPES = {"index-missing", "referenced-absent"}
CSV_COLUMNS = ["register_id", "category", "ask", "evidence", "source_refs"]


def load(path, kind, required_keys):
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError) as e:
        sys.exit(f"export_register: cannot read {kind} at {path}: {e}")
    for k in required_keys:
        if k not in data:
            sys.exit(f"export_register: {kind} missing key '{k}'")
    return data


def register_id(category, source_refs):
    h = hashlib.sha256(f"{category}|{source_refs}".encode()).hexdigest()[:12]
    return f"{category}-{h}"


def doc_path(doc_id, docs):
    d = docs.get(doc_id)
    return d.get("path", doc_id) if d else doc_id


def issue_lookup(framework):
    """issue_id -> (question, unresolved_when). Empty when no framework given."""
    table = {}
    if not framework:
        return table
    for lens in framework.get("lenses", []):
        for item in lens.get("items", []):
            table[item.get("issue_id", "")] = (
                item.get("question", ""),
                item.get("evidence", {}).get("unresolved_when", ""),
            )
    return table


def one_sentence(s):
    s = " ".join(str(s).split()).strip()
    if s and not s.endswith("."):
        s += "."
    return s


def gap_rows(gaps):
    rows = []
    entries = sorted(
        (e for e in gaps.get("entries", []) if e.get("type") in MISSING_GAP_TYPES),
        key=lambda e: (e.get("type", ""), e.get("detail", "")),
    )
    for e in entries:
        detail = e.get("detail", "")
        rows.append(
            {
                "category": "missing-document",
                "ask": one_sentence(f"Supply the missing document: {detail}"),
                "evidence": e.get("evidence", ""),
                "source_refs": f"gap:{e.get('type', '')}:{e.get('evidence', '')}",
            }
        )
    return rows


def unresolved_rows(findings, docs, issues):
    rows = []
    unresolved = sorted(
        (f for f in findings.get("findings", []) if f.get("status") == "unresolved"),
        key=lambda f: f.get("finding_id", ""),
    )
    for f in unresolved:
        path = doc_path(f.get("doc_id", ""), docs)
        question, rule = issues.get(f.get("issue_id", ""), ("", ""))
        if question and rule:
            ask = f"To answer '{question}' for {path}, clarify: {rule}"
        else:
            ask = (
                f"Clarify the unresolved finding on {f.get('issue_id', '')} for {path}: "
                f"{f.get('characterization', '')}"
            )
        rows.append(
            {
                "category": "unresolved",
                "ask": one_sentence(ask),
                "evidence": (f.get("quote") or "").strip()
                or f.get("characterization", ""),
                "source_refs": f.get("finding_id", ""),
            }
        )
    return rows


def inconsistency_rows(findings, families, docs):
    family_ids = {f.get("family_id", "") for f in families.get("families", [])}
    groups = {}
    for f in findings.get("findings", []):
        if f.get("unit_id") in family_ids:
            groups.setdefault((f.get("issue_id", ""), f.get("unit_id", "")), []).append(
                f
            )
    rows = []
    for issue_id, unit_id in sorted(groups):
        members = sorted(
            groups[(issue_id, unit_id)], key=lambda f: f.get("finding_id", "")
        )
        present = [f for f in members if f.get("status") == "present"]
        absent = [f for f in members if f.get("status") == "absent"]
        if not (present and absent):
            continue
        fam_label = doc_path(unit_id, docs)
        p_paths = ", ".join(doc_path(f.get("doc_id", ""), docs) for f in present)
        a_paths = ", ".join(doc_path(f.get("doc_id", ""), docs) for f in absent)
        ask = (
            f"Reconcile the conflicting positions on {issue_id} within the family based on "
            f"{fam_label}: {p_paths} reports present while {a_paths} reports absent"
        )
        refs = ";".join(sorted(f.get("finding_id", "") for f in present + absent))
        rows.append(
            {
                "category": "inconsistency",
                "ask": one_sentence(ask),
                "evidence": (present[0].get("quote") or "").strip()
                or present[0].get("characterization", ""),
                "source_refs": refs,
            }
        )
    return rows


def build_rows(findings, gaps, families, manifest, framework=None):
    docs = {d["id"]: d for d in manifest.get("documents", [])}
    issues = issue_lookup(framework)
    raw = (
        gap_rows(gaps)
        + unresolved_rows(findings, docs, issues)
        + inconsistency_rows(findings, families, docs)
    )
    # Dedup: one row per underlying ask; merge source refs.
    merged = {}
    for r in raw:
        key = (r["category"], r["ask"].casefold())
        if key in merged:
            refs = set(merged[key]["source_refs"].split(";")) | set(
                r["source_refs"].split(";")
            )
            merged[key]["source_refs"] = ";".join(sorted(refs))
        else:
            merged[key] = r
    rows = []
    for r in merged.values():
        r["register_id"] = register_id(r["category"], r["source_refs"])
        rows.append({c: r[c] for c in CSV_COLUMNS})
    rows.sort(key=lambda r: (r["category"], r["register_id"]))
    return rows


def main():
    ap = argparse.ArgumentParser(description="Export the further-enquiries register.")
    ap.add_argument("--findings", required=True)
    ap.add_argument("--gaps", required=True)
    ap.add_argument("--families", required=True)
    ap.add_argument("--manifest", required=True)
    ap.add_argument(
        "--framework",
        default=None,
        help="optional; supplies question and unresolved_when text for asks",
    )
    ap.add_argument(
        "--out",
        required=True,
        help="path for register.csv; register.json lands beside it",
    )
    args = ap.parse_args()

    findings = load(args.findings, "findings", ["findings"])
    gaps = load(args.gaps, "gap report", ["entries"])
    families = load(args.families, "families", ["families"])
    manifest = load(args.manifest, "manifest", ["documents", "counts"])
    framework = (
        load(args.framework, "framework", ["framework_version", "lenses"])
        if args.framework
        else None
    )

    rows = build_rows(findings, gaps, families, manifest, framework)

    with open(args.out, "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f, quoting=csv.QUOTE_MINIMAL)
        w.writerow(CSV_COLUMNS)
        for r in rows:
            w.writerow([r[c] for c in CSV_COLUMNS])

    json_path = os.path.splitext(args.out)[0] + ".json"
    with open(json_path, "w", encoding="utf-8", newline="\n") as f:
        json.dump({"rows": rows}, f, sort_keys=True, indent=2)
        f.write("\n")

    print(f"wrote {args.out} and {json_path} ({len(rows)} rows)")


if __name__ == "__main__":
    main()