"""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 logging import uuid from pathlib import Path from fastapi import UploadFile from .answer_grid import seed_from_parsed, to_dict as grid_to_dict from .config import get_settings from .database import ( delete_parsed_data, get_parsed_data, save_answer_grid, save_parsed_data, save_raw_file, ) from .parsers import AUTO, get_parser, pick_auto from .parsers.base import ParserFile, ParserResult logger = logging.getLogger(__name__) # Original uploaded scans/PDFs, kept so a teacher/admin can view the source # file later — separate from ParsedData, which only holds the OCR-extracted # structured fields. _UPLOAD_DIR = Path(__file__).parent.parent / "uploads" async def _save_raw_file( session_id: int, data_type: str, filename: str, content_type: str, content: bytes ) -> None: settings = get_settings() base_dir = Path(settings.upload_dir) if settings.upload_dir else _UPLOAD_DIR quiz_dir = base_dir / str(session_id) quiz_dir.mkdir(parents=True, exist_ok=True) ext = Path(filename).suffix stored_name = f"{uuid.uuid4().hex}{ext}" dest = quiz_dir / stored_name dest.write_bytes(content) await save_raw_file( quiz_id=session_id, data_type=data_type, file_name=filename, content_type=content_type, size_bytes=len(content), storage_path=str(dest), ) async def process_uploaded_files( files: list[UploadFile], data_type: str, session_id: int, description: str = "", model: str = "gpt-5.4", parser: str = AUTO, ) -> dict: """Parse uploaded files and persist the structured result. `data_type='answers'` triggers a dual extraction: the same files are parsed twice (once for student_answers, once for teacher_answers) and both rows are stored. The response merges both shapes under their respective keys. """ parser_files: list[ParserFile] = [] for file in files: 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) ) await _save_raw_file( session_id, data_type, file.filename or "unknown", file.content_type or "", content, ) if not parser_files: raise ValueError("No files uploaded") if data_type == "answers": return await _process_combined_answers( parser_files, session_id, description, model, parser ) result = await _run_with_fallback( parser_files, data_type, description, model, parser ) _validate_extraction(data_type, result.data) await _persist(session_id, data_type, parser_files, result) return _response_payload(result) def _validate_extraction(data_type: str, data: dict) -> None: """Reject an extraction that "succeeded" but found nothing — the model is instructed to return an empty list rather than fabricate data when it can't read the content (see NO_FABRICATION_RULE), so a bad scan or wrong file doesn't raise on its own; without this check it silently produces an unusable session that only surfaces later as an unexplained dead end in step 2. Only `student_answers` is checked (never legitimately empty) — `questions`/`teacher_answers` can legitimately be uploaded separately or filled in later via the answer grid editor, so an empty result there is not necessarily a failure. """ if data_type == "student_answers" and not data.get("students"): raise ValueError( "No students could be extracted from this file. The scan may be " "blurry, blank, or not an answer sheet — please check the file " "and try again." ) async def _process_combined_answers( parser_files: list[ParserFile], session_id: int, description: str, model: str, parser: str, ) -> dict: student_result = await _run_with_fallback( parser_files, "student_answers", description, model, parser ) _validate_extraction("student_answers", student_result.data) teacher_result = await _run_with_fallback( parser_files, "teacher_answers", description, model, parser ) await _persist(session_id, "student_answers", parser_files, student_result) await _persist(session_id, "teacher_answers", parser_files, teacher_result) # Seed and save a draft answer grid so step 2 has all data (students + correct answers). # Use only the answers extracted from teacher_result to avoid inheriting the question count # from a separately uploaded questions PDF (which may have a different total). t_data = teacher_result.data draft_grid = seed_from_parsed({}, student_result.data, t_data) await save_answer_grid(session_id, grid_to_dict(draft_grid)) return { "student_answers": _response_payload(student_result), "teacher_answers": _response_payload(teacher_result), } async def _persist( session_id: int, data_type: str, parser_files: list[ParserFile], result: ParserResult, ) -> None: if data_type == "questions": await delete_parsed_data(session_id, data_type) for q in result.data.get("questions", []): text = (q.get("text", "") or "").strip() options = q.get("options") or [] if not text and not options: logger.warning( "session %d: question %s extracted with no text and no " "options — likely a failed extraction, not a legitimate " "picture-only question", session_id, q.get("number", "?"), ) await save_parsed_data( session_id, q.get("number", 0), text, "", "", [], options=options, ) elif data_type == "teacher_answers": existing = {r["question_num"]: r for r in await get_parsed_data(session_id)} answers_map = { a.get("question_number", 0): ( a.get("correct_answer") or a.get("answer") or "" ) for a in result.data.get("answers", []) } all_nums = sorted(set(existing) | set(answers_map)) await delete_parsed_data(session_id, data_type) for qnum in all_nums: row = existing.get(qnum, {}) await save_parsed_data( session_id, qnum, row.get("question_str", ""), answers_map.get(qnum, row.get("answer", "")), row.get("main_category", ""), row.get("tags", []), options=row.get("options", []), ) # student_answers are not stored in ParsedData def _response_payload(result: ParserResult) -> dict: return { **result.data, "_meta": { "parser": result.parser_name, "notes": list(result.notes), }, } async def _run_with_fallback( parser_files: list[ParserFile], data_type: str, 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)