rebuild the pdf parsing logic

chatkit/backend/app/file_processor.py

@@ -1,157 +1,19 @@
-"""File processing: upload files and extract structured Q&A via GPT-4 Vision / GPT-4o."""
+"""File processing orchestrator.
+
+Delegates extraction to a pluggable parser chosen by name or auto-picked.
+Saves the parsed structured data to the database, along with metadata about
+which parser produced it.
+"""
 
 from __future__ import annotations
 
-import base64
 import json
-import mimetypes
-from pathlib import Path
-from typing import Optional
 
-import fitz  # pymupdf
-from openai import AsyncOpenAI
 from fastapi import UploadFile
 
-from .config import get_settings
-from .database import save_parsed_data, delete_parsed_data
-
-
-# Extensions that should be read as text, not sent as images
-TEXT_EXTENSIONS = {".txt", ".csv", ".tsv", ".json", ".xml", ".md", ".html", ".htm", ".log"}
-
-# Image extensions with their MIME types
-IMAGE_MIME = {
-    ".png": "image/png",
-    ".jpg": "image/jpeg",
-    ".jpeg": "image/jpeg",
-    ".gif": "image/gif",
-    ".bmp": "image/bmp",
-    ".webp": "image/webp",
-    ".tiff": "image/tiff",
-    ".tif": "image/tiff",
-}
-
-NO_FABRICATION_RULE = """
-CRITICAL: Only extract data that is ACTUALLY present in the provided content.
-Do NOT invent, fabricate, or hallucinate any data.
-If the content is empty, unreadable, or does not contain the expected data, return an empty result like {"questions": []} or {"students": []} or {"answers": []}.
-"""
-
-IMAGE_DESCRIPTION_RULE = """
-IMPORTANT - Images, Diagrams, Charts, and Figures:
-- If a question contains or references an image, diagram, chart, graph, map, table, or any visual element, you MUST describe it in detail as part of the question text.
-- Use the format: [圖片描述: ...detailed description...] inserted where the image appears in the question.
-- The description must be detailed enough that someone who CANNOT see the image can still fully understand and answer the question.
-- Include all relevant data points, labels, axes, values, shapes, positions, colors, relationships, and any text visible in the image.
-- For charts/graphs: describe the type, axes labels, data values, trends, and all visible data points.
-- For diagrams: describe all components, connections, labels, measurements, and spatial relationships.
-- For maps: describe locations, boundaries, labels, and any marked features.
-- For tables embedded as images: transcribe the full table content.
-"""
-
-EXTRACTION_PROMPTS = {
-    "questions": """You are an exam data extraction expert. Extract all exam questions from the provided content.
-
-This system supports SINGLE-CHOICE questions only (one correct letter: A, B, C, D, ...).
-
-Return a JSON object with this structure:
-{
-  "questions": [
-    {
-      "number": 1,
-      "text": "the full question text (including [圖片描述: ...] for any images/diagrams)",
-      "options": ["A) ...", "B) ...", "C) ...", "D) ..."]
-    }
-  ]
-}
-
-Rules:
-- Extract EVERY question you can see
-- Preserve the original language of the questions
-- Always include the options exactly as written; every question must have options
-- Do NOT include fields other than number/text/options
-""" + IMAGE_DESCRIPTION_RULE + NO_FABRICATION_RULE + "- Return valid JSON only",
-
-    "student_answers": """You are an exam data extraction expert. Extract all student answers from the provided content.
-
-This system supports SINGLE-CHOICE questions only.
-
-Return a JSON object with this structure:
-{
-  "students": [
-    {
-      "name": "student name or ID",
-      "id": "student ID if visible",
-      "answers": [
-        {
-          "question_number": 1,
-          "answer": "A" | "B" | "C" | ... | "=" | null
-        }
-      ]
-    }
-  ]
-}
-
-Rules:
-- Extract answers for EVERY student visible
-- Answer format (use exactly one of these):
-  - A single uppercase letter (A, B, C, D, ...) — the letter the student chose
-  - "=" — the student answered correctly (e.g., source uses "-" / "✓" / "correct" / matches the answer key)
-  - null — the student left the question blank
-- Use the student's name or ID as shown in the document
-""" + NO_FABRICATION_RULE + "- Return valid JSON only",
-
-    "teacher_answers": """You are an exam data extraction expert. Extract the correct/model answers (answer key) from the provided content.
-
-This system supports SINGLE-CHOICE questions only.
-
-Return a JSON object with this structure:
-{
-  "answers": [
-    {
-      "question_number": 1,
-      "correct_answer": "A" | "B" | "C" | ...,
-      "explanation": "explanation if provided, otherwise null"
-    }
-  ]
-}
-
-Rules:
-- Extract the correct answer for EVERY question
-- correct_answer MUST be a single uppercase letter (A, B, C, D, ...)
-- Include explanations if they are provided
-""" + NO_FABRICATION_RULE + "- Return valid JSON only",
-}
-
-
-def pdf_to_images(pdf_bytes: bytes) -> list[bytes]:
-    """Convert PDF bytes to a list of PNG image bytes, one per page."""
-    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
-    images = []
-    for page in doc:
-        # Render at 2x resolution for better OCR quality
-        pix = page.get_pixmap(matrix=fitz.Matrix(2, 2))
-        images.append(pix.tobytes("png"))
-    doc.close()
-    return images
-
-
-def try_read_as_text(file_bytes: bytes, filename: str) -> Optional[str]:
-    """Try to decode file bytes as text. Returns None if not text."""
-    ext = Path(filename).suffix.lower()
-    if ext in TEXT_EXTENSIONS:
-        for encoding in ("utf-8", "utf-8-sig", "big5", "gb2312", "shift_jis", "latin-1"):
-            try:
-                return file_bytes.decode(encoding)
-            except (UnicodeDecodeError, ValueError):
-                continue
-    return None
-
-
-def get_image_mime(filename: str) -> Optional[str]:
-    """Get MIME type for image files. Returns None if not a known image type."""
-    ext = Path(filename).suffix.lower()
-    return IMAGE_MIME.get(ext)
+from .database import delete_parsed_data, save_parsed_data
+from .parsers import AUTO, get_parser, pick_auto
+from .parsers.base import ParserFile, ParserResult
 
 
 async def process_uploaded_files(
@@ -160,124 +22,78 @@ async def process_uploaded_files(
     session_id: int,
     description: str = "",
     model: str = "gpt-5.4",
+    parser: str = AUTO,
 ) -> dict:
-    """
-    Process uploaded files: convert to images/text, then extract data via GPT-4o.
-    Returns the structured data extracted.
-    """
-    image_parts: list[dict] = []      # image content parts for GPT
-    text_parts: list[str] = []         # text content from text files
-    source_files: list[str] = []
+    """Parse uploaded files and persist the structured result.
 
+    Returns the parsed data dict with a `_meta` entry describing which parser ran.
+    """
+    parser_files: list[ParserFile] = []
     for file in files:
-        file_bytes = await file.read()
-        filename = file.filename or "unknown"
-
-        if not file_bytes:
-            raise ValueError(f"File '{filename}' is empty (0 bytes). Please upload a valid file.")
-
-        ext = Path(filename).suffix.lower()
-        source_files.append(filename)
-
-        if ext == ".pdf":
-            page_images = pdf_to_images(file_bytes)
-            for i, img in enumerate(page_images):
-                b64 = base64.b64encode(img).decode()
-                image_parts.append({
-                    "type": "image_url",
-                    "image_url": {"url": f"data:image/png;base64,{b64}", "detail": "high"},
-                })
-        else:
-            # Try to read as text first
-            text_content = try_read_as_text(file_bytes, filename)
-            if text_content is not None:
-                text_parts.append(f"--- Content from {filename} ---\n{text_content}")
-            else:
-                # Try as image with correct MIME type
-                mime = get_image_mime(filename) or "image/png"
-                b64 = base64.b64encode(file_bytes).decode()
-                image_parts.append({
-                    "type": "image_url",
-                    "image_url": {"url": f"data:{mime};base64,{b64}", "detail": "high"},
-                })
+        content = await file.read()
+        if not content:
+            raise ValueError(f"File '{file.filename}' is empty.")
+        parser_files.append(
+            ParserFile(filename=file.filename or "unknown", content=content)
+        )
 
-    if not image_parts and not text_parts:
-        raise ValueError("No valid content found in uploaded files")
+    if not parser_files:
+        raise ValueError("No files uploaded")
 
-    # Extract data using selected model
-    structured = await extract_data(
-        image_parts=image_parts,
-        text_parts=text_parts,
-        data_type=data_type,
-        description=description,
-        model=model,
-    )
+    result = await _run_with_fallback(parser_files, data_type, description, model, parser)
 
-    # Clear old data of this type for the session, then save new
     await delete_parsed_data(session_id, data_type)
 
-    raw_text = json.dumps({"source_files": source_files}, ensure_ascii=False)
-    await save_parsed_data(session_id, data_type, files[0].filename or "upload", raw_text, structured)
+    source_files = [f.filename for f in parser_files]
+    raw_text = json.dumps(
+        {
+            "source_files": source_files,
+            "parser": result.parser_name,
+            "notes": list(result.notes),
+        },
+        ensure_ascii=False,
+    )
+    await save_parsed_data(
+        session_id,
+        data_type,
+        parser_files[0].filename,
+        raw_text,
+        result.data,
+    )
 
-    return structured
+    return {
+        **result.data,
+        "_meta": {
+            "parser": result.parser_name,
+            "notes": list(result.notes),
+        },
+    }
 
 
-async def extract_data(
-    image_parts: list[dict],
-    text_parts: list[str],
+async def _run_with_fallback(
+    parser_files: list[ParserFile],
     data_type: str,
-    description: str = "",
-    model: str = "gpt-5.4",
-) -> dict:
-    """Call GPT-4o to extract structured data from images and/or text."""
-    settings = get_settings()
-    client = AsyncOpenAI(api_key=settings.openai_api_key)
-
-    system_prompt = EXTRACTION_PROMPTS.get(data_type)
-    if not system_prompt:
-        raise ValueError(f"Unknown data_type: {data_type}")
-
-    # Build user message content
-    user_text = f"Please extract the {data_type.replace('_', ' ')} from the provided content."
-    if description.strip():
-        user_text = f"Context about this data: {description.strip()}\n\n{user_text}"
-
-    # If we have text content, include it
-    if text_parts:
-        combined_text = "\n\n".join(text_parts)
-        user_text += f"\n\n--- FILE CONTENT ---\n{combined_text}"
-
-    content_parts: list[dict] = [{"type": "text", "text": user_text}]
-    content_parts.extend(image_parts)
-
-    messages = [
-        {"role": "system", "content": system_prompt},
-        {"role": "user", "content": content_parts},
-    ]
-
-    # Build kwargs — some models don't support all parameters
-    kwargs: dict = {
-        "model": model,
-        "messages": messages,
-        "max_completion_tokens": 16384,
-    }
-    # json_object response_format may not work on all models; try with it first
-    try:
-        response = await client.chat.completions.create(
-            **kwargs,
-            response_format={"type": "json_object"},
-        )
-    except Exception:
-        # Fallback: no response_format, rely on prompt to return JSON
-        response = await client.chat.completions.create(**kwargs)
-
-    choice = response.choices[0]
-    content = choice.message.content
-    if not content:
-        reason = getattr(choice, "finish_reason", "unknown")
-        refusal = getattr(choice.message, "refusal", None)
-        detail = f"finish_reason={reason}"
-        if refusal:
-            detail += f", refusal={refusal}"
-        raise ValueError(f"Model returned empty content ({detail}). Try a different model.")
-    return json.loads(content)
+    description: str,
+    model: str,
+    parser_name: str,
+) -> ParserResult:
+    """Run the selected parser; in auto mode, fall back to LLM on failure."""
+    if parser_name == AUTO:
+        first_file = parser_files[0]
+        chosen = pick_auto(first_file.content, first_file.filename, data_type)
+        try:
+            return await chosen.parse(parser_files, data_type, description, model)
+        except Exception as primary_err:
+            if chosen.name == "llm_vision":
+                raise
+            # Fallback to LLM vision on any native-parser failure
+            llm = get_parser("llm_vision")
+            result = await llm.parse(parser_files, data_type, description, model)
+            return ParserResult(
+                data=result.data,
+                parser_name=result.parser_name,
+                notes=(f"auto-fallback from {chosen.name}: {primary_err}", *result.notes),
+            )
+
+    chosen = get_parser(parser_name)
+    return await chosen.parse(parser_files, data_type, description, model)

chatkit/backend/app/main.py

@@ -36,6 +36,7 @@ from .database import (
 from .auth import get_current_user, register_user, login_user
 from .file_processor import process_uploaded_files
 from .answer_grid import from_dict as grid_from_dict, seed_from_parsed, to_dict as grid_to_dict
+from .parsers import AUTO, list_parsers
 from .report_generator import generate_student_report, build_student_markdown
 
 STATIC_DIR = Path(__file__).parent.parent / "static"
@@ -125,6 +126,12 @@ async def api_get_session(session_id: int, user=Depends(get_current_user)):
 # File Upload & Processing Endpoints
 # =============================================================================
 
+@app.get("/api/parsers")
+async def api_list_parsers():
+    """List available parser backends for the upload UI."""
+    return {"parsers": list_parsers()}
+
+
 @app.post("/api/sessions/{session_id}/upload")
 async def api_upload_files(
     session_id: int,
@@ -132,6 +139,7 @@ async def api_upload_files(
     files: list[UploadFile] = File(...),
     description: str = Form(""),
     model: str = Form("gpt-5.4"),
+    parser: str = Form(AUTO),
     user=Depends(get_current_user),
 ):
     """Upload files for a session. data_type: 'questions', 'student_answers', or 'teacher_answers'."""
@@ -143,7 +151,14 @@ async def api_upload_files(
         raise HTTPException(status_code=400, detail="data_type must be 'questions', 'student_answers', or 'teacher_answers'")
 
     try:
-        structured = await process_uploaded_files(files, data_type, session_id, description=description, model=model)
+        structured = await process_uploaded_files(
+            files,
+            data_type,
+            session_id,
+            description=description,
+            model=model,
+            parser=parser,
+        )
         await update_session_status(session_id, "processed")
         # Invalidate any previously-confirmed answer grid since inputs changed
         await delete_parsed_data(session_id, ANSWER_GRID_DATA_TYPE)

chatkit/backend/app/parsers/__init__.py

@@ -0,0 +1,71 @@
+"""Pluggable parser registry for answer-sheet extraction.
+
+Auto mode: picks the best parser based on file type and data_type.
+Manual mode: caller specifies parser name from `list_parsers()`.
+
+Adding a new parser:
+1. Create a module that exposes a subclass of `AnswerSheetParser`.
+2. Import and register it below in `_REGISTRY`.
+"""
+
+from __future__ import annotations
+
+from .base import AnswerSheetParser, ParserResult
+from .llm_vision import LLMVisionParser
+from .pdfplumber_tables import PdfPlumberTablesParser
+from .pymupdf_tables import PyMuPDFTablesParser
+
+AUTO = "auto"
+
+_REGISTRY: dict[str, AnswerSheetParser] = {
+    PyMuPDFTablesParser.name: PyMuPDFTablesParser(),
+    PdfPlumberTablesParser.name: PdfPlumberTablesParser(),
+    LLMVisionParser.name: LLMVisionParser(),
+}
+
+
+def list_parsers() -> list[dict[str, str]]:
+    """Return parser metadata for UI dropdowns."""
+    out: list[dict[str, str]] = [
+        {
+            "id": AUTO,
+            "name": "Auto (recommended)",
+            "description": "System picks the best parser for each file.",
+        }
+    ]
+    for p in _REGISTRY.values():
+        out.append({"id": p.name, "name": p.display_name, "description": p.description})
+    return out
+
+
+def get_parser(name: str) -> AnswerSheetParser:
+    if name not in _REGISTRY:
+        raise ValueError(f"Unknown parser: {name}")
+    return _REGISTRY[name]
+
+
+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.
+
+    Questions data never tries native parsers — question text rarely fits a
+    clean table grid.
+    """
+    if data_type == "questions":
+        return _REGISTRY[LLMVisionParser.name]
+
+    for parser_name in (PyMuPDFTablesParser.name, PdfPlumberTablesParser.name):
+        parser = _REGISTRY[parser_name]
+        if parser.can_handle(file_bytes, filename, data_type):
+            return parser
+    return _REGISTRY[LLMVisionParser.name]
+
+
+__all__ = [
+    "AUTO",
+    "AnswerSheetParser",
+    "ParserResult",
+    "get_parser",
+    "list_parsers",
+    "pick_auto",
+]

chatkit/backend/app/parsers/_table_common.py

@@ -0,0 +1,217 @@
+"""Shared helpers for native-table parsers.
+
+Answer-sheet PDFs (e.g. the `考生作答一覽表` format) contain a grid where each
+logical student row is split by the extractor into:
+  - a "header" row carrying class / seat / name (and per-row score cells)
+  - one or more follow-up "data" rows carrying the letter strings
+
+Example extracted rows:
+    ['902', '13', '陳筱琳', '', '', '', '', ..., '51', None, None, None, '51', ...]
+    [None,  None, None,     '==CBBBA==C', '===B======', 'B=B==BB==B', ...]
+
+We walk the rows statefully, grouping follow-up data rows into the most recent
+header, then extract letters from the combined cell set.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Optional
+
+from ..answer_grid import normalize_letter
+
+# Markers that identify the answer-key header row.
+ANSWER_KEY_MARKERS = ("預設標準答案", "標準答案", "answer key")
+
+# Chars kept when extracting letters from a cell.
+_ANSWER_CELL_ALLOWED = re.compile(r"[A-Za-z=\-–—]")
+
+# A cell counts as a "letter cell" if most of its chars survive filtering and
+# it has at least 3 chars. This prevents single-char artifacts like a class
+# number from being mis-read as one letter.
+_LETTER_CELL_MIN_LEN = 3
+
+
+@dataclass
+class _Group:
+    """A header row plus any trailing data rows that belong to it."""
+
+    header: list
+    data_rows: list[list] = field(default_factory=list)
+
+
+def _cell_text(cell) -> str:
+    if cell is None:
+        return ""
+    return str(cell).strip()
+
+
+def _is_classlike(value: str) -> bool:
+    return bool(re.fullmatch(r"\d{2,4}", value))
+
+
+def _is_empty_cell(cell) -> bool:
+    return _cell_text(cell) == ""
+
+
+def _is_answer_key_header(row: list) -> bool:
+    text = " ".join(_cell_text(c) for c in row)
+    return any(marker in text for marker in ANSWER_KEY_MARKERS)
+
+
+def _is_student_header(row: list) -> bool:
+    """Header row has a class-like number + a non-empty name cell."""
+    first_vals = [c for c in row if not _is_empty_cell(c)]
+    if len(first_vals) < 2:
+        return False
+    # Look for class pattern in the leading cells
+    head = [_cell_text(c) for c in row[:3]]
+    head_nonempty = [h for h in head if h]
+    if not head_nonempty:
+        return False
+    if not _is_classlike(head_nonempty[0]):
+        return False
+    # At least one later value should look like a name (non-letter-cell)
+    return True
+
+
+def _is_data_row(row: list) -> bool:
+    """A data row has empty name/class columns and letter cells elsewhere."""
+    head = row[:3]
+    if any(not _is_empty_cell(c) for c in head):
+        return False
+    return any(_looks_like_letter_cell(c) for c in row)
+
+
+def _looks_like_letter_cell(cell) -> bool:
+    text = _cell_text(cell)
+    if len(text) < _LETTER_CELL_MIN_LEN:
+        return False
+    kept = "".join(_ANSWER_CELL_ALLOWED.findall(text))
+    return len(kept) >= _LETTER_CELL_MIN_LEN and len(kept) >= len(text) * 0.7
+
+
+def _extract_letters_from_cell(cell) -> list[Optional[str]]:
+    clean = "".join(_ANSWER_CELL_ALLOWED.findall(_cell_text(cell)))
+    return [normalize_letter(ch) for ch in clean]
+
+
+def _collect_letters(group: _Group) -> list[Optional[str]]:
+    """Concatenate letter cells across header + data rows, in order."""
+    letters: list[Optional[str]] = []
+    for row in (group.header, *group.data_rows):
+        for cell in row:
+            if _looks_like_letter_cell(cell):
+                letters.extend(_extract_letters_from_cell(cell))
+    return letters
+
+
+def _extract_name(row: list) -> Optional[str]:
+    """Return the first non-empty, non-numeric cell in positions 0-4."""
+    for cell in row[:5]:
+        text = _cell_text(cell)
+        if not text:
+            continue
+        if re.fullmatch(r"-?\d+(\.\d+)?", text):
+            continue
+        if _looks_like_letter_cell(text):
+            continue
+        return text
+    return None
+
+
+def _extract_student_id(row: list) -> Optional[str]:
+    """Return 'class-seat' from the header row when present."""
+    head = [_cell_text(c) for c in row[:3]]
+    class_val = next((h for h in head if _is_classlike(h)), None)
+    if not class_val:
+        return None
+    seat_val = next(
+        (h for h in head if h and h != class_val and re.fullmatch(r"\d{1,3}", h)),
+        None,
+    )
+    return f"{class_val}-{seat_val}" if seat_val else class_val
+
+
+def _group_rows(rows: list[list]) -> tuple[Optional[_Group], list[_Group]]:
+    """Walk rows and group each header with its trailing data rows.
+
+    Returns (answer_key_group, student_groups).
+    """
+    answer_key: Optional[_Group] = None
+    students: list[_Group] = []
+    current: Optional[_Group] = None
+    bucket: Optional[str] = None  # "key" or "student"
+
+    for row in rows:
+        if _is_answer_key_header(row):
+            current = _Group(header=row)
+            answer_key = current
+            bucket = "key"
+            continue
+        if _is_student_header(row):
+            current = _Group(header=row)
+            students.append(current)
+            bucket = "student"
+            continue
+        if current is not None and _is_data_row(row):
+            current.data_rows.append(row)
+            continue
+        # Non-data, non-header row ends the current group.
+        if bucket is not None and not _is_data_row(row):
+            # Only reset on rows with visible header-like content (avoid noise rows)
+            if any(not _is_empty_cell(c) for c in row[:3]):
+                current = None
+                bucket = None
+
+    return answer_key, students
+
+
+def table_to_student_answers(rows: list[list]) -> dict:
+    _, student_groups = _group_rows(rows)
+    students: list[dict] = []
+    for g in student_groups:
+        name = _extract_name(g.header)
+        if not name:
+            continue
+        letters = _collect_letters(g)
+        if not letters:
+            continue
+        students.append(
+            {
+                "name": name,
+                "id": _extract_student_id(g.header) or "",
+                "answers": [
+                    {"question_number": i + 1, "answer": letter}
+                    for i, letter in enumerate(letters)
+                ],
+            }
+        )
+    return {"students": students}
+
+
+def table_to_teacher_answers(rows: list[list]) -> dict:
+    key_group, _ = _group_rows(rows)
+    if key_group is None:
+        return {"answers": []}
+    letters = _collect_letters(key_group)
+    return {
+        "answers": [
+            {
+                "question_number": i + 1,
+                "correct_answer": letter,
+                "explanation": None,
+            }
+            for i, letter in enumerate(letters)
+            if letter is not None
+        ]
+    }
+
+
+def table_to_data(rows: list[list], data_type: str) -> dict:
+    if data_type == "student_answers":
+        return table_to_student_answers(rows)
+    if data_type == "teacher_answers":
+        return table_to_teacher_answers(rows)
+    raise ValueError(f"Native table parser does not support data_type={data_type}")

chatkit/backend/app/parsers/base.py

@@ -0,0 +1,111 @@
+"""Parser base types and shared file utilities."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional, Protocol
+
+TEXT_EXTENSIONS = {".txt", ".csv", ".tsv", ".json", ".xml", ".md", ".html", ".htm", ".log"}
+
+IMAGE_MIME = {
+    ".png": "image/png",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".gif": "image/gif",
+    ".bmp": "image/bmp",
+    ".webp": "image/webp",
+    ".tiff": "image/tiff",
+    ".tif": "image/tiff",
+}
+
+
+@dataclass(frozen=True)
+class ParserFile:
+    """A file uploaded for parsing."""
+
+    filename: str
+    content: bytes
+
+
+@dataclass(frozen=True)
+class ParserResult:
+    """Structured output of a parser run."""
+
+    data: dict
+    parser_name: str
+    notes: tuple[str, ...] = field(default_factory=tuple)
+
+
+class AnswerSheetParser(Protocol):
+    """Protocol for answer-sheet parsers.
+
+    Implementations should be stateless — one instance per process is fine.
+    """
+
+    name: str
+    display_name: str
+    description: str
+
+    def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool: ...
+
+    async def parse(
+        self,
+        files: list[ParserFile],
+        data_type: str,
+        description: str = "",
+        model: str = "gpt-5.4",
+    ) -> ParserResult: ...
+
+
+# ---------------------------------------------------------------------------
+# Shared file helpers
+# ---------------------------------------------------------------------------
+
+
+def get_extension(filename: str) -> str:
+    return Path(filename).suffix.lower()
+
+
+def is_pdf(filename: str) -> bool:
+    return get_extension(filename) == ".pdf"
+
+
+def is_image(filename: str) -> bool:
+    return get_extension(filename) in IMAGE_MIME
+
+
+def get_image_mime(filename: str) -> Optional[str]:
+    return IMAGE_MIME.get(get_extension(filename))
+
+
+def try_read_as_text(file_bytes: bytes, filename: str) -> Optional[str]:
+    """Decode text files using a list of likely encodings."""
+    if get_extension(filename) not in TEXT_EXTENSIONS:
+        return None
+    for encoding in ("utf-8", "utf-8-sig", "big5", "gb2312", "shift_jis", "latin-1"):
+        try:
+            return file_bytes.decode(encoding)
+        except (UnicodeDecodeError, ValueError):
+            continue
+    return None
+
+
+def pdf_has_text_layer(pdf_bytes: bytes) -> bool:
+    """True if the PDF has an extractable text layer (i.e., not an image scan)."""
+    try:
+        import fitz  # local import so parsers without pymupdf don't die
+    except ImportError:
+        return False
+    try:
+        doc = fitz.open(stream=pdf_bytes, filetype="pdf")
+        try:
+            for page in doc:
+                text = page.get_text("text").strip()
+                if text:
+                    return True
+            return False
+        finally:
+            doc.close()
+    except Exception:
+        return False

chatkit/backend/app/parsers/llm_vision.py

@@ -0,0 +1,231 @@
+"""LLM vision parser: converts files to images/text and calls OpenAI to extract
+structured JSON. Works on any input format (PDF, images, text files)."""
+
+from __future__ import annotations
+
+import base64
+import json
+from typing import Optional
+
+from openai import AsyncOpenAI
+
+from ..config import get_settings
+from .base import (
+    AnswerSheetParser,
+    ParserFile,
+    ParserResult,
+    get_image_mime,
+    is_pdf,
+    try_read_as_text,
+)
+
+NO_FABRICATION_RULE = """
+CRITICAL: Only extract data that is ACTUALLY present in the provided content.
+Do NOT invent, fabricate, or hallucinate any data.
+If the content is empty, unreadable, or does not contain the expected data, return an empty result like {"questions": []} or {"students": []} or {"answers": []}.
+"""
+
+IMAGE_DESCRIPTION_RULE = """
+IMPORTANT - Images, Diagrams, Charts, and Figures:
+- If a question contains or references an image, diagram, chart, graph, map, table, or any visual element, you MUST describe it in detail as part of the question text.
+- Use the format: [圖片描述: ...detailed description...] inserted where the image appears in the question.
+- The description must be detailed enough that someone who CANNOT see the image can still fully understand and answer the question.
+- Include all relevant data points, labels, axes, values, shapes, positions, colors, relationships, and any text visible in the image.
+"""
+
+EXTRACTION_PROMPTS = {
+    "questions": """You are an exam data extraction expert. Extract all exam questions from the provided content.
+
+This system supports SINGLE-CHOICE questions only (one correct letter: A, B, C, D, ...).
+
+Return a JSON object with this structure:
+{
+  "questions": [
+    {
+      "number": 1,
+      "text": "the full question text (including [圖片描述: ...] for any images/diagrams)",
+      "options": ["A) ...", "B) ...", "C) ...", "D) ..."]
+    }
+  ]
+}
+
+Rules:
+- Extract EVERY question you can see
+- Preserve the original language of the questions
+- Always include the options exactly as written; every question must have options
+- Do NOT include fields other than number/text/options
+""" + IMAGE_DESCRIPTION_RULE + NO_FABRICATION_RULE + "- Return valid JSON only",
+    "student_answers": """You are an exam data extraction expert. Extract all student answers from the provided content.
+
+This system supports SINGLE-CHOICE questions only.
+
+Return a JSON object with this structure:
+{
+  "students": [
+    {
+      "name": "student name or ID",
+      "id": "student ID if visible",
+      "answers": [
+        {
+          "question_number": 1,
+          "answer": "A" | "B" | "C" | ... | "=" | null
+        }
+      ]
+    }
+  ]
+}
+
+Rules:
+- Extract answers for EVERY student visible
+- Answer format (use exactly one of these):
+  - A single uppercase letter (A, B, C, D, ...) — the letter the student chose
+  - "=" — the student answered correctly (source uses "-" / "✓" / "correct" / matches the answer key)
+  - null — the student left the question blank
+- Use the student's name or ID as shown in the document
+""" + NO_FABRICATION_RULE + "- Return valid JSON only",
+    "teacher_answers": """You are an exam data extraction expert. Extract the correct/model answers (answer key) from the provided content.
+
+This system supports SINGLE-CHOICE questions only.
+
+Return a JSON object with this structure:
+{
+  "answers": [
+    {
+      "question_number": 1,
+      "correct_answer": "A" | "B" | "C" | ...,
+      "explanation": "explanation if provided, otherwise null"
+    }
+  ]
+}
+
+Rules:
+- Extract the correct answer for EVERY question
+- correct_answer MUST be a single uppercase letter (A, B, C, D, ...)
+- Include explanations if they are provided
+""" + NO_FABRICATION_RULE + "- Return valid JSON only",
+}
+
+
+def _pdf_to_images(pdf_bytes: bytes) -> list[bytes]:
+    import fitz
+
+    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
+    images: list[bytes] = []
+    try:
+        for page in doc:
+            pix = page.get_pixmap(matrix=fitz.Matrix(2, 2))
+            images.append(pix.tobytes("png"))
+    finally:
+        doc.close()
+    return images
+
+
+class LLMVisionParser:
+    name = "llm_vision"
+    display_name = "LLM Vision (GPT)"
+    description = "Sends images/text to GPT. Works on any format; slower and uses API quota."
+
+    def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool:
+        return True
+
+    async def parse(
+        self,
+        files: list[ParserFile],
+        data_type: str,
+        description: str = "",
+        model: str = "gpt-5.4",
+    ) -> ParserResult:
+        if data_type not in EXTRACTION_PROMPTS:
+            raise ValueError(f"Unknown data_type: {data_type}")
+
+        image_parts: list[dict] = []
+        text_parts: list[str] = []
+
+        for f in files:
+            if not f.content:
+                raise ValueError(f"File '{f.filename}' is empty.")
+            if is_pdf(f.filename):
+                for page_png in _pdf_to_images(f.content):
+                    b64 = base64.b64encode(page_png).decode()
+                    image_parts.append(
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": f"data:image/png;base64,{b64}",
+                                "detail": "high",
+                            },
+                        }
+                    )
+                continue
+            text = try_read_as_text(f.content, f.filename)
+            if text is not None:
+                text_parts.append(f"--- Content from {f.filename} ---\n{text}")
+                continue
+            mime = get_image_mime(f.filename) or "image/png"
+            b64 = base64.b64encode(f.content).decode()
+            image_parts.append(
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:{mime};base64,{b64}",
+                        "detail": "high",
+                    },
+                }
+            )
+
+        if not image_parts and not text_parts:
+            raise ValueError("No valid content found in uploaded files")
+
+        data = await _call_llm(image_parts, text_parts, data_type, description, model)
+        return ParserResult(data=data, parser_name=self.name)
+
+
+async def _call_llm(
+    image_parts: list[dict],
+    text_parts: list[str],
+    data_type: str,
+    description: str,
+    model: str,
+) -> dict:
+    settings = get_settings()
+    client = AsyncOpenAI(api_key=settings.openai_api_key)
+
+    user_text = f"Please extract the {data_type.replace('_', ' ')} from the provided content."
+    if description.strip():
+        user_text = f"Context about this data: {description.strip()}\n\n{user_text}"
+    if text_parts:
+        user_text += "\n\n--- FILE CONTENT ---\n" + "\n\n".join(text_parts)
+
+    content_parts: list[dict] = [{"type": "text", "text": user_text}]
+    content_parts.extend(image_parts)
+
+    messages = [
+        {"role": "system", "content": EXTRACTION_PROMPTS[data_type]},
+        {"role": "user", "content": content_parts},
+    ]
+
+    kwargs: dict = {
+        "model": model,
+        "messages": messages,
+        "max_completion_tokens": 16384,
+    }
+
+    try:
+        response = await client.chat.completions.create(
+            **kwargs,
+            response_format={"type": "json_object"},
+        )
+    except Exception:
+        response = await client.chat.completions.create(**kwargs)
+
+    choice = response.choices[0]
+    content = choice.message.content
+    if not content:
+        reason = getattr(choice, "finish_reason", "unknown")
+        raise ValueError(f"Model returned empty content (finish_reason={reason}).")
+    return json.loads(content)
+
+
+# Optional helper retained for callers who only need the prompt text
+def extraction_prompt(data_type: str) -> Optional[str]:
+    return EXTRACTION_PROMPTS.get(data_type)

chatkit/backend/app/parsers/pdfplumber_tables.py

@@ -0,0 +1,55 @@
+"""Native PDF table parser using pdfplumber.
+
+Useful as a second-chance native parser when PyMuPDF misses a table."""
+
+from __future__ import annotations
+
+from .base import AnswerSheetParser, ParserFile, ParserResult, is_pdf, pdf_has_text_layer
+from ._table_common import table_to_data
+
+
+class PdfPlumberTablesParser:
+    name = "pdfplumber_tables"
+    display_name = "pdfplumber Tables (native PDF)"
+    description = (
+        "Alternative deterministic table extractor for native PDFs. "
+        "Try this if PyMuPDF returns wrong or partial tables."
+    )
+
+    def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool:
+        if data_type not in ("student_answers", "teacher_answers"):
+            return False
+        if not is_pdf(filename):
+            return False
+        return pdf_has_text_layer(file_bytes)
+
+    async def parse(
+        self,
+        files: list[ParserFile],
+        data_type: str,
+        description: str = "",
+        model: str = "gpt-5.4",
+    ) -> ParserResult:
+        import pdfplumber
+        import io
+
+        all_rows: list[list] = []
+        notes: list[str] = []
+        for f in files:
+            if not is_pdf(f.filename):
+                notes.append(f"skipped non-PDF: {f.filename}")
+                continue
+            with pdfplumber.open(io.BytesIO(f.content)) as pdf:
+                for page_idx, page in enumerate(pdf.pages):
+                    tables = page.extract_tables()
+                    if not tables:
+                        notes.append(f"{f.filename} page {page_idx + 1}: no tables found")
+                        continue
+                    for t in tables:
+                        all_rows.extend(t)
+
+        if not all_rows:
+            raise ValueError("pdfplumber found no tables in the uploaded PDF(s).")
+
+        data = table_to_data(all_rows, data_type)
+        return ParserResult(data=data, parser_name=self.name, notes=tuple(notes))

chatkit/backend/app/parsers/pymupdf_tables.py

@@ -0,0 +1,55 @@
+"""Native PDF table parser using PyMuPDF's page.find_tables()."""
+
+from __future__ import annotations
+
+from .base import AnswerSheetParser, ParserFile, ParserResult, is_pdf, pdf_has_text_layer
+from ._table_common import table_to_data
+
+
+class PyMuPDFTablesParser:
+    name = "pymupdf_tables"
+    display_name = "PyMuPDF Tables (native PDF)"
+    description = (
+        "Deterministic table extraction for native PDFs (e.g. 考生作答一覽表 exports). "
+        "Fast, no LLM call. Only for PDFs with a text layer."
+    )
+
+    def can_handle(self, file_bytes: bytes, filename: str, data_type: str) -> bool:
+        if data_type not in ("student_answers", "teacher_answers"):
+            return False
+        if not is_pdf(filename):
+            return False
+        return pdf_has_text_layer(file_bytes)
+
+    async def parse(
+        self,
+        files: list[ParserFile],
+        data_type: str,
+        description: str = "",
+        model: str = "gpt-5.4",
+    ) -> ParserResult:
+        import fitz
+
+        all_rows: list[list] = []
+        notes: list[str] = []
+        for f in files:
+            if not is_pdf(f.filename):
+                notes.append(f"skipped non-PDF: {f.filename}")
+                continue
+            doc = fitz.open(stream=f.content, filetype="pdf")
+            try:
+                for page_idx, page in enumerate(doc):
+                    tables = list(page.find_tables())
+                    if not tables:
+                        notes.append(f"{f.filename} page {page_idx + 1}: no tables found")
+                        continue
+                    for t in tables:
+                        all_rows.extend(t.extract())
+            finally:
+                doc.close()
+
+        if not all_rows:
+            raise ValueError("PyMuPDF found no tables in the uploaded PDF(s).")
+
+        data = table_to_data(all_rows, data_type)
+        return ParserResult(data=data, parser_name=self.name, notes=tuple(notes))

chatkit/backend/pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "pydantic>=2.0.0",
     "httpx>=0.25.0",
     "pymupdf>=1.23.0",
+    "pdfplumber>=0.11.0",
 ]
 
 [project.optional-dependencies]

chatkit/backend/tests/test_native_parsers.py

@@ -0,0 +1,76 @@
+"""Integration tests for native PDF table parsers against sample data."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from app.parsers import get_parser, pick_auto
+from app.parsers.base import ParserFile
+
+SAMPLE_PDF = Path(
+    "/Users/yu.c/Desktop/npo-ca-classlens-data copy/example-2/9A作答一覽表.pdf"
+)
+
+
+@pytest.fixture
+def sample_file() -> ParserFile:
+    if not SAMPLE_PDF.exists():
+        pytest.skip(f"Sample PDF not available: {SAMPLE_PDF}")
+    return ParserFile(filename=SAMPLE_PDF.name, content=SAMPLE_PDF.read_bytes())
+
+
+@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")
+    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"
+    first_10 = [a["answer"] for a in by_name["陳筱琳"]["answers"][:10]]
+    # From the PDF: "==CBBBA==C"
+    assert first_10 == ["=", "=", "C", "B", "B", "B", "A", "=", "=", "C"]
+
+
+@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")
+    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"
+
+
+@pytest.mark.integration
+async def test_auto_picks_native_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")
+
+
+@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

chatkit/frontend/src/components/step1/FileUploadPanel.tsx

@@ -1,7 +1,13 @@
-import { useState, useRef } from "react";
-import { apiUpload } from "../../lib/api";
+import { useEffect, useRef, useState } from "react";
+import { apiGet, apiUpload } from "../../lib/api";
 import { ModelSelector } from "../ModelSelector";
 
+interface ParserOption {
+  id: string;
+  name: string;
+  description: string;
+}
+
 interface FileUploadPanelProps {
   sessionId: number;
   parsedData: Record<string, unknown>;
@@ -60,8 +66,20 @@ export function FileUploadPanel({
   const [analyzing, setAnalyzing] = useState<Record<string, boolean>>({});
   const [errors, setErrors] = useState<Record<string, string>>({});
   const [model, setModel] = useState("gpt-5.4");
+  const [parsers, setParsers] = useState<ParserOption[]>([]);
+  const [parserByZone, setParserByZone] = useState<Record<string, string>>({
+    questions: "auto",
+    student_answers: "auto",
+    teacher_answers: "auto",
+  });
   const fileInputRefs = useRef<Record<string, HTMLInputElement | null>>({});
 
+  useEffect(() => {
+    apiGet<{ parsers: ParserOption[] }>("/api/parsers")
+      .then((res) => setParsers(res.parsers))
+      .catch(() => {});
+  }, []);
+
   const handleFileSelect = (type: string, selected: FileList | null) => {
     if (!selected) return;
     setFiles((prev) => ({ ...prev, [type]: Array.from(selected) }));
@@ -85,6 +103,7 @@ export function FileUploadPanel({
       formData.append("data_type", type);
       formData.append("description", descriptions[type] || "");
       formData.append("model", model);
+      formData.append("parser", parserByZone[type] || "auto");
       for (const f of typeFiles) {
         formData.append("files", f);
       }
@@ -149,6 +168,31 @@ export function FileUploadPanel({
                 rows={2}
               />
 
+              {parsers.length > 0 && (
+                <div className="mb-3">
+                  <label className="block text-xs text-[var(--color-text-muted)] mb-1">
+                    解析工具
+                  </label>
+                  <select
+                    value={parserByZone[zone.type] || "auto"}
+                    onChange={(e) =>
+                      setParserByZone((prev) => ({ ...prev, [zone.type]: e.target.value }))
+                    }
+                    className="input !py-1.5 text-xs w-full"
+                    title={
+                      parsers.find((p) => p.id === (parserByZone[zone.type] || "auto"))
+                        ?.description ?? ""
+                    }
+                  >
+                    {parsers.map((p) => (
+                      <option key={p.id} value={p.id}>
+                        {p.name}
+                      </option>
+                    ))}
+                  </select>
+                </div>
+              )}
+
               <input
                 ref={(el) => { fileInputRefs.current[zone.type] = el; }}
                 type="file"
@@ -201,7 +245,17 @@ export function FileUploadPanel({
               {/* Parsed status + summary */}
               {hasParsed && (
                 <div className="mt-3 p-3 bg-[var(--color-background)] rounded-lg">
-                  <p className="text-xs text-[var(--color-success)] font-semibold mb-1">✓ 解析完成</p>
+                  <p className="text-xs text-[var(--color-success)] font-semibold mb-1">
+                    ✓ 解析完成
+                    {(() => {
+                      const meta = (zoneData as { _meta?: { parser?: string } })._meta;
+                      return meta?.parser ? (
+                        <span className="ml-2 text-[var(--color-text-muted)] font-normal">
+                          (by {meta.parser})
+                        </span>
+                      ) : null;
+                    })()}
+                  </p>
                   <ZoneSummary type={zone.type} data={zoneData} />
                 </div>
               )}

chatkit/frontend/src/components/step2/DataPreviewModal.tsx

@@ -52,7 +52,7 @@ export function DataPreviewModal({
         onClick={onClose}
       />
 
-      <div className="relative w-full max-w-4xl max-h-[90vh] flex flex-col card overflow-hidden">
+      <div className="relative w-full max-w-6xl h-[92vh] flex flex-col card overflow-hidden">
         <div className="flex items-center justify-between p-5 border-b border-[var(--color-border)]">
           <h3 className="font-display text-xl font-bold text-[var(--color-text)]">
             🔍 最終提示詞預覽
@@ -116,7 +116,7 @@ export function DataPreviewModal({
             <textarea
               value={preview?.markdown_prompt ?? ""}
               readOnly
-              className="input !rounded-xl font-mono text-sm leading-relaxed resize-y flex-1"
+              className="input !rounded-xl font-mono text-sm leading-relaxed resize-none flex-1 min-h-0"
               spellCheck={false}
             />
           )}