Upload duplicate removal file.

training/remove_exact_mmlu_question_duplicates.py

@@ -0,0 +1,135 @@
+"""Remove records whose question exactly matches a reference dataset question.
+"""
+
+import argparse
+import json
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+
+DEFAULT_TARGET = Path(
+    r"D:\other\Research\ICTER\combined_extracted_answers_new.json"
+)
+DEFAULT_REFERENCE = Path(
+    r"D:\other\Research\ICTER\Sinhala-MMLU-Test-Dataset\Combined_All.json"
+)
+
+
+def load_json_array(path: Path) -> list:
+    with path.open("r", encoding="utf-8-sig") as file:
+        data = json.load(file)
+    if not isinstance(data, list):
+        raise TypeError(f"Expected a top-level JSON array in: {path}")
+    return data
+
+
+def next_backup_path(target: Path) -> Path:
+    candidate = target.with_name(target.stem + ".before_exact_removal.json")
+    number = 2
+    while candidate.exists():
+        candidate = target.with_name(
+            f"{target.stem}.before_exact_removal_{number}.json"
+        )
+        number += 1
+    return candidate
+
+
+def remove_exact_matches(target: Path, reference: Path) -> dict:
+    target_records = load_json_array(target)
+    reference_records = load_json_array(reference)
+
+    # No normalization is performed here. Membership uses exact string equality.
+    reference_questions = {
+        record["question"]
+        for record in reference_records
+        if isinstance(record, dict) and isinstance(record.get("question"), str)
+    }
+
+    kept_records = []
+    removed_records = []
+    for record in target_records:
+        question = record.get("question") if isinstance(record, dict) else None
+        if isinstance(question, str) and question in reference_questions:
+            removed_records.append(record)
+        else:
+            kept_records.append(record)
+
+    backup = next_backup_path(target)
+    shutil.copy2(target, backup)
+
+    temporary_file = tempfile.NamedTemporaryFile(
+        mode="w",
+        encoding="utf-8",
+        suffix=".json",
+        prefix="exact_match_filtered_",
+        dir=target.parent,
+        delete=False,
+    )
+    temporary_path = Path(temporary_file.name)
+
+    try:
+        with temporary_file:
+            json.dump(kept_records, temporary_file, ensure_ascii=False, indent=2)
+            temporary_file.write("\n")
+
+        # Reload and verify before replacing the original file.
+        verified_records = load_json_array(temporary_path)
+        matches_remaining = sum(
+            1
+            for record in verified_records
+            if isinstance(record, dict)
+            and isinstance(record.get("question"), str)
+            and record["question"] in reference_questions
+        )
+
+        if len(verified_records) != len(kept_records):
+            raise ValueError("Output record count verification failed")
+        if matches_remaining != 0:
+            raise ValueError("Exact matching questions remain in the output")
+
+        os.replace(temporary_path, target)
+    except Exception:
+        temporary_path.unlink(missing_ok=True)
+        raise
+
+    return {
+        "records_before": len(target_records),
+        "reference_records": len(reference_records),
+        "unique_reference_questions": len(reference_questions),
+        "records_removed": len(removed_records),
+        "unique_questions_removed": len(
+            {record["question"] for record in removed_records}
+        ),
+        "records_after": len(kept_records),
+        "exact_matches_remaining": 0,
+        "updated_file": str(target),
+        "backup_file": str(backup),
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Remove records with question strings exactly matching a reference JSON dataset."
+    )
+    parser.add_argument(
+        "--target",
+        type=Path,
+        default=DEFAULT_TARGET,
+        help=f"JSON file to update (default: {DEFAULT_TARGET})",
+    )
+    parser.add_argument(
+        "--reference",
+        type=Path,
+        default=DEFAULT_REFERENCE,
+        help=f"Reference JSON dataset (default: {DEFAULT_REFERENCE})",
+    )
+    args = parser.parse_args()
+
+    result = remove_exact_matches(args.target, args.reference)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()