scripts/evaluate_teacher_quality.py

from __future__ import annotations

import argparse
import csv
import json
from pathlib import Path
from statistics import mean, median


DEFAULT_INPUT_PATH = Path("quality/corpus/teacher_evaluation_cases_template.csv")
DEFAULT_OUTPUT_PATH = Path("quality/results/teacher_quality_metrics.json")


def parse_float(value: str) -> float | None:
    cleaned = value.strip()
    if not cleaned:
        return None

    return float(cleaned)


def parse_int(value: str) -> int:
    cleaned = value.strip()
    if not cleaned:
        return 0

    return int(cleaned)


def parse_bool(value: str) -> bool | None:
    cleaned = value.strip().lower()

    if not cleaned:
        return None

    if cleaned in {"true", "1", "yes", "oui"}:
        return True

    if cleaned in {"false", "0", "no", "non"}:
        return False

    raise ValueError(f"Valeur booléenne invalide: {value}")


def compute_metrics(input_path: Path) -> dict[str, object]:
    if not input_path.exists():
        raise SystemExit(f"Fichier introuvable: {input_path}")

    rows = list(csv.DictReader(input_path.open(newline="", encoding="utf-8")))
    evaluated_rows = []

    for row in rows:
        teacher_score = parse_float(row.get("teacher_score", ""))
        bacpilot_score = parse_float(row.get("bacpilot_score", ""))

        if teacher_score is None or bacpilot_score is None:
            continue

        absolute_error = abs(bacpilot_score - teacher_score)

        evaluated_rows.append(
            {
                "submission_case_id": row.get("submission_case_id", ""),
                "chapter": row.get("chapter", ""),
                "teacher_score": teacher_score,
                "bacpilot_score": bacpilot_score,
                "absolute_error": absolute_error,
                "false_error_count": parse_int(row.get("false_error_count", "")),
                "missed_error_count": parse_int(row.get("missed_error_count", "")),
                "needs_human_review_expected": parse_bool(
                    row.get("needs_human_review_expected", "")
                ),
                "needs_human_review_bacpilot": parse_bool(
                    row.get("needs_human_review_bacpilot", "")
                ),
            }
        )

    if not evaluated_rows:
        return {
            "status": "no_evaluable_rows",
            "message": "Remplir teacher_score et bacpilot_score pour calculer les métriques.",
            "evaluated_count": 0,
            "input_file": str(input_path),
        }

    absolute_errors = [row["absolute_error"] for row in evaluated_rows]
    false_errors = [row["false_error_count"] for row in evaluated_rows]
    missed_errors = [row["missed_error_count"] for row in evaluated_rows]

    review_rows = [
        row
        for row in evaluated_rows
        if row["needs_human_review_expected"] is not None
        and row["needs_human_review_bacpilot"] is not None
    ]

    review_mismatches = [
        row
        for row in review_rows
        if row["needs_human_review_expected"] != row["needs_human_review_bacpilot"]
    ]

    return {
        "status": "ok",
        "input_file": str(input_path),
        "evaluated_count": len(evaluated_rows),
        "mae": round(mean(absolute_errors), 4),
        "median_absolute_error": round(median(absolute_errors), 4),
        "within_0_5_point_count": sum(error <= 0.5 for error in absolute_errors),
        "within_1_point_count": sum(error <= 1.0 for error in absolute_errors),
        "within_0_5_point_rate": round(
            sum(error <= 0.5 for error in absolute_errors) / len(absolute_errors),
            4,
        ),
        "within_1_point_rate": round(
            sum(error <= 1.0 for error in absolute_errors) / len(absolute_errors),
            4,
        ),
        "false_error_total": sum(false_errors),
        "missed_error_total": sum(missed_errors),
        "human_review_evaluated_count": len(review_rows),
        "human_review_mismatch_count": len(review_mismatches),
        "human_review_mismatch_rate": (
            round(len(review_mismatches) / len(review_rows), 4)
            if review_rows
            else None
        ),
    }


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Évalue les métriques qualité BacPilot contre corrections enseignants."
    )
    parser.add_argument(
        "--input",
        type=Path,
        default=DEFAULT_INPUT_PATH,
        help="Chemin du CSV à évaluer.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=DEFAULT_OUTPUT_PATH,
        help="Chemin du JSON de sortie.",
    )

    args = parser.parse_args()

    metrics = compute_metrics(args.input)

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(metrics, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )

    print(json.dumps(metrics, ensure_ascii=False, indent=2))
    print(f"OUTPUT={args.output}")


if __name__ == "__main__":
    main()