custom csv upload logic

chatkit/backend/app/parsers/__init__.py

@@ -10,7 +10,8 @@ Adding a new parser:
 
 from __future__ import annotations
 
-from .base import AnswerSheetParser, ParserResult
+from .base import AnswerSheetParser, ParserResult, get_extension
+from .csv_strict import StrictCsvParser
 from .llm_vision import LLMVisionParser
 from .pdfplumber_tables import PdfPlumberTablesParser
 from .pymupdf_tables import PyMuPDFTablesParser
@@ -20,6 +21,7 @@ AUTO = "auto"
 _REGISTRY: dict[str, AnswerSheetParser] = {
     PyMuPDFTablesParser.name: PyMuPDFTablesParser(),
     PdfPlumberTablesParser.name: PdfPlumberTablesParser(),
+    StrictCsvParser.name: StrictCsvParser(),
     LLMVisionParser.name: LLMVisionParser(),
 }
 
@@ -45,15 +47,19 @@ def get_parser(name: str) -> AnswerSheetParser:
 
 
 def pick_auto(file_bytes: bytes, filename: str, data_type: str) -> AnswerSheetParser:
-    """Auto-mode selector: try native table parsers first on text-layer PDFs,
-    otherwise fall back to the LLM vision parser.
+    """Auto-mode selector.
 
-    Questions data never tries native parsers — question text rarely fits a
-    clean table grid.
+    - Questions always use LLM vision (text rarely fits a clean grid).
+    - .csv answer-side uploads → StrictCsvParser.
+    - Native PDF tables for student/teacher answers when there's a text layer.
+    - Otherwise fall back to LLM vision.
     """
     if data_type == "questions":
         return _REGISTRY[LLMVisionParser.name]
 
+    if get_extension(filename) == ".csv":
+        return _REGISTRY[StrictCsvParser.name]
+
     for parser_name in (PyMuPDFTablesParser.name, PdfPlumberTablesParser.name):
         parser = _REGISTRY[parser_name]
         if parser.can_handle(file_bytes, filename, data_type):

chatkit/backend/app/parsers/csv_strict.py

@@ -0,0 +1,248 @@
+"""Strict CSV parser for already-tabulated answer sheets.
+
+## Contract
+
+A single CSV file represents the full answer table:
+
+- Row 0: the answer key. First cell is the name marker (e.g. `正確解答` /
+  `標準答案` / `KEY`). Remaining cells are the correct letters per question.
+- Row 1..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").
+
+There is no separate header row. Column 0 is always the name column;
+remaining columns are positional Q1..QN. Other column-header text is not
+required and is ignored if present.
+
+## Example
+
+```
+正確解答,B,A,A,C,D
+梁祐邦,A,=,C,C,D
+田瑜婕,=,A,C,B,D
+```
+
+When this file is uploaded to the `student_answers` zone, only rows 1..N are
+extracted. When uploaded to the `teacher_answers` zone, only row 0 is used.
+"""
+
+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 = answer key (first cell '正確解答'), row 1+ = 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}
+
+
+# ---------------------------------------------------------------------------
+# 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) < 2:
+            raise ValueError(
+                f"{f.filename}: expected the answer key on row 1 plus at least one "
+                f"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
+
+        # Skip the answer-key row (typically row 0). If the first row's name
+        # cell does not look like a key marker, warn but still skip it so the
+        # user's positional contract is respected.
+        first_name = rows[0][0] if rows[0] else ""
+        if not _is_key_row(first_name):
+            notes.append(
+                f"{f.filename}: row 1 first column is '{first_name}', expected "
+                f"'正確解答' / '標準答案' — treating it as the key anyway."
+            )
+
+        for idx, row in enumerate(rows[1:], start=2):
+            padded = _pad_row(row, n_cols)
+            name = padded[0]
+            if not name:
+                skipped += 1
+                continue
+            students.append(
+                {
+                    "name": name,
+                    "id": "",
+                    "answers": _row_to_answers(padded, n_questions),
+                }
+            )
+
+    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.")
+
+    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
+
+    key_row = _pad_row(rows[0], 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 1 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),
+    )

chatkit/backend/tests/test_csv_strict.py

@@ -0,0 +1,182 @@
+"""Tests for the StrictCsvParser."""
+
+from __future__ import annotations
+
+import pytest
+
+from app.parsers import get_parser, pick_auto
+from app.parsers.base import ParserFile
+
+
+def _file(text: str, name: str = "answers.csv", encoding: str = "utf-8") -> ParserFile:
+    return ParserFile(filename=name, content=text.encode(encoding))
+
+
+@pytest.mark.unit
+def test_can_handle_csv_only_for_answer_zones():
+    p = get_parser("csv_strict")
+    assert p.can_handle(b"", "x.csv", "student_answers") is True
+    assert p.can_handle(b"", "x.csv", "teacher_answers") is True
+    assert p.can_handle(b"", "x.csv", "questions") is False
+    assert p.can_handle(b"", "x.pdf", "student_answers") is False
+
+
+@pytest.mark.unit
+async def test_student_answers_basic():
+    csv_text = "正確解答,B,A,A,C,D\n梁祐邦,A,=,C,C,D\n田瑜婕,=,A,C,B,D\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    students = result.data["students"]
+    assert [s["name"] for s in students] == ["梁祐邦", "田瑜婕"]
+    assert [a["answer"] for a in students[0]["answers"]] == ["A", "=", "C", "C", "D"]
+    assert [a["answer"] for a in students[1]["answers"]] == ["=", "A", "C", "B", "D"]
+    assert all(a["question_number"] == i + 1 for i, a in enumerate(students[0]["answers"]))
+
+
+@pytest.mark.unit
+async def test_student_answers_blank_cell_becomes_none():
+    csv_text = "正確解答,B,A,A\n梁祐邦,A,,C\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
+    assert answers == ["A", None, "C"]
+
+
+@pytest.mark.unit
+async def test_student_answers_legacy_dash_becomes_equals():
+    csv_text = "正確解答,B,A,A\n梁祐邦,-,A,A\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    assert result.data["students"][0]["answers"][0]["answer"] == "="
+
+
+@pytest.mark.unit
+async def test_student_answers_skips_empty_name_rows():
+    csv_text = "正確解答,B,A\n梁祐邦,A,A\n,B,B\n田瑜婕,B,A\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    assert [s["name"] for s in result.data["students"]] == ["梁祐邦", "田瑜婕"]
+    assert any("Skipped" in n for n in result.notes)
+
+
+@pytest.mark.unit
+async def test_student_answers_warns_when_first_name_isnt_marker():
+    csv_text = "梁祐邦,A,B\n田瑜婕,B,A\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    # First row is treated as the key — student count is 1
+    assert len(result.data["students"]) == 1
+    assert result.data["students"][0]["name"] == "田瑜婕"
+    assert any("treating it as the key" in n for n in result.notes)
+
+
+@pytest.mark.unit
+async def test_student_answers_rejects_too_few_rows():
+    csv_text = "正確解答,A,B,C\n"
+    p = get_parser("csv_strict")
+    with pytest.raises(ValueError, match="at least one student"):
+        await p.parse([_file(csv_text)], "student_answers")
+
+
+@pytest.mark.unit
+async def test_student_answers_rejects_single_column():
+    csv_text = "正確解答\n梁祐邦\n"
+    p = get_parser("csv_strict")
+    with pytest.raises(ValueError, match="at least 2 columns"):
+        await p.parse([_file(csv_text)], "student_answers")
+
+
+@pytest.mark.unit
+async def test_student_answers_pads_short_rows():
+    csv_text = "正確解答,B,A,A,C\n梁祐邦,A,=\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
+    assert answers == ["A", "=", None, None]
+
+
+@pytest.mark.unit
+async def test_student_answers_arbitrary_other_column_text_ignored():
+    """Header text in row 0 cells beyond col 0 doesn't matter — col headers
+    are not validated, the file is positional."""
+    csv_text = "正確解答,B,A,A\n梁祐邦,A,=,C\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    assert len(result.data["students"]) == 1
+
+
+@pytest.mark.unit
+async def test_teacher_answers_basic():
+    csv_text = "正確解答,B,A,A,C,D\n梁祐邦,A,=,C,C,D\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "teacher_answers")
+    answers = result.data["answers"]
+    assert [a["correct_answer"] for a in answers] == ["B", "A", "A", "C", "D"]
+    assert [a["question_number"] for a in answers] == [1, 2, 3, 4, 5]
+
+
+@pytest.mark.unit
+async def test_teacher_answers_rejects_equals_in_key():
+    csv_text = "正確解答,B,=,A\n梁祐邦,A,A,A\n"
+    p = get_parser("csv_strict")
+    with pytest.raises(ValueError, match="must be a concrete letter"):
+        await p.parse([_file(csv_text)], "teacher_answers")
+
+
+@pytest.mark.unit
+async def test_teacher_answers_rejects_blank_cell_in_key():
+    csv_text = "正確解答,B,,A\n梁祐邦,A,A,A\n"
+    p = get_parser("csv_strict")
+    with pytest.raises(ValueError, match="empty or not a letter"):
+        await p.parse([_file(csv_text)], "teacher_answers")
+
+
+@pytest.mark.unit
+async def test_utf8_bom_accepted():
+    csv_text = "正確解答,B,A\n梁祐邦,A,A\n"
+    file = ParserFile(filename="answers.csv", content=("" + csv_text).encode("utf-8"))
+    p = get_parser("csv_strict")
+    result = await p.parse([file], "student_answers")
+    assert result.data["students"][0]["name"] == "梁祐邦"
+
+
+@pytest.mark.unit
+async def test_non_utf8_raises_clear_error():
+    csv_text = "正確解答,B,A\n梁祐邦,A,A\n"
+    file = ParserFile(filename="answers.csv", content=csv_text.encode("big5"))
+    p = get_parser("csv_strict")
+    with pytest.raises(ValueError, match="UTF-8"):
+        await p.parse([file], "student_answers")
+
+
+@pytest.mark.unit
+def test_pick_auto_routes_csv_to_strict():
+    chosen = pick_auto(b"", "answers.csv", "student_answers")
+    assert chosen.name == "csv_strict"
+    chosen = pick_auto(b"", "answers.csv", "teacher_answers")
+    assert chosen.name == "csv_strict"
+
+
+@pytest.mark.unit
+def test_pick_auto_questions_never_uses_csv():
+    chosen = pick_auto(b"", "answers.csv", "questions")
+    assert chosen.name == "llm_vision"
+
+
+@pytest.mark.unit
+async def test_lowercase_marker_accepted():
+    csv_text = "key,B,A\n梁祐邦,A,A\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    # Should NOT warn — "key" is in the marker set
+    assert not any("treating it as the key" in n for n in result.notes)
+    assert len(result.data["students"]) == 1
+
+
+@pytest.mark.unit
+async def test_excel_trailing_empty_columns_ignored():
+    csv_text = "正確解答,B,A,A,,\n梁祐邦,A,=,C,,\n"
+    p = get_parser("csv_strict")
+    result = await p.parse([_file(csv_text)], "student_answers")
+    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
+    assert answers == ["A", "=", "C"]