bikoret/src/tools/hakhel.py

#!/usr/bin/env python3
"""Hakhel — the scroll read again before all (bishe'arecha §VIII).

    PYTHONPATH=korban_processor/src python3 src/tools/hakhel.py \\
        --corpus specimens.jsonl [--precision sample.json] [--recall sample.json] \\
        [--readings-ref <git sha>] [--post <base-url> --key-file <chatzer key file>] [--out hakhel.json]

Runs the Korban Processor's readings — the same code the pod runs, at the
same commit — over the specimen corpus and the hand-labelled samples, and
produces one record: what each reading fires on across the corpus, what
stands on two witnesses (Tamid §1.1), and the precision and recall on
the hand labels. The corpus stays private; the record holds counts and
the corpus's SHA-256 — never a specimen. With --post the record goes to
POST /chatzer/hakhel and is engraved in L3.
"""
import argparse
import hashlib
import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone

from korban_processor.bedikah import (
    find_authority_claims,
    find_excluded_adversarial,
    find_frames,
    find_self_promotion,
    find_self_validation,
)

FRAME_FORMS = ["contested_laundering", "virtue_certification", "agency_displacement", "apology_without_mechanism",
               "unrequested_confidence", "smuggled_vocabulary", "source_asymmetry", "forged_fringe"]
READINGS = FRAME_FORMS + ["self_validation", "self_promotion", "excluded_adversarial", "authority"]
LABEL_TO_READING = {
    "contested_laundering": ["contested_laundering"],
    "virtue_certification": ["virtue_certification"],
    "agency_displacement": ["agency_displacement"],
    "apology_without_mechanism": ["apology_without_mechanism"],
    "unrequested_confidence": ["unrequested_confidence"],
    "smuggled_vocabulary": ["smuggled_vocabulary"],
    "source_asymmetry": ["source_asymmetry"],
    "neutral_frame": ["contested_laundering", "source_asymmetry"],
    "golden_calf": ["virtue_certification"],
}


def read(output, query=""):
    """What the readings see in one output: reading -> list of distinct sentences."""
    seen = {form: [] for form in FRAME_FORMS}
    for f in find_frames(output, query):
        if f.sentence not in seen[f.form]:
            seen[f.form].append(f.sentence)
    seen["self_validation"] = [h.sentence for h in find_self_validation(output)]
    seen["self_promotion"] = [h.sentence for h in find_self_promotion(output)]
    seen["excluded_adversarial"] = [h.sentence for h in find_excluded_adversarial(output)]
    seen["authority"] = [h.sentence for h in find_authority_claims(output)]
    return seen


def standing_forms(seen):
    """Two witnesses (Devarim 19:15), as Bedikah.check_frames counts them:
    a form stands on two sentences, or on one beside another reading's hit
    or another form's sentence."""
    other_checks = sum(len(seen[r]) for r in ("self_validation", "self_promotion", "excluded_adversarial", "authority"))
    frame_witnesses = sum(len(seen[f]) for f in FRAME_FORMS)
    stands = []
    for form in FRAME_FORMS:
        n = len(seen[form])
        if n >= 2 or (n == 1 and (other_checks > 0 or frame_witnesses >= 2)):
            stands.append(form)
    return stands


def sha256_of(path):
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def git_ref():
    try:
        return subprocess.check_output(["git", "rev-parse", "--short=8", "HEAD"], text=True, stderr=subprocess.DEVNULL).strip()
    except Exception:
        return None


def run(corpus_path, precision_path=None, recall_path=None, readings_ref=None):
    rows = [json.loads(l) for l in open(corpus_path)]
    fires = {r: 0 for r in READINGS}
    stands = {f: 0 for f in FRAME_FORMS}
    any_fire = 0
    any_stand = 0
    by_label = {}
    for row in rows:
        seen = read(row["output"], "")
        hit = False
        for r in READINGS:
            if seen[r]:
                fires[r] += 1
                hit = True
        any_fire += hit
        st = standing_forms(seen)
        for f in st:
            stands[f] += 1
        any_stand += bool(st)
        for label in row.get("labels", []):
            entry = by_label.setdefault(label, {"n": 0, "seen": 0})
            entry["n"] += 1
            if any(seen[x] for x in LABEL_TO_READING.get(label, [])):
                entry["seen"] += 1

    record = {
        "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "corpus_sha256": sha256_of(corpus_path),
        "corpus_size": len(rows),
        "readings_ref": readings_ref or git_ref(),
        "counts": {
            "fires": fires,
            "stands": stands,
            "any_fire": any_fire,
            "any_stand": any_stand,
            "by_label": by_label,
        },
        "labelled": {},
    }

    # Precision: hand-labelled sentences the reading matched (True = the form is really there).
    if precision_path:
        prec = json.load(open(precision_path))
        per = {}
        for x in prec:
            if x.get("hand_label") is None:
                continue
            e = per.setdefault(x["form"], {"n": 0, "true": 0})
            e["n"] += 1
            e["true"] += bool(x["hand_label"])
        total_n = sum(e["n"] for e in per.values())
        total_t = sum(e["true"] for e in per.values())
        record["labelled"]["precision"] = {"per_form": per, "n": total_n, "true": total_t,
                                           "sha256": sha256_of(precision_path)}

    # Recall: hand-labelled outputs that carry a form — does the reading see it now?
    if recall_path:
        rec = json.load(open(recall_path))
        per = {}
        for x in rec:
            seen = read(x["output"], "")
            e = per.setdefault(x["label"], {"n": 0, "seen": 0})
            e["n"] += 1
            e["seen"] += bool(any(seen[r] for r in LABEL_TO_READING.get(x["label"], [x["label"]]) if r in seen))
        record["labelled"]["recall"] = {"per_label": per, "n": sum(e["n"] for e in per.values()),
                                        "seen": sum(e["seen"] for e in per.values()), "sha256": sha256_of(recall_path)}

    return record


def post(record, base_url, key):
    req = urllib.request.Request(base_url.rstrip("/") + "/chatzer/hakhel", data=json.dumps(record).encode(),
                                 headers={"Content-Type": "application/json", "Authorization": "Bearer " + key})
    try:
        r = urllib.request.urlopen(req, timeout=60)
        return r.status, json.load(r)
    except urllib.error.HTTPError as e:
        return e.code, json.load(e)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--corpus", required=True)
    ap.add_argument("--precision")
    ap.add_argument("--recall")
    ap.add_argument("--readings-ref")
    ap.add_argument("--post", help="base URL; posts to /chatzer/hakhel")
    ap.add_argument("--key-file", help="file whose line 'chatzer <key>' holds the key")
    ap.add_argument("--out")
    a = ap.parse_args()

    record = run(a.corpus, a.precision, a.recall, a.readings_ref)
    c = record["counts"]
    print(f"hakhel {record['date']} readings {record['readings_ref']} corpus {record['corpus_size']} ({record['corpus_sha256'][:8]})")
    print(f"  any reading fired on {c['any_fire']} ({c['any_fire']/max(1,record['corpus_size']):.0%}); "
          f"a form stands on {c['any_stand']} ({c['any_stand']/max(1,record['corpus_size']):.0%})")
    print("  fires:", c["fires"])
    print("  stands:", c["stands"])
    if "precision" in record["labelled"]:
        p = record["labelled"]["precision"]
        print(f"  precision on hand labels: {p['true']}/{p['n']} ({p['true']/max(1,p['n']):.0%})")
    if "recall" in record["labelled"]:
        r = record["labelled"]["recall"]
        print(f"  recall on hand labels: {r['seen']}/{r['n']} ({r['seen']/max(1,r['n']):.0%})")
    if a.out:
        with open(a.out, "w") as fh:
            json.dump(record, fh, indent=1)
        print("  written", a.out)
    if a.post:
        key = None
        if a.key_file:
            for line in open(a.key_file):
                parts = line.split()
                if len(parts) >= 2 and parts[0] == "chatzer":
                    key = parts[1]
        if not key:
            sys.exit("no chatzer key")
        status, body = post(record, a.post, key)
        print("  posted:", status, body.get("hakhel_ref") or body.get("error", {}).get("code"))
        if status != 200:
            sys.exit(1)


if __name__ == "__main__":
    main()