ClassLens-dev / chatkit /backend /app /parsers /csv_strict.py
Yu Chen
make csv no required to have equal mark as correct
363b160
Raw
History Blame Contribute Delete
10.7 kB
"""Strict CSV parser for already-tabulated answer sheets.
## Contract
A single CSV file represents the full answer table:
- Row 0: header row (column titles like `Q1, Q2, ...` — content is ignored,
only used to determine column count).
- Row 1: the answer key. First cell is the name marker (e.g. `正確解答` /
`標準答案` / `KEY`). Remaining cells are the correct letters per question.
- Row 2..N: one row per student. First cell is the student's name. Remaining
cells are the student's answer per question (single letter A-Z, `=` for
correct, or blank for "did not answer").
Column 0 is always the name column; remaining columns are positional
Q1..QN. Header text in row 0 is not required and is ignored if present.
## Example
```
,Q1,Q2,Q3,Q4,Q5
正確解答,B,A,A,C,D
梁祐邦,A,=,C,C,D
田瑜婕,=,A,C,B,D
```
"""
from __future__ import annotations
import csv
import io
from ..answer_grid import normalize_letter
from .base import AnswerSheetParser, ParserFile, ParserResult, get_extension
# Markers we treat as "this row is the answer key" if a stricter reader needs
# them — kept loose because the file is positional, not phrase-driven.
_KEY_NAME_MARKERS = {"正確解答", "標準答案", "預設標準答案", "key", "answer"}
class StrictCsvParser:
name = "csv_strict"
display_name = "CSV (strict format)"
description = (
"Direct upload of an already-tabulated answer sheet. "
"Row 0 = header (Q1, Q2, ... — ignored), row 1 = answer key "
"(first cell '正確解答'), row 2+ = students. "
"Column 0 = name; remaining columns are positional Q1..QN. "
"Cells: A-Z, '=' for correct, or blank."
)
def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool:
if data_type not in ("student_answers", "teacher_answers"):
return False
return get_extension(filename) == ".csv"
async def parse(
self,
files: list[ParserFile],
data_type: str,
description: str = "",
model: str = "gpt-5.4",
) -> ParserResult:
if data_type == "student_answers":
return _parse_students(files, self.name)
if data_type == "teacher_answers":
return _parse_teacher(files, self.name)
raise ValueError(f"csv_strict does not support data_type={data_type}")
# ---------------------------------------------------------------------------
# Pure helpers
# ---------------------------------------------------------------------------
def _decode(file_bytes: bytes) -> str:
for encoding in ("utf-8-sig", "utf-8"):
try:
return file_bytes.decode(encoding)
except UnicodeDecodeError:
continue
raise ValueError(
"CSV must be UTF-8 encoded; please save as UTF-8 in your spreadsheet tool."
)
def _read_rows(text: str) -> list[list[str]]:
reader = csv.reader(io.StringIO(text), skipinitialspace=True)
rows: list[list[str]] = []
for row in reader:
# Drop fully-blank trailing cells (Excel adds these); keep empty cells
# in the middle so column positions stay aligned.
while row and row[-1].strip() == "":
row = row[:-1]
if not row:
continue
rows.append([c.strip() for c in row])
return rows
def _column_count(rows: list[list[str]]) -> int:
return max((len(r) for r in rows), default=0)
def _pad_row(row: list[str], n: int) -> list[str]:
if len(row) >= n:
return row
return row + [""] * (n - len(row))
def _row_to_answers(row: list[str], n_questions: int) -> list[dict]:
"""Project a row's answer cells to the canonical answer list."""
answers: list[dict] = []
for i in range(n_questions):
cell = row[i + 1] if i + 1 < len(row) else ""
letter = normalize_letter(cell)
answers.append({"question_number": i + 1, "answer": letter})
return answers
def _is_key_row(name_cell: str) -> bool:
return name_cell.strip().lower() in {m.lower() for m in _KEY_NAME_MARKERS}
CORRECT_MARKER = "="
def _has_any_equals(students: list[dict]) -> bool:
"""True if any student cell holds the correct-marker."""
for student in students:
for ans in student.get("answers") or []:
if ans.get("answer") == CORRECT_MARKER:
return True
return False
def _auto_fill_correct_marker(
students: list[dict],
official_letters: list[str | None],
) -> tuple[list[dict], int]:
"""Replace student cells matching the official letter with '='.
Pure: returns a new list of student dicts, never mutates input.
Returns (new_students, n_replaced).
"""
new_students: list[dict] = []
replaced = 0
for student in students:
new_answers: list[dict] = []
for ans in student.get("answers") or []:
qn = ans.get("question_number")
value = ans.get("answer")
idx = qn - 1 if isinstance(qn, int) else -1
key_letter = (
official_letters[idx]
if 0 <= idx < len(official_letters)
else None
)
if (
value is not None
and value != CORRECT_MARKER
and key_letter is not None
and value == key_letter
):
new_answers.append({**ans, "answer": CORRECT_MARKER})
replaced += 1
else:
new_answers.append(dict(ans))
new_students.append({**student, "answers": new_answers})
return new_students, replaced
# ---------------------------------------------------------------------------
# Per-data_type extractors
# ---------------------------------------------------------------------------
def _parse_students(files: list[ParserFile], parser_name: str) -> ParserResult:
students: list[dict] = []
notes: list[str] = []
skipped = 0
for f in files:
text = _decode(f.content)
rows = _read_rows(text)
if not rows:
raise ValueError(f"{f.filename}: CSV is empty.")
if len(rows) < 3:
raise ValueError(
f"{f.filename}: expected a header row, an answer-key row, and at "
f"least one student row (got {len(rows)} row(s))."
)
n_cols = _column_count(rows)
if n_cols < 2:
raise ValueError(
f"{f.filename}: CSV must have at least 2 columns (name + Q1)."
)
n_questions = n_cols - 1
# rows[0] is the column header (Q1..QN). It's ignored beyond column count.
# rows[1] should be the answer key — warn if the name cell isn't a marker.
key_row = _pad_row(rows[1], n_cols)
key_name = key_row[0]
if not _is_key_row(key_name):
notes.append(
f"{f.filename}: row 2 first column is '{key_name}', expected "
f"'正確解答' / '標準答案' — treating it as the key anyway."
)
official_letters = [
normalize_letter(key_row[i + 1]) for i in range(n_questions)
]
file_students: list[dict] = []
for row in rows[2:]:
padded = _pad_row(row, n_cols)
name = padded[0]
if not name:
skipped += 1
continue
file_students.append(
{
"name": name,
"id": "",
"answers": _row_to_answers(padded, n_questions),
}
)
# If this file's student section has no '=' anywhere, infer correctness
# by comparing each cell to the official key letter.
if file_students and not _has_any_equals(file_students):
file_students, replaced = _auto_fill_correct_marker(
file_students, official_letters
)
if replaced:
notes.append(
f"{f.filename}: auto-converted {replaced} correct answer(s) "
f"to '=' (no '=' marker was found in this file)."
)
students.extend(file_students)
if skipped:
notes.append(f"Skipped {skipped} row(s) with empty name.")
if not students:
raise ValueError("No student rows found in CSV.")
return ParserResult(
data={"students": students},
parser_name=parser_name,
notes=tuple(notes),
)
def _parse_teacher(files: list[ParserFile], parser_name: str) -> ParserResult:
if len(files) > 1:
# If multiple files come in, we keep only the first (single-key contract).
notes_prefix = (
f"Multiple files uploaded; only '{files[0].filename}' was used as the key.",
)
else:
notes_prefix = ()
f = files[0]
text = _decode(f.content)
rows = _read_rows(text)
if not rows:
raise ValueError(f"{f.filename}: CSV is empty.")
if len(rows) < 2:
raise ValueError(
f"{f.filename}: expected a header row plus an answer-key row "
f"(got {len(rows)} row(s))."
)
n_cols = _column_count(rows)
if n_cols < 2:
raise ValueError(
f"{f.filename}: CSV must have at least 2 columns (name + Q1)."
)
n_questions = n_cols - 1
# Row 0 is the column header; row 1 is the answer key.
key_row = _pad_row(rows[1], n_cols)
notes: list[str] = list(notes_prefix)
first_name = key_row[0]
if not _is_key_row(first_name):
notes.append(
f"{f.filename}: row 2 first column is '{first_name}', expected "
f"'正確解答' / '標準答案' — treating it as the key anyway."
)
answers: list[dict] = []
for i in range(n_questions):
cell = key_row[i + 1]
letter = normalize_letter(cell)
if letter is None:
raise ValueError(
f"{f.filename}: answer key cell at Q{i + 1} is empty or not a letter "
f"(got '{cell}'). The teacher key must be a single letter A-Z."
)
if letter == "=":
raise ValueError(
f"{f.filename}: answer key cell at Q{i + 1} is '='. The teacher key "
f"must be a concrete letter, not the correct-marker."
)
answers.append(
{
"question_number": i + 1,
"correct_answer": letter,
"explanation": None,
}
)
return ParserResult(
data={"answers": answers},
parser_name=parser_name,
notes=tuple(notes),
)