geointel/scripts/validate_detection_false_negative_review_decisions.py

#!/usr/bin/env python3
"""Validate explicit operator decisions for detection QA false-negatives."""

from __future__ import annotations

import argparse
import csv
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


JSON_NAME = "detection_false_negative_review_validation.json"
MARKDOWN_NAME = "detection_false_negative_review_validation.md"
CONFIRMED_NAME = "confirmed_model_false_negatives.geojson"
DECISIONS = (
    "confirmed_model_false_negative",
    "qa_alignment_mismatch",
    "reference_gap_or_change",
    "imagery_obscured_or_uncertain",
    "unreviewed",
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate manual false-negative review decisions without inferring labels."
    )
    parser.add_argument("--review-summary", required=True, type=Path)
    parser.add_argument("--decisions-csv", required=True, type=Path)
    parser.add_argument("--output-dir", required=True, type=Path)
    parser.add_argument(
        "--require-complete",
        action="store_true",
        help="Return a non-zero exit code while any selected record remains unreviewed.",
    )
    return parser.parse_args()


def load_json(path: Path) -> dict[str, Any]:
    if not path.is_file():
        raise SystemExit(f"Review summary is not readable: {path}")
    payload = json.loads(path.read_text(encoding="utf-8-sig"))
    if not isinstance(payload, dict):
        raise SystemExit(f"Review summary must be a JSON object: {path}")
    return payload


def load_decisions(path: Path) -> list[dict[str, str]]:
    if not path.is_file():
        raise SystemExit(f"Review decisions CSV is not readable: {path}")
    with path.open(newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle)
        required = {"reference_feature_id", "review_decision", "review_notes"}
        missing = required - set(reader.fieldnames or [])
        if missing:
            raise SystemExit(
                "Review decisions CSV lacks required columns: "
                + ", ".join(sorted(missing))
            )
        return [dict(row) for row in reader]


def validate(
    summary: dict[str, Any], rows: list[dict[str, str]]
) -> tuple[dict[str, Any], dict[str, Any]]:
    features = summary.get("selected_features") or []
    if not isinstance(features, list) or not all(
        isinstance(item, dict) for item in features
    ):
        raise SystemExit("Review summary selected_features must be a list of objects")

    expected: dict[str, dict[str, Any]] = {}
    for feature in features:
        reference_id = str(feature.get("reference_feature_id") or "").strip()
        if not reference_id or reference_id in expected:
            raise SystemExit(
                "Review summary contains a missing or duplicate reference_feature_id"
            )
        expected[reference_id] = feature

    provided: dict[str, dict[str, str]] = {}
    for row in rows:
        reference_id = str(row.get("reference_feature_id") or "").strip()
        if not reference_id or reference_id in provided:
            raise SystemExit(
                "Review decisions contain a missing or duplicate reference_feature_id"
            )
        decision = str(row.get("review_decision") or "").strip()
        if decision not in DECISIONS:
            raise SystemExit(
                f"Invalid review decision for {reference_id}: {decision}. "
                + "Allowed values: "
                + ", ".join(DECISIONS)
            )
        provided[reference_id] = row

    missing_ids = set(expected) - set(provided)
    extra_ids = set(provided) - set(expected)
    if missing_ids or extra_ids:
        details = []
        if missing_ids:
            details.append("missing: " + ", ".join(sorted(missing_ids)))
        if extra_ids:
            details.append("unexpected: " + ", ".join(sorted(extra_ids)))
        raise SystemExit(
            "Review decisions do not match the selected evidence ("
            + "; ".join(details)
            + ")"
        )

    counts = {decision: 0 for decision in DECISIONS}
    confirmed_features: list[dict[str, Any]] = []
    for reference_id, feature in expected.items():
        row = provided[reference_id]
        decision = row["review_decision"].strip()
        counts[decision] += 1
        if decision != "confirmed_model_false_negative":
            continue
        source_properties = feature.get("properties") or {}
        properties = (
            dict(source_properties) if isinstance(source_properties, dict) else {}
        )
        properties.update(
            {
                "reference_feature_id": reference_id,
                "sample_slug": feature.get("sample_slug"),
                "area_m2": feature.get("area_m2"),
                "area_bucket": feature.get("area_bucket"),
                "source_tile_path": feature.get("source_tile_path"),
                "review_decision": decision,
                "review_notes": row.get("review_notes", "").strip(),
            }
        )
        confirmed_features.append(
            {
                "type": "Feature",
                "id": f"confirmed_model_false_negative:{reference_id}",
                "geometry": feature.get("geometry"),
                "properties": properties,
            }
        )

    status = "complete" if counts["unreviewed"] == 0 else "review_required"
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "schema_version": 1,
        "status": status,
        "review_summary_path": str(summary.get("portfolio_path") or ""),
        "selected_feature_count": len(features),
        "decision_counts": counts,
        "confirmed_model_false_negative_count": len(confirmed_features),
        "safety_rule": (
            "Only explicit confirmed_model_false_negative decisions are exported; "
            "QA false-negatives are never inferred as model training labels."
        ),
    }
    return report, {"type": "FeatureCollection", "features": confirmed_features}


def write_markdown(report: dict[str, Any], output_dir: Path) -> None:
    lines = [
        "# Detection false-negative review validation",
        "",
        f"- Status: `{report['status']}`",
        f"- Selected records: {report['selected_feature_count']}",
        f"- Confirmed model false-negatives: {report['confirmed_model_false_negative_count']}",
        "",
        "## Decision counts",
        "",
    ]
    lines.extend(
        f"- `{decision}`: {count}"
        for decision, count in report["decision_counts"].items()
    )
    lines.extend(["", f"> {report['safety_rule']}", ""])
    (output_dir / MARKDOWN_NAME).write_text("\n".join(lines), encoding="utf-8")


def main() -> int:
    args = parse_args()
    summary_path = args.review_summary.expanduser().resolve()
    decisions_path = args.decisions_csv.expanduser().resolve()
    output_dir = args.output_dir.expanduser().resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    report, confirmed = validate(load_json(summary_path), load_decisions(decisions_path))
    report["review_summary_path"] = str(summary_path)
    report["decisions_csv_path"] = str(decisions_path)
    (output_dir / JSON_NAME).write_text(
        json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
    )
    (output_dir / CONFIRMED_NAME).write_text(
        json.dumps(confirmed, indent=2, sort_keys=True), encoding="utf-8"
    )
    write_markdown(report, output_dir)
    print(f"False-negative review status: {report['status']}")
    print(f"Validation: {output_dir / JSON_NAME}")
    print(f"Confirmed evidence: {output_dir / CONFIRMED_NAME}")
    if args.require_complete and report["status"] != "complete":
        print("Review remains incomplete", flush=True)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())