Spaces:
Sleeping
Sleeping
File size: 3,604 Bytes
761d149 830d137 feb1b1c 830d137 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
from __future__ import annotations
import csv
import re
from pathlib import Path
import typer
from redstack.domain.ids import CANDIDATE_ID_PATTERN
__all__: tuple[str, ...] = ("validate",)
_HEADER: tuple[str, ...] = ("candidate_id", "rank", "score", "reasoning")
_ID_RE = re.compile(CANDIDATE_ID_PATTERN)
def _check_rows(rows: list[dict[str, str]], expected_size: int) -> list[str]:
"""Return every structural violation found in ``rows`` (empty if valid)."""
errors: list[str] = []
if len(rows) != expected_size:
errors.append(f"expected {expected_size} rows, got {len(rows)}")
seen_ids: set[str] = set()
prev_score: float | None = None
prev_id: str | None = None
for position, row in enumerate(rows, start=1):
candidate_id = row["candidate_id"]
try:
rank = int(row["rank"])
except ValueError:
errors.append(f"row {position}: rank {row['rank']!r} is not an integer")
continue
try:
score = float(row["score"])
except ValueError:
errors.append(f"row {position}: score {row['score']!r} is not a float")
continue
if rank != position:
errors.append(f"row {position}: rank {rank} != row position {position}")
if _ID_RE.fullmatch(candidate_id) is None:
errors.append(f"row {position}: malformed candidate_id {candidate_id!r}")
if candidate_id in seen_ids:
errors.append(f"row {position}: duplicate candidate_id {candidate_id!r}")
seen_ids.add(candidate_id)
if not row["reasoning"].strip():
errors.append(f"row {position} ({candidate_id}): reasoning is blank")
if prev_score is not None and prev_id is not None:
if score > prev_score:
errors.append(
f"row {position}: score {score} > previous row's score {prev_score}"
)
elif score == prev_score and not (prev_id < candidate_id):
errors.append(
f"row {position}: tie-break violated — equal scores must order "
f"by ascending candidate_id ({prev_id!r} then {candidate_id!r})"
)
prev_score, prev_id = score, candidate_id
return errors
def validate(
submission: Path = typer.Option(
..., "--submission", help="Path to the finished submission.csv."
),
expected_size: int = typer.Option(
100, "--expected-size", help="Required row count (the top-K cut)."
),
) -> None:
"""Validate ``submission`` against the structural submission rules."""
try:
with submission.open(newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
if reader.fieldnames is None or tuple(reader.fieldnames) != _HEADER:
typer.secho(
f"header mismatch: expected {_HEADER}, got {reader.fieldnames}",
fg=typer.colors.RED,
err=True,
)
raise typer.Exit(code=1)
rows = list(reader)
except OSError as exc:
typer.secho(f"cannot read {submission}: {exc}", fg=typer.colors.RED, err=True)
raise typer.Exit(code=1) from exc
errors = _check_rows(rows, expected_size)
if errors:
for error in errors:
typer.secho(error, fg=typer.colors.RED, err=True)
typer.secho(f"INVALID: {len(errors)} violation(s)", fg=typer.colors.RED)
raise typer.Exit(code=1)
typer.secho(f"VALID: {len(rows)} rows", fg=typer.colors.GREEN)
|