scripts/convert_corpus_json_to_csv.py

#!/usr/bin/env python3
"""
Convertit le corpus enseignant JSON vers le CSV du workflow qualité BacPilot.

Usage:
    python3 scripts/convert_corpus_json_to_csv.py <corpus_dir> <output_csv>

Exemple:
    python3 scripts/convert_corpus_json_to_csv.py \
        /home/debpc/corpus_teacher_100 \
        quality/corpus/teacher_evaluation_cases_100.csv

Seules les copies avec status TEACHER_VALIDATED ou READY_FOR_EVAL sont exportées.
Les copies DRAFT sont ignorées.
"""

import csv
import json
import sys
from pathlib import Path


FIELDNAMES = [
    "submission_case_id", "exercise_id", "chapter", "skill",
    "student_answer_anonymized", "teacher_score", "teacher_max_score",
    "teacher_errors", "teacher_comment",
    "bacpilot_score", "bacpilot_max_score", "bacpilot_verdict",
    "bacpilot_confidence", "bacpilot_main_error",
    "absolute_score_error", "false_error_count", "missed_error_count",
    "needs_human_review_expected", "needs_human_review_bacpilot", "decision",
]

VERDICTS_REVIEW = {
    "PARTIELLEMENT_CORRECT",
    "METHODE_CORRECTE_RESULTAT_FAUX",
    "RESULTAT_CORRECT_JUSTIFICATION_INSUFFISANTE",
    "NON_EVALUABLE",
}


def extract_student_answer(path: Path) -> str:
    text = path.read_text(encoding="utf-8")
    if "Réponse :" in text:
        after = text.split("Réponse :")[-1].strip()
        if after.startswith("---"):
            after = after[3:].strip()
        return after
    return text.strip()


def convert(corpus_dir: Path, output_csv: Path) -> None:
    copies_dir = corpus_dir / "copies"
    if not copies_dir.exists():
        print(f"ERREUR: dossier copies introuvable dans {corpus_dir}")
        sys.exit(1)

    rows = []
    converted = 0
    skipped_draft = 0
    errors = []

    for copy_dir in sorted(copies_dir.iterdir()):
        if not copy_dir.is_dir():
            continue

        meta_path = copy_dir / "metadata.json"
        answer_path = copy_dir / "student_answer.md"
        correction_path = copy_dir / "teacher_correction.json"

        if not all(p.exists() for p in [meta_path, answer_path, correction_path]):
            errors.append(f"{copy_dir.name}: fichiers manquants")
            continue

        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
            correction = json.loads(correction_path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as e:
            errors.append(f"{copy_dir.name}: JSON invalide — {e}")
            continue

        status = correction.get("status", "DRAFT")
        if status not in ("TEACHER_VALIDATED", "READY_FOR_EVAL"):
            skipped_draft += 1
            continue

        student_answer = extract_student_answer(answer_path)
        main_errors = correction.get("main_errors", [])
        verdict = correction.get("verdict", "")
        needs_review = "true" if verdict in VERDICTS_REVIEW else "false"

        rows.append({
            "submission_case_id": correction.get("copy_id", copy_dir.name),
            "exercise_id": correction.get("exercise_id", ""),
            "chapter": meta.get("chapter", ""),
            "skill": meta.get("skill", ""),
            "student_answer_anonymized": student_answer,
            "teacher_score": correction.get("teacher_score", ""),
            "teacher_max_score": correction.get("max_score", ""),
            "teacher_errors": ";".join(main_errors),
            "teacher_comment": correction.get("teacher_comment", ""),
            "bacpilot_score": "",
            "bacpilot_max_score": "",
            "bacpilot_verdict": "",
            "bacpilot_confidence": "",
            "bacpilot_main_error": "",
            "absolute_score_error": "",
            "false_error_count": "",
            "missed_error_count": "",
            "needs_human_review_expected": needs_review,
            "needs_human_review_bacpilot": "",
            "decision": "",
        })
        converted += 1

    output_csv.parent.mkdir(parents=True, exist_ok=True)
    with output_csv.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
        writer.writeheader()
        writer.writerows(rows)

    print(f"CONVERTED={converted}")
    print(f"SKIPPED_DRAFT={skipped_draft}")
    print(f"ERRORS={len(errors)}")
    print(f"OUTPUT={output_csv}")

    if errors:
        for e in errors:
            print(f"  ERREUR: {e}")

    if converted == 0:
        print("STATUS=NO_READY_COPIES — remplir les copies et passer status à TEACHER_VALIDATED")
    else:
        print("STATUS=OK")


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print(f"Usage: {sys.argv[0]} <corpus_dir> <output_csv>")
        sys.exit(1)

    convert(Path(sys.argv[1]), Path(sys.argv[2]))