simplify the answer parsing tool selection

chatkit/backend/app/parsers/__init__.py

@@ -13,14 +13,12 @@ from __future__ import annotations
 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
+from .native_pdf_tables import NativePdfParser
 
 AUTO = "auto"
 
 _REGISTRY: dict[str, AnswerSheetParser] = {
-    PyMuPDFTablesParser.name: PyMuPDFTablesParser(),
-    PdfPlumberTablesParser.name: PdfPlumberTablesParser(),
+    NativePdfParser.name: NativePdfParser(),
     StrictCsvParser.name: StrictCsvParser(),
     LLMVisionParser.name: LLMVisionParser(),
 }
@@ -64,10 +62,9 @@ def pick_auto(file_bytes: bytes, filename: str, data_type: str) -> AnswerSheetPa
     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, routing_data_type):
-            return parser
+    native = _REGISTRY[NativePdfParser.name]
+    if native.can_handle(file_bytes, filename, routing_data_type):
+        return native
     return _REGISTRY[LLMVisionParser.name]
 
 

chatkit/backend/app/parsers/native_pdf_tables.py

@@ -0,0 +1,77 @@
+"""Facade for native-PDF table extraction.
+
+Tries PyMuPDF first; if it raises or yields no tables, falls back to
+pdfplumber. If both fail, raises a single user-facing error message
+recommending the CSV path.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from .base import AnswerSheetParser, ParserFile, ParserResult
+from .pdfplumber_tables import PdfPlumberTablesParser
+from .pymupdf_tables import PyMuPDFTablesParser
+
+logger = logging.getLogger(__name__)
+
+
+class NativePdfParser:
+    name = "native_pdf_tables"
+    display_name = "Native PDF tables"
+    description = (
+        "Deterministic table extraction for native (digital) PDFs such as "
+        "考生作答一覽表 exports. Internally tries PyMuPDF first and falls "
+        "back to pdfplumber. Fast, no LLM call, no API quota."
+    )
+
+    def __init__(self) -> None:
+        self._engines: list[AnswerSheetParser] = [
+            PyMuPDFTablesParser(),
+            PdfPlumberTablesParser(),
+        ]
+
+    def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool:
+        return any(
+            engine.can_handle(file_bytes, filename, data_type)
+            for engine in self._engines
+        )
+
+    async def parse(
+        self,
+        files: list[ParserFile],
+        data_type: str,
+        description: str = "",
+        model: str = "gpt-5.4",
+    ) -> ParserResult:
+        errors: list[str] = []
+        for engine in self._engines:
+            try:
+                result = await engine.parse(files, data_type, description, model)
+                logger.info(
+                    "native_pdf_tables: %s succeeded for data_type=%s",
+                    engine.name,
+                    data_type,
+                )
+                return ParserResult(
+                    data=result.data,
+                    parser_name=self.name,
+                    notes=(
+                        *result.notes,
+                        f"native PDF engine: {engine.name}",
+                    ),
+                )
+            except Exception as exc:  # noqa: BLE001 — try every engine
+                logger.warning(
+                    "native_pdf_tables: %s failed (%s); trying next engine",
+                    engine.name,
+                    exc,
+                )
+                errors.append(f"{engine.name}: {exc}")
+                continue
+
+        joined = "; ".join(errors) if errors else "no suitable engine"
+        raise ValueError(
+            "Unable to parse the PDF format. Please convert to CSV using our "
+            f"template and re-upload. (engines tried: {joined})"
+        )

chatkit/backend/tests/test_native_parsers.py

@@ -1,4 +1,8 @@
-"""Integration tests for native PDF table parsers against sample data."""
+"""Integration tests for native PDF table extraction.
+
+Covers the public `native_pdf_tables` facade (registered) and exercises the
+underlying engine classes directly to keep regression coverage on each.
+"""
 
 from __future__ import annotations
 
@@ -8,6 +12,8 @@ import pytest
 
 from app.parsers import get_parser, pick_auto
 from app.parsers.base import ParserFile
+from app.parsers.pdfplumber_tables import PdfPlumberTablesParser
+from app.parsers.pymupdf_tables import PyMuPDFTablesParser
 
 SAMPLE_PDF = Path(
     "/Users/yu.c/Desktop/npo-ca-classlens-data copy/example-2/9A作答一覽表.pdf"
@@ -21,19 +27,25 @@ def sample_file() -> ParserFile:
     return ParserFile(filename=SAMPLE_PDF.name, content=SAMPLE_PDF.read_bytes())
 
 
+# ---------------------------------------------------------------------------
+# Engine-level coverage (PyMuPDF + pdfplumber)
+# ---------------------------------------------------------------------------
+
+
+def _engines():
+    return [PyMuPDFTablesParser(), PdfPlumberTablesParser()]
+
+
 @pytest.mark.integration
-@pytest.mark.parametrize("parser_name", ["pymupdf_tables", "pdfplumber_tables"])
-async def test_parse_student_answers(sample_file: ParserFile, parser_name: str):
-    parser = get_parser(parser_name)
-    result = await parser.parse([sample_file], "student_answers")
+@pytest.mark.parametrize("engine", _engines(), ids=lambda e: e.name)
+async def test_engine_parses_student_answers(sample_file: ParserFile, engine):
+    result = await engine.parse([sample_file], "student_answers")
     students = result.data["students"]
     assert len(students) == 7, f"expected 7 students, got {len(students)}"
-    # Every student should have ~49 answers
     for s in students:
         assert 40 <= len(s["answers"]) <= 50
         assert s["name"]
         assert s["id"]
-    # Spot-check a known student
     by_name = {s["name"]: s for s in students}
     assert "陳筱琳" in by_name
     assert by_name["陳筱琳"]["id"] == "902-13"
@@ -43,34 +55,56 @@ async def test_parse_student_answers(sample_file: ParserFile, parser_name: str):
 
 
 @pytest.mark.integration
-@pytest.mark.parametrize("parser_name", ["pymupdf_tables", "pdfplumber_tables"])
-async def test_parse_teacher_answers(sample_file: ParserFile, parser_name: str):
-    parser = get_parser(parser_name)
-    result = await parser.parse([sample_file], "teacher_answers")
+@pytest.mark.parametrize("engine", _engines(), ids=lambda e: e.name)
+async def test_engine_parses_teacher_answers(sample_file: ParserFile, engine):
+    result = await engine.parse([sample_file], "teacher_answers")
     answers = result.data["answers"]
-    # Answer key from the PDF is 49 letters (non-None)
     assert 40 <= len(answers) <= 50
-    # From the PDF: "BAACCACACB..."
     first_10 = "".join(a["correct_answer"] for a in answers[:10])
     assert first_10 == "BAACCACACB"
 
 
+# ---------------------------------------------------------------------------
+# Facade behavior
+# ---------------------------------------------------------------------------
+
+
 @pytest.mark.integration
-async def test_auto_picks_native_for_text_pdf(sample_file: ParserFile):
+async def test_native_facade_succeeds_via_first_engine(sample_file: ParserFile):
+    parser = get_parser("native_pdf_tables")
+    result = await parser.parse([sample_file], "student_answers")
+    assert result.parser_name == "native_pdf_tables"
+    assert any("native PDF engine" in n for n in result.notes)
+    assert len(result.data["students"]) == 7
+
+
+@pytest.mark.unit
+async def test_native_facade_raises_csv_hint_when_all_engines_fail():
+    parser = get_parser("native_pdf_tables")
+    fake = ParserFile(filename="not-a.pdf", content=b"%PDF-1.4 garbage")
+    with pytest.raises(ValueError, match=r"(?i)please convert to csv"):
+        await parser.parse([fake], "student_answers")
+
+
+@pytest.mark.unit
+def test_native_facade_refuses_image_file():
+    parser = get_parser("native_pdf_tables")
+    png_bytes = b"\x89PNG\r\n\x1a\n"
+    assert parser.can_handle(png_bytes, "scan.png", "student_answers") is False
+
+
+# ---------------------------------------------------------------------------
+# Auto-router
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_auto_picks_native_facade_for_text_pdf(sample_file: ParserFile):
     chosen = pick_auto(sample_file.content, sample_file.filename, "student_answers")
-    assert chosen.name in ("pymupdf_tables", "pdfplumber_tables")
+    assert chosen.name == "native_pdf_tables"
 
 
 @pytest.mark.integration
 async def test_auto_falls_back_to_llm_for_questions(sample_file: ParserFile):
-    # Native parsers don't handle `questions`; auto should pick LLM.
     chosen = pick_auto(sample_file.content, sample_file.filename, "questions")
     assert chosen.name == "llm_vision"
-
-
-@pytest.mark.integration
-async def test_native_parser_refuses_image_file():
-    parser = get_parser("pymupdf_tables")
-    png_bytes = b"\x89PNG\r\n\x1a\n"  # header-only fake PNG
-    fake = ParserFile(filename="scan.png", content=png_bytes)
-    assert parser.can_handle(fake.content, fake.filename, "student_answers") is False