scripts/audit_expert_annotations.py

#!/usr/bin/env python3
"""Fail closed unless expert event and qrel annotation gates both pass."""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "src"))

from telen.annotation import (  # noqa: E402
    PROTOCOL_VERSION,
    audit_event_annotations,
    audit_event_qrel_links,
    audit_question_annotations,
    audit_question_event_links,
    audit_qrel_annotations,
)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def run(args: argparse.Namespace) -> dict:
    event_frame = pd.read_csv(args.events, dtype=str).fillna("")
    question_frame = pd.read_csv(args.questions, dtype=str).fillna("")
    qrel_frame = pd.read_csv(args.qrels, dtype=str).fillna("")
    event_audit = audit_event_annotations(
        event_frame, minimum_kappa=args.minimum_event_kappa
    )
    qrel_audit = audit_qrel_annotations(
        qrel_frame,
        minimum_weighted_kappa=args.minimum_qrel_kappa,
        minimum_event_groups=args.minimum_event_groups,
    )
    question_audit = audit_question_annotations(
        question_frame, minimum_kappa=args.minimum_question_kappa
    )
    question_link_audit = audit_question_event_links(event_frame, question_frame)
    package_link_audit = audit_event_qrel_links(event_frame, qrel_frame)
    learned_authorized = bool(
        event_audit.authorized
        and question_audit.authorized
        and question_link_audit["authorized"]
        and qrel_audit.authorized
        and package_link_audit["authorized"]
    )
    result = {
        "schema_version": 2,
        "artifact_type": "expert_annotation_gate_audit",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "protocol_version": PROTOCOL_VERSION,
        "inputs": {
            "events": str(args.events),
            "events_sha256": sha256_file(args.events),
            "questions": str(args.questions),
            "questions_sha256": sha256_file(args.questions),
            "qrels": str(args.qrels),
            "qrels_sha256": sha256_file(args.qrels),
        },
        "event_gate": event_audit.to_dict(),
        "question_gate": question_audit.to_dict(),
        "question_event_link_gate": question_link_audit,
        "qrel_gate": qrel_audit.to_dict(),
        "package_link_gate": package_link_audit,
        "learned_residual_gate": {
            "authorized": learned_authorized,
            "rule": (
                "event_gate.authorized AND question_gate.authorized AND "
                "question_event_link_gate.authorized AND qrel_gate.authorized "
                "AND package_link_gate.authorized"
            ),
        },
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )
    return result


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    root = REPO_ROOT / "tmp/telen/temporal-pilot"
    parser.add_argument(
        "--events", type=Path, default=root / "official_verification_template.csv"
    )
    parser.add_argument(
        "--qrels", type=Path, default=root / "qrel_judgment_template.csv"
    )
    parser.add_argument(
        "--questions", type=Path, default=root / "expert_annotation_template.csv"
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=REPO_ROOT / "tmp/telen/expert-annotation-gate-audit.json",
    )
    parser.add_argument("--minimum-event-kappa", type=float, default=0.80)
    parser.add_argument("--minimum-qrel-kappa", type=float, default=0.70)
    parser.add_argument("--minimum-question-kappa", type=float, default=0.70)
    parser.add_argument("--minimum-event-groups", type=int, default=5)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    result = run(args)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if not result["learned_residual_gate"]["authorized"]:
        raise SystemExit(2)


if __name__ == "__main__":
    main()