diff --git a/.env.example b/.env.example index a83056e8..b6b8ca9b 100644 --- a/.env.example +++ b/.env.example @@ -1,2 +1,22 @@ -# Copy this file to .env and paste your real key after the equals sign. +OPENROUTER_API_KEY= GEMINI_API_KEY= +LLM_PROVIDER=openrouter +OPENROUTER_MODEL=google/gemini-3.5-flash +GEMINI_MODEL=gemini-2.0-flash +PORT=8501 +IS_QUESTIONS_NEED_CSV=Yes +BIHAR_WORKERS=2 +BIHAR_RENDER_DPI=200 +BIHAR_SOURCE_SPREAD_COUNT=6 +BIHAR_LOGICAL_PAGE_COUNT=12 +BIHAR_REFERENCE_MASTER_PDF=../reference_master.pdf +BIHAR_SPREAD_GUTTER_FRACTION=0.018 +BIHAR_MIN_IMAGE_QUALITY=0.35 +BIHAR_REVIEW_CONFIDENCE_THRESHOLD=0.72 +BIHAR_MAX_RETRIES=3 +BIHAR_RETRY_BACKOFF_SECONDS=1.5 +BIHAR_REQUEST_TIMEOUT_SECONDS=120 +BIHAR_MAX_COMPLETION_TOKENS=20000 +BIHAR_OUTPUT_WORKBOOK=questionnaire_results.xlsx +BIHAR_QUESTION_CATALOG=questions.txt +BIHAR_STATE_FILE=processing_state.json diff --git a/.gitignore b/.gitignore index 484606ef..66d6cacb 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,6 @@ __pycache__/ .env output/ .DS_Store +results/ +env/ +test_runs/ diff --git a/README.md b/README.md index 8151d9be..e94df9f4 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,15 @@ master before generating answer crops. ```powershell python -m venv .venv; .\.venv\Scripts\Activate.ps1 pip install -r requirements.txt -$env:GEMINI_API_KEY = "..." # optional; deterministic local extraction still runs +$env:OPENROUTER_API_KEY = "..." +$env:OPENROUTER_MODEL = "google/gemini-3.5-flash" +$env:GEMINI_API_KEY = "..." +$env:GEMINI_MODEL = "gemini-2.0-flash" +$env:LLM_PROVIDER = "openrouter" # or "gemini" +$env:PORT = "8501" python main.py process --input C:\scans --output C:\results +or +python3 main.py process --input input_pdfs --output results streamlit run core/review/streamlit_review.py -- --run-dir C:\results ``` @@ -25,3 +32,19 @@ The calibrated answer regions are saved once in `config/pages.json`; they are not edited for individual student PDFs. The system deliberately supports only this format. It records intermediate crops and JSON audit records in the output directory and can resume completed PDFs. + +All application configuration is loaded from environment variables. Copy +`.env.example` to `.env` and adjust values there; `.env` is excluded from Git. +Set `IS_QUESTIONS_NEED_CSV=Yes` to add `Q1 — ` headers to the +response report; set it to `No` to use compact `Q1`–`Q34` headers only. +Set `BIHAR_REQUEST_TIMEOUT_SECONDS` to bound each LiteLLM request, including +each attempt made by the retry policy. +Set `LLM_PROVIDER=gemini` with `GEMINI_API_KEY` to use direct Gemini, or leave +it as `openrouter` with `OPENROUTER_API_KEY` to use OpenRouter. +The Gemini LiteLLM route uses LiteLLM's chat-completion interface; LiteLLM +model names are qualified automatically (for example `gemini/` when +using the LiteLLM Gemini route). The OpenRouter route sends the value of +`OPENROUTER_MODEL` directly to OpenRouter — do not prefix this value with +`openrouter/` in your `.env` file, since that exact string will be used. +At startup, the application validates all `BIHAR_*` settings, `PORT`, the +selected model, and the credentials required by the selected provider. diff --git a/config/prompts.yaml b/config/prompts.yaml index 8252c4d7..fe721859 100644 --- a/config/prompts.yaml +++ b/config/prompts.yaml @@ -1,14 +1,74 @@ system: | You are the answer-extraction engine for one fixed Hindi questionnaire. - Each image is preceded by its fixed question ID, type, and allowed options. - Never infer question IDs, never return question text, and never invent an answer. - For a single_choice question, return exactly one allowed code only when one checkbox is clearly marked. - If zero or multiple boxes are marked, or the mark is ambiguous, return an empty selected_codes array and set review_required true. - For multiple_choice, return every clearly marked allowed code; uncertainty requires review_required true. - For short_text and long_text, transcribe only legible handwritten Hindi/Devanagari exactly as written. - Do not translate Hindi to English. Preserve names, numbers, and punctuation. If handwriting cannot be read, - return answer_text as an empty string and set review_required true. + + The first page contains student information. + The remaining pages contain questionnaire responses. + + Extract both the student information and the questionnaire answers. + + Never invent values. + Never guess unreadable handwriting. + If a field is blank or unreadable, return an empty string (""). + + For questionnaire answers: + + - Never infer question IDs. + - Never return question text. + - Never invent an answer. + - For a single_choice question, return exactly one allowed code only when one checkbox is clearly marked. + - If zero or multiple boxes are marked, or the mark is ambiguous, return an empty selected_codes array and set review_required to true. + - For multiple_choice, return every clearly marked allowed code. If uncertain, set review_required to true. + - For short_text and long_text, transcribe only legible handwritten Hindi/Devanagari exactly as written. + - Do not translate Hindi to English. + - Preserve names, numbers and punctuation exactly. + - If handwriting cannot be read, return answer_text as an empty string and set review_required to true. + - Every answer object must always include all six fields shown in the JSON + template below, with exactly those names, even when a field does not + apply to that question. For example, single_choice and multiple_choice + questions have no free text: still return answer_text as "" rather than + omitting the key. + instruction: | - Return a JSON array only. Each item must contain exactly question_id, selected_codes, - answer_text, confidence, review_required, raw_observations. selected_codes is an array. - Use only allowed option codes. confidence must be a number from 0 to 1. + Return ONLY one valid JSON object. + + The JSON object must have exactly two top-level keys: + + { + "student": { + "student_name": "", + "gender": "", + "school_name": "", + "school_udise": "", + "crc_name": "", + "crc_udise": "", + "block": "", + "district": "", + "grade": "", + "meena_manch_participation": "" + }, + "answers": [ + { + "question_id": "", + "selected_codes": [], + "answer_text": "", + "confidence": 0.0, + "review_required": false, + "raw_observations": "" + } + ] + } + + Rules: + + - Return ONLY valid JSON. + - Do not include markdown. + - Do not include explanations. + - student contains information extracted only from the first page. + - If any student field is blank or unreadable, return an empty string. + - answers must contain exactly one object for every detected questionnaire answer. + - Every answer object must contain all six keys from the template above + (question_id, selected_codes, answer_text, confidence, review_required, + raw_observations), never fewer. Use "" or [] as the empty value, never omit the key. + - selected_codes must always be an array. + - Use only the allowed option codes. + - confidence must be a number between 0 and 1. \ No newline at end of file diff --git a/config/questions.txt b/config/questions.txt new file mode 100644 index 00000000..a6572284 --- /dev/null +++ b/config/questions.txt @@ -0,0 +1,109 @@ +1. +अनीता के विद्यालय का पानी का नल अक्सर टूट जाता है। समस्या हल करने का पहला कदम क्या होगा? + +2. +किसी समस्या के दो समाधान चुने गए हैं। उन्हें चुनने से पहले क्या करना चाहिए? + +3. +यदि दो लोगों की जानकारी अलग है, तो क्या करना चाहिए? + +4. +यदि कोई योजना काम नहीं करती उसे बेहतर तरीके से सोचने का क्या तरीका है? + +5. +यदि किसी नई बीमारी की अफवाह फैलती है तो हमें सबसे पहले क्या करना चाहिए? + +6. +आपके गाँव में लड़कियों को किन समस्याओं का सामना करना पड़ता है, और ऐसा क्यों होता है? +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +7. +अच्छा संचार का मतलब है: + +8. +यदि कोई आपको नहीं समझता, तो आपको क्या करना चाहिए? + +9. +समूह में विचार साझा करते समय सबसे मददगार क्या है? + +10. +मीना मंच बैठक में लड़कियाँ विद्यालय की सफाई पर चर्चा कर रही हैं, लेकिन कुछ लड़कियाँ बोलने में संकोच करती हैं। चर्चा को प्रोत्साहित करने का सबसे अच्छा तरीका क्या है? + +11. +हम यह कैसे सुनिश्चित कर सकते हैं कि हमारी स्कूल में लड़कियाँ विद्यालय में सुरक्षित रहें और अच्छी तरह से पढ़ाई कर सकें? +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +12. +आपका मीना समूह चाहता है कि लड़कियाँ नियमित रूप से विद्यालय जाएँ। इसके लिए सबसे अच्छा तरीका क्या होगा? + +13. +यदि समूह (group) के दो सदस्यों के बीच असहमति हो, तो सबसे अच्छा क्या है? + +14. +समूह में छात्र छात्राएँ अलग-अलग भाषाएँ बोलते हैं। टीमवर्क में सबसे अधिक मददगार क्या है? + +15. +समूह कार्य में आपकी भूमिका स्पष्ट नहीं है। इसके लिए आपको क्या करना चाहिए? + +16. +आप और आपके दोस्त विद्यालय की किसी समस्या को हल करने में विद्यालय की कैसे मदद कर सकते हैं? +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +17. +रचनात्मकता का मतलब है: + +18. +आपका मीना समूह चाहता है कि लड़कियाँ नियमित रूप से विद्यालय जाएँ। इसके लिए कौन-सा विचार सबसे अधिक रचनात्मक है? + +19. +गाँव की लड़कियाँ विद्यालय जाते समय असुरक्षित महसूस करती हैं। सबसे रचनात्मक समाधान क्या होगा? + +20. +एक नया विचार लिखें जिससे आप अपने विद्यालय या गाँव को बेहतर बना सकते हैं। +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +21. +अच्छा नेता कौन होता है? + +22. +समूह का नेतृत्व करते समय क्या महत्वपूर्ण है? + +23. +यदि आपकी टीम भ्रमित है, तो नेता को क्या करना चाहिए? + +24. +मीना मंच बैठक में केवल दो लड़कियाँ बोल रही हैं और बाकी चुप हैं। एक अच्छा नेता क्या करेगा? + +25. +एक ऐसी स्थिति का वर्णन करें जिसमें आपने जिम्मेदारी ली हो या दूसरों का नेतृत्व करने में मदद की हो। +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +26. +कुछ छात्र एक लड़की के उच्चारण पर हँसते हैं। आपको क्या करना चाहिए? + +27. +यदि आप सम्मानपूर्वक असहमत होते हैं, तो आप क्या दिखा रहे हैं? + +28. +चुनौती का सामना करते समय सशक्त लड़कियों को क्या करना चाहिए? + +29. +आपके पड़ोस की एक लड़की विद्यालय आना बंद कर देती है। सबसे पहले आप क्या कर सकते हैं? + +30. +आपके सपने क्या हैं? बड़े होकर आप क्या बनना चाहते हैं? +खुला प्रश्न — लिखित उत्तर के लिए स्थान दिया गया है। + +31. +पिछले सप्ताह में आप कितने दिन स्कूल गए? + +32. +पिछले एक महीने में आप कितनी बार स्कूल से अनुपस्थित रहे हैं? + +33. +स्कूल न जाने का आपका मुख्य कारण क्या है? (एक या अधिक विकल्प चुनें) + +34. +अगर आप स्कूल नहीं जाते हैं, तो आप आमतौर पर क्या करते हैं? + + diff --git a/config/settings.yaml b/config/settings.yaml deleted file mode 100644 index 722d929c..00000000 --- a/config/settings.yaml +++ /dev/null @@ -1,13 +0,0 @@ -workers: 2 -render_dpi: 200 -source_spread_count: 6 -logical_page_count: 12 -reference_master_pdf: ../../reference_master.pdf -spread_gutter_fraction: 0.018 -min_image_quality: 0.35 -review_confidence_threshold: 0.72 -gemini_model: gemini-3.5-flash -max_retries: 3 -retry_backoff_seconds: 1.5 -output_workbook: questionnaire_results.xlsx -state_file: processing_state.json diff --git a/core/ai/confidence.py b/core/ai/confidence.py index 49ba4f23..1d1bba02 100644 --- a/core/ai/confidence.py +++ b/core/ai/confidence.py @@ -1,5 +1,21 @@ -"""Fuse independent quality indicators into a reviewable confidence score.""" +# """Fuse independent quality indicators into a reviewable confidence score.""" +# class ConfidenceEngine: +# def calculate(self, llm: float, checkbox: float, image: float, handwriting: float, rules_ok: bool) -> float: +# visual=max(checkbox, handwriting) +# return round(max(0.,min(1., .45*llm+.25*visual+.20*image+.10*(1. if rules_ok else 0.))),3) + class ConfidenceEngine: - def calculate(self, gemini: float, checkbox: float, image: float, handwriting: float, rules_ok: bool) -> float: - visual=max(checkbox, handwriting) - return round(max(0.,min(1., .45*gemini+.25*visual+.20*image+.10*(1. if rules_ok else 0.))),3) + def calculate( + self, + llm: float, + image: float, + rules_ok: bool, + ) -> float: + + score = ( + 0.70 * llm + + 0.20 * image + + 0.10 * (1.0 if rules_ok else 0.0) + ) + + return round(max(0.0, min(1.0, score)), 3) \ No newline at end of file diff --git a/core/ai/gemini_provider.py b/core/ai/gemini_provider.py deleted file mode 100644 index 7d786927..00000000 --- a/core/ai/gemini_provider.py +++ /dev/null @@ -1,56 +0,0 @@ -"""Google GenAI provider for checkbox interpretation and Hindi handwriting.""" -from pathlib import Path -import os -import json -import time -import traceback -from dotenv import load_dotenv -from google.genai import types -from core.ai.retry import Retry -from core.ai.response_parser import ResponseParser -from core.models import Question, Answer -class GeminiProvider: - def __init__(self, model: str, parser: ResponseParser, retry: Retry): self.model,self.parser,self.retry=model,parser,retry - def extract(self, prompt: str, crops: dict[str, Path], questions: list[Question], debug_dir: Path | None = None) -> list[Answer]: - # Supports a local .env for VS Code runs while retaining normal environment - # variable deployment. The .env file is excluded from version control. - load_dotenv() - key=os.getenv("GEMINI_API_KEY") - if not key: - return [Answer(q.id, confidence=0, review_required=True, raw_observations="Gemini was not called: GEMINI_API_KEY is unavailable") for q in questions] - for question in questions: - crop = crops.get(question.id) - if crop is None or not crop.exists() or crop.stat().st_size == 0: - raise FileNotFoundError(f"Missing or empty crop for {question.id}: {crop}") - if debug_dir: - debug_dir.mkdir(parents=True, exist_ok=True) - (debug_dir / "request.json").write_text(json.dumps({"model": self.model, "question_ids": [q.id for q in questions], "crop_paths": {q.id: str(crops[q.id]) for q in questions}, "prompt_length": len(prompt)}, indent=2), encoding="utf-8") - from google import genai - client=genai.Client(api_key=key) - parts = [types.Part.from_text(text=prompt)] - parts.extend( - types.Part.from_bytes(data=crops[question.id].read_bytes(), mime_type="image/png") - for question in questions - ) - contents = [types.Content(role="user", parts=parts)] - started = time.monotonic() - try: - response=self.retry.run(lambda: client.models.generate_content( - model=self.model, - contents=contents, - config=types.GenerateContentConfig( - response_mime_type="application/json", - temperature=0, - ), - )) - if debug_dir: - (debug_dir / "raw_response.txt").write_text(response.text or "", encoding="utf-8") - (debug_dir / "response_metadata.json").write_text(json.dumps({"latency_seconds": round(time.monotonic() - started, 3), "finish_reason": str(getattr(response, "finish_reason", None)), "usage": str(getattr(response, "usage_metadata", None))}, indent=2), encoding="utf-8") - answers = self.parser.parse(response.text) - if debug_dir: - (debug_dir / "parsed.json").write_text(json.dumps([answer.to_dict() for answer in answers], ensure_ascii=False, indent=2), encoding="utf-8") - return answers - except Exception: - if debug_dir: - (debug_dir / "exception.txt").write_text(traceback.format_exc(), encoding="utf-8") - raise diff --git a/core/ai/litellm_provider.py b/core/ai/litellm_provider.py new file mode 100644 index 00000000..aa5ea195 --- /dev/null +++ b/core/ai/litellm_provider.py @@ -0,0 +1,132 @@ +"""Common LiteLLM vision provider for OpenRouter and direct Gemini.""" +from __future__ import annotations + +import base64 +import json +import time +import traceback +from pathlib import Path + +from dotenv import load_dotenv + +from core.ai.response_parser import ResponseParser +from core.ai.retry import Retry +from core.models import Answer, Question, StudentInfo +from core.logging.logger import get_logger + + +class LiteLLMProvider: + """Extract questionnaire responses through LiteLLM's common interface.""" + + def __init__(self, provider: str, model: str, parser: ResponseParser, retry: Retry, + request_timeout_seconds: float, max_completion_tokens: int): + self.provider = provider.lower() + self.model = self._qualified_model(self.provider, model) + self.parser, self.retry = parser, retry + self.request_timeout_seconds = request_timeout_seconds + self.max_completion_tokens = max_completion_tokens + self.logger = get_logger("ai.litellm") + + @staticmethod + def _qualified_model(provider: str, model: str) -> str: + if provider == "openrouter" and not model.startswith("openrouter/"): + return f"openrouter/{model}" + if provider in {"gemini", "direct_gemini", "google"} and not model.startswith("gemini/"): + return f"gemini/{model}" + return model + + def extract(self, prompt: str, crops: dict[str | int, Path], questions: list[Question], + debug_dir: Path | None = None) -> tuple[StudentInfo, list[Answer]]: + load_dotenv() + import litellm + + for image in crops.values(): + if image is None or not image.exists() or image.stat().st_size == 0: + raise FileNotFoundError(f"Missing image: {image}") + if debug_dir: + debug_dir.mkdir(parents=True, exist_ok=True) + (debug_dir / "request.json").write_text(json.dumps({ + "provider": self.provider, "model": self.model, + "image_count": len(crops), "image_paths": {str(k): str(v) for k, v in crops.items()}, + "prompt_length": len(prompt), "timeout_seconds": self.request_timeout_seconds, + }, indent=2), encoding="utf-8") + + content = [{"type": "text", "text": prompt}] + if "STUDENT_INFO" in crops: + student_image = crops["STUDENT_INFO"] + content.extend([ + {"type": "text", "text": "Student information page. Extract only the student object from this image."}, + {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64.b64encode(student_image.read_bytes()).decode('ascii')}"}}, + ]) + + # Append question metadata blocks (so the model knows question ids, types and page numbers) + for question in questions: + content.append({ + "type": "text", + "text": ( + f"Question metadata: question_id={question.id}; type={question.type}; page={question.page}; " + f"allowed_codes={json.dumps(question.options)}." + ), + }) + + # Then append the page images (one image per logical page) + page_items = sorted(((k, v) for k, v in crops.items() if k != "STUDENT_INFO"), key=lambda kv: int(kv[0]) if isinstance(kv[0], int) or str(kv[0]).isdigit() else str(kv[0])) + for _, path in page_items: + content.append({ + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{base64.b64encode(path.read_bytes()).decode('ascii')}"}, + }) + started = time.monotonic() + try: + self.logger.info("Calling %s with %d images (timeout=%.1fs, max_completion_tokens=%d)", + self.model, len(crops), self.request_timeout_seconds, self.max_completion_tokens) + # A `reasoning` field (both `max_tokens` and `effort` shapes were + # tried, via `extra_body` since litellm's own reasoning kwargs + # don't yet recognise this model) was intended to cap Gemini's + # internal "thinking" pass, which can otherwise consume nearly + # the entire output-token ceiling and truncate the visible JSON + # answer. In practice, adding *any* `reasoning` field to the + # request caused a chunk of calls to abort almost immediately + # (finish_reason "error", 0 tokens billed) on this very new + # model/route — worse than the truncation it was meant to fix. + # Reverted. + # + # `max_tokens` is a plain, universally-supported completion cap + # (not gated behind litellm's reasoning-capability allowlist the + # way `reasoning`/`thinking` were), so it doesn't hit the same + # instability. It can't prevent a reasoning-runaway response from + # still being truncated -- that failure is still absorbed by the + # per-PDF handling in batch_processor -- but it bounds how much + # of the ~65k-token ceiling (and therefore cost) a single runaway + # call can burn through before failing: observed runaway calls + # were billed for ~65k completion tokens (~$0.62); capping at + # max_completion_tokens caps the worst case at a fraction of that, + # while staying well above what a normal successful call needs + # (~3.6k reasoning + ~2.5k answer tokens observed on average). + response = self.retry.run(lambda: litellm.completion( + model=self.model, + messages=[{"role": "user", "content": content}], + temperature=0, + timeout=self.request_timeout_seconds, + max_tokens=self.max_completion_tokens, + )) + text = response.choices[0].message.content or "" + if debug_dir: + (debug_dir / "raw_response.txt").write_text(text, encoding="utf-8") + usage = getattr(response, "usage", None) + (debug_dir / "response_metadata.json").write_text(json.dumps({ + "latency_seconds": round(time.monotonic() - started, 3), + "usage": getattr(usage, "model_dump", lambda: vars(usage) if usage else {})(), + }, indent=2, default=str), encoding="utf-8") + student, answers = self.parser.parse(text, questions) + self.logger.info("%s response parsed in %.2fs", self.model, time.monotonic() - started) + if debug_dir: + (debug_dir / "parsed.json").write_text(json.dumps({ + "student": student.to_dict(), "answers": [a.to_dict() for a in answers], + }, ensure_ascii=False, indent=2), encoding="utf-8") + return student, answers + except Exception: + self.logger.exception("%s request failed", self.model) + if debug_dir: + (debug_dir / "exception.txt").write_text(traceback.format_exc(), encoding="utf-8") + raise diff --git a/core/ai/response_parser.py b/core/ai/response_parser.py index a5df3284..f4f887fe 100644 --- a/core/ai/response_parser.py +++ b/core/ai/response_parser.py @@ -1,24 +1,182 @@ -"""Validate Gemini's answer contract while tolerating harmless extra fields.""" +"""Validate the LLM answer contract while tolerating harmless extra fields.""" + import json import re import warnings -from core.models import Answer + +from core.models import Answer, Question, StudentInfo +from core.logging.logger import get_logger + +logger = get_logger("ai.response_parser") + + class ResponseParser: - REQUIRED={"question_id","selected_codes","answer_text","confidence","review_required","raw_observations"} - def parse(self, text: str) -> list[Answer]: - match=re.search(r"\[[\s\S]*\]", text) - if not match: raise ValueError("AI response did not contain a JSON array") - payload=json.loads(match.group()) - if not isinstance(payload,list): raise ValueError("AI response must be an array") - answers=[] - for item in payload: + # question_id is the only field an answer object cannot be recovered + # without (there is no question to attach it to otherwise). The model + # (observed with Gemini 3.5 Flash via OpenRouter) intermittently omits + # answer_text for multiple_choice/single_choice questions since there is + # no free text to report; failing the whole 34-answer response over one + # absent, inapplicable field wastes an already-paid-for LLM call. Missing + # optional fields are defaulted instead, and the affected answer is + # forced into review so a person still checks it. + REQUIRED = {"question_id"} + OPTIONAL_DEFAULTS = { + "selected_codes": [], + "answer_text": "", + "confidence": 0.0, + "review_required": False, + "raw_observations": "", + } + + STUDENT_FIELDS = { + "student_name", + "gender", + "school_name", + "school_udise", + "crc_name", + "crc_udise", + "block", + "district", + "grade", + "meena_manch_participation", + } + + def parse(self, text: str, questions: list[Question] | None = None) -> tuple[StudentInfo, list[Answer]]: + + # Accept either a JSON object with {student:..., answers:[...]} or a + # bare JSON array containing just the answers. When the latter is + # received treat the student as empty/default. + stripped = text.strip() + # If the response starts with an array, treat it as a bare answers list. + if stripped.startswith("["): + match_arr = re.search(r"\[[\s\S]*\]", text) + if not match_arr: + raise ValueError("AI response did not contain a JSON array") + answers_data = json.loads(match_arr.group()) + student_data = {} + else: + # Otherwise look for an object payload containing student + answers + match_obj = re.search(r"\{[\s\S]*\}", text) + if not match_obj: + raise ValueError("AI response did not contain a JSON object or array") + payload = json.loads(match_obj.group()) + if not isinstance(payload, dict): + raise ValueError("AI response must be a JSON object when a student object is present") + if "student" not in payload: + raise ValueError("Missing 'student' object") + if "answers" not in payload: + raise ValueError("Missing 'answers' array") + student_data = payload["student"] + answers_data = payload["answers"] + + if not isinstance(answers_data, list): + raise ValueError("'answers' must be an array") + + # ---------------- Student ---------------- + # When the AI returned a bare answers array there is no student + # object to validate; use an empty/default StudentInfo in that case. + if not student_data: + student = StudentInfo() + else: + if not isinstance(student_data, dict): + raise ValueError("'student' must be an object") + + missing = self.STUDENT_FIELDS - set(student_data) + + if missing: + raise ValueError( + f"Student object missing fields: {sorted(missing)}" + ) + + extras = set(student_data) - self.STUDENT_FIELDS + + if extras: + warnings.warn( + f"Ignoring unexpected student fields: {sorted(extras)}", + RuntimeWarning, + ) + + student = StudentInfo( + student_name=str(student_data["student_name"]), + gender=str(student_data["gender"]), + school_name=str(student_data["school_name"]), + school_udise=str(student_data["school_udise"]), + crc_name=str(student_data["crc_name"]), + crc_udise=str(student_data["crc_udise"]), + block=str(student_data["block"]), + district=str(student_data["district"]), + grade=str(student_data["grade"]), + meena_manch_participation=str(student_data["meena_manch_participation"]), + ) + + # ---------------- Answers ---------------- + + answers = [] + + for item in answers_data: + if not isinstance(item, dict): - raise ValueError("Every AI response item must be an object") + raise ValueError( + "Every answer must be an object" + ) + missing = self.REQUIRED - set(item) + if missing: - raise ValueError(f"AI response is missing required fields: {sorted(missing)}") - extras = set(item) - self.REQUIRED + raise ValueError( + f"AI response is missing required fields: {sorted(missing)}" + ) + + missing_optional = set(self.OPTIONAL_DEFAULTS) - set(item) + if missing_optional: + logger.warning( + "Answer for %s missing optional fields %s; defaulting and flagging for review", + item.get("question_id"), sorted(missing_optional), + ) + for field, default in self.OPTIONAL_DEFAULTS.items(): + item.setdefault(field, default) + + extras = set(item) - self.REQUIRED - set(self.OPTIONAL_DEFAULTS) + if extras: - warnings.warn(f"Ignoring unexpected Gemini response fields: {sorted(extras)}", RuntimeWarning) - answers.append(Answer(item["question_id"], list(item["selected_codes"]), str(item["answer_text"]), float(item["confidence"]), bool(item["review_required"]), str(item["raw_observations"]))) - return answers + warnings.warn( + f"Ignoring unexpected answer fields: {sorted(extras)}", + RuntimeWarning, + ) + + if not isinstance(item["question_id"], str): + raise ValueError("question_id must be a string") + if not isinstance(item["selected_codes"], list) or not all(isinstance(code, str) for code in item["selected_codes"]): + raise ValueError("selected_codes must be an array of strings") + answers.append( + Answer( + question_id=item["question_id"], + selected_codes=item["selected_codes"], + answer_text=str(item["answer_text"]), + confidence=float(item["confidence"]), + review_required=bool(item["review_required"]) or bool(missing_optional), + raw_observations=str(item["raw_observations"]), + ) + ) + + if questions is not None: + self._validate_against_questions(answers, questions) + return student, answers + + @staticmethod + def _validate_against_questions(answers: list[Answer], questions: list[Question]) -> None: + questions_by_id = {question.id: question for question in questions} + seen_ids: set[str] = set() + for answer in answers: + if answer.question_id not in questions_by_id: + raise ValueError(f"Response contains unknown question_id: {answer.question_id}") + if answer.question_id in seen_ids: + raise ValueError(f"Response contains duplicate question_id: {answer.question_id}") + seen_ids.add(answer.question_id) + allowed_codes = questions_by_id[answer.question_id].options + invalid_codes = [code for code in answer.selected_codes if code not in allowed_codes] + if invalid_codes: + raise ValueError( + f"Response for {answer.question_id} contains invalid codes: {invalid_codes}; " + f"allowed codes are {allowed_codes}" + ) diff --git a/core/batch/batch_processor.py b/core/batch/batch_processor.py index 6828ae7f..c3600238 100644 --- a/core/batch/batch_processor.py +++ b/core/batch/batch_processor.py @@ -2,9 +2,9 @@ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path -import hashlib, fitz, cv2, yaml +import hashlib, os, fitz, cv2 import numpy as np -from core.models import Answer, Record +from core.models import Answer, Record, StudentInfo from core.utils import read_json, write_json from core.questionnaire.schema_loader import SchemaLoader from core.questionnaire.prompt_builder import PromptBuilder @@ -14,39 +14,40 @@ from core.preprocessing.noise_removal import NoiseRemover from core.preprocessing.image_quality import ImageQuality from core.localization.booklet_layout import BookletLayout -from core.localization.question_locator import QuestionLocator -from core.localization.crop_generator import CropGenerator from core.localization.page_alignment import PageAligner from core.localization.spread_splitter import SpreadSplitter -from core.checkbox.checkbox_detector import CheckboxDetector -from core.checkbox.checkbox_classifier import CheckboxClassifier -from core.handwriting.handwriting_cropper import HandwritingCropper -from core.handwriting.handwriting_quality import HandwritingQuality -from core.ai.gemini_provider import GeminiProvider +# Checkbox and handwriting per-question CV were removed; we now send page +# images to the LLM and let it extract answers from whole pages. +from core.ai.litellm_provider import LiteLLMProvider from core.ai.response_parser import ResponseParser from core.ai.retry import Retry from core.ai.confidence import ConfidenceEngine from core.rules.rule_engine import RuleEngine from core.questionnaire.validator import Validator +from core.models import StudentInfo +from core.runtime_settings import RuntimeSettings, load_runtime_settings +from core.logging.logger import get_logger + class BatchProcessor: def __init__(self, config_dir: str | Path, output_dir: str | Path): self.config=Path(config_dir); self.output=Path(output_dir); self.output.mkdir(parents=True,exist_ok=True) - self.settings=yaml.safe_load((self.config/"settings.yaml").read_text(encoding="utf-8")); self.questions=SchemaLoader(self.config/"questionnaire.json").load() - self.page_count = int(self.settings["logical_page_count"]) + self.logger = get_logger(f"batch.{self.output.name}", self.output / "logs") + self.settings: RuntimeSettings = load_runtime_settings(); self.questions=SchemaLoader(self.config/"questionnaire.json").load() + self.page_count = self.settings.logical_page_count if self.page_count != max(question.page for question in self.questions): raise ValueError("logical_page_count must match the highest question page") - self.source_spread_count = int(self.settings["source_spread_count"]) + self.source_spread_count = self.settings.source_spread_count self.spread_map = {1: (12, 1), 2: (2, 3), 3: (4, 5), 4: (6, 7), 5: (8, 9), 6: (10, 11)} if self.source_spread_count != len(self.spread_map): raise ValueError("source_spread_count does not match the configured booklet spread map") - self.splitter = SpreadSplitter(float(self.settings["spread_gutter_fraction"])) + self.splitter = SpreadSplitter(self.settings.spread_gutter_fraction) self.layout=BookletLayout(self.config/"pages.json"); self.prompt=PromptBuilder(self.config/"prompts.yaml").build(self.questions) - self.state_path=self.output/self.settings["state_file"] + self.state_path=self.output/self.settings.state_file self.reference_pages = self._load_reference_pages() def _render_pdf_page(self, page: fitz.Page) -> np.ndarray: - pix = page.get_pixmap(dpi=int(self.settings["render_dpi"]), alpha=False) + pix = page.get_pixmap(dpi=self.settings.render_dpi, alpha=False) image = cv2.imdecode(np.frombuffer(pix.tobytes("png"), np.uint8), cv2.IMREAD_COLOR) if image is None: raise ValueError("Unable to render source PDF page") @@ -54,7 +55,7 @@ def _render_pdf_page(self, page: fitz.Page) -> np.ndarray: def _load_reference_pages(self) -> dict[int, np.ndarray]: """Create twelve logical reference pages from the one master booklet.""" - path = (self.config / self.settings["reference_master_pdf"]).resolve() + path = (self.config / self.settings.reference_master_pdf).resolve() if not path.exists(): raise FileNotFoundError(f"Reference master PDF is missing: {path}") document = fitz.open(path) @@ -72,63 +73,196 @@ def estimate(self, pdfs: list[Path]) -> dict: return {"pdfs":len(pdfs),"source_spreads":spreads,"logical_pages":spreads * 2,"estimated_minutes":round(spreads*.16,1),"estimated_ai_cost_usd":round(spreads*.004,2)} def process_folder(self, input_dir: str | Path) -> list[Record]: pdfs=sorted(Path(input_dir).glob("*.pdf")); state=self._state(); pending=[p for p in pdfs if str(p.resolve()) not in state["completed"]] - records=[] - with ThreadPoolExecutor(max_workers=int(self.settings["workers"])) as pool: + self.logger.info( + "Processing folder %s: %d PDFs found, %d already completed, %d pending", + input_dir, len(pdfs), len(pdfs) - len(pending), len(pending), + ) + records=[]; failed=[] + with ThreadPoolExecutor(max_workers=self.settings.workers) as pool: futures={pool.submit(self.process_pdf,p):p for p in pending} for future in as_completed(futures): - record=future.result(); records.append(record); state["completed"][str(futures[future].resolve())]=record.to_dict(); write_json(self.state_path,state) + pdf = futures[future] + try: + record = future.result() + except Exception as exc: + # process_pdf() catches its own failures and returns a + # failed Record instead of raising; this is a last-resort + # guard so one bad PDF can never abort the whole batch or + # stop the remaining Excel report from being written. + self.logger.exception("PDF %s: unexpected failure outside process_pdf", pdf.name) + try: + record_id = hashlib.sha256(pdf.read_bytes()).hexdigest()[:16] + except Exception: + record_id = hashlib.sha256(pdf.name.encode()).hexdigest()[:16] + record = Record( + record_id=record_id, + pdf=pdf.name, + student=StudentInfo(), + answers={}, + confidence=0, + review=True, + audit=[f"Unexpected failure: {type(exc).__name__}: {exc}"], + ) + records.append(record) + # Persist every outcome (success or failure) immediately so a + # PDF that has already been attempted once is never + # reprocessed (and re-billed) on a later run of this folder. + state["completed"][str(pdf.resolve())]=record.to_dict(); write_json(self.state_path,state) + if not record.answers: + failed.append((pdf.name, record.audit)) records.extend(self._record_from_dict(x) for x in state["completed"].values() if x["record_id"] not in {r.record_id for r in records}) - return sorted(records,key=lambda r:r.pdf) + records = sorted(records,key=lambda r:r.pdf) + self.logger.info( + "Folder %s complete: %d PDFs attempted this run, %d failed, %d total records available for the report", + input_dir, len(pending), len(failed), len(records), + ) + for name, reasons in failed: + self.logger.warning("PDF %s did not produce answers: %s", name, "; ".join(reasons) or "unknown reason") + return records def _record_from_dict(self, data: dict) -> Record: - return Record(data["record_id"],data["pdf"],{k:Answer(**v) for k,v in data["answers"].items()},data["confidence"],data["review"],data.get("audit",[])) + return Record( + record_id=data["record_id"], + pdf=data["pdf"], + student=StudentInfo(**data.get("student", {})), + answers={k: Answer(**v) for k, v in data["answers"].items()}, + confidence=data["confidence"], + review=data["review"], + audit=data.get("audit", []), + ) def process_pdf(self, pdf: Path) -> Record: errors=Validator.validate_pdf(pdf); record_id=hashlib.sha256(pdf.read_bytes()).hexdigest()[:16] - if errors: return Record(record_id,pdf.name,{},0,True,errors) - doc=fitz.open(pdf) + self.logger.info("PDF %s: starting (record_id=%s)", pdf.name, record_id) + if errors: + self.logger.warning("PDF %s: failed validation: %s", pdf.name, "; ".join(errors)) + return Record( + record_id=record_id, + pdf=pdf.name, + student=StudentInfo(), + answers={}, + confidence=0, + review=True, + audit=errors, + ) + doc = fitz.open(pdf) if len(doc) != self.source_spread_count: + self.logger.warning( + "PDF %s: unexpected spread count %d (expected %d)", + pdf.name, len(doc), self.source_spread_count, + ) return Record( - record_id, - pdf.name, - {}, - 0, - True, - [f"Expected {self.source_spread_count} two-page spreads; found {len(doc)}"], + record_id=record_id, + pdf=pdf.name, + student=StudentInfo(), + answers={}, + confidence=0, + review=True, + audit=[f"Expected {self.source_spread_count} two-page spreads; found {len(doc)}"], + ) + + # Everything below this point talks to disk, OpenCV, and the LLM + # provider, all of which can fail for a single PDF (a bad scan, + # a network blip, a malformed model response, ...). None of that + # should ever take down the rest of the batch, so it is all caught + # here and turned into a failed-but-recorded Record instead of an + # uncaught exception. This also means a PDF that fails is written to + # processing_state.json exactly once, so re-running this folder will + # not reprocess (and re-bill) it. + try: + # We persist full page images and do not create per-question crops. + pages = {} + student_page = None + page_quality = {} + self.logger.info("PDF %s: splitting and aligning %d spreads", pdf.name, self.source_spread_count) + for spread_number, (left_page, right_page) in self.spread_map.items(): + split = self.splitter.split(self._render_pdf_page(doc[spread_number - 1]), left_page, right_page) + for number, raw_image in ((split.left_page, split.left), (split.right_page, split.right)): + image = Deskewer().correct(NoiseRemover().remove(ShadowRemover().remove(raw_image))) + image = PageAligner().align(image, self.reference_pages[number]) + image = ImageCleaner().clean(image) + + page_path = self.output / "pages" / record_id + page_path.mkdir(parents=True, exist_ok=True) + + page_file = page_path / f"page_{number}.png" + cv2.imwrite(str(page_file), image) + + pages[number] = page_file + if number == 1: + student_page = page_file + page_quality[number] = ImageQuality().score(image) + self.logger.info("PDF %s: %d pages rendered, sending to LLM", pdf.name, len(pages)) + + debug_root = self.output / "debug" + (debug_root / "prompts").mkdir(parents=True, exist_ok=True) + (debug_root / "prompts" / f"{record_id}.txt").write_text(self.prompt, encoding="utf-8") + retry = Retry(self.settings.max_retries, self.settings.retry_backoff_seconds) + provider_name = self.settings.llm_provider.lower() + model = self.settings.openrouter_model if provider_name == "openrouter" else self.settings.gemini_model + provider = LiteLLMProvider( + provider_name, model, ResponseParser(), retry, + self.settings.request_timeout_seconds, + self.settings.max_completion_tokens, + ) + if student_page is None: + raise ValueError("Student information page is missing") + # Provide the student page and all logical page images to the LLM. + request_images = {"STUDENT_INFO": student_page, **pages} + student, ai_answers = provider.extract( + self.prompt, + request_images, + self.questions, + debug_root / "responses" / record_id, + ) + self.logger.info("PDF %s: LLM returned %d answers, scoring confidence", pdf.name, len(ai_answers)) + + ai = {a.question_id: a for a in ai_answers} + answers={}; audit=[]; engine=ConfidenceEngine(); rules=RuleEngine() + for question in self.questions: + answer = ai.get(question.id, Answer(question.id, confidence=0, review_required=True, raw_observations="missing AI answer")) + # Checkbox and handwriting signals are not computed when using + # full-page images; leave them at neutral defaults. + issues=rules.evaluate(answer,question) + llm_confidence = answer.confidence + page_confidence = page_quality[question.page] + rules_passed = not issues + + answer.final_confidence = engine.calculate( + llm_confidence, + page_confidence, + rules_passed, + ) + + answer.review_required=answer.review_required or bool(issues) or answer.final_confidence < self.settings.review_confidence_threshold + self.logger.debug( + "PDF %s question %s: llm_confidence=%.3f page_quality=%.3f rules_passed=%s final_confidence=%.3f review=%s", + pdf.name, question.id, llm_confidence, page_confidence, rules_passed, + answer.final_confidence, answer.review_required, + ) + audit.extend(f"{question.id}: {issue}" for issue in issues); answers[question.id]=answer + total_confidence = sum(a.final_confidence for a in answers.values()) + + confidence = round(total_confidence / len(answers), 3) + self.logger.info( + "PDF %s completed: questions=%d average_confidence=%.3f review=%s", + pdf.name, len(answers), confidence, any(a.review_required for a in answers.values()), + ) + return Record( + record_id=record_id, + pdf=pdf.name, + student=student, + answers=answers, + confidence=confidence, + review=any(a.review_required for a in answers.values()), + audit=audit, + ) + except Exception as exc: + self.logger.exception("PDF %s: processing failed", pdf.name) + return Record( + record_id=record_id, + pdf=pdf.name, + student=StudentInfo(), + answers={}, + confidence=0, + review=True, + audit=[f"Processing failed: {type(exc).__name__}: {exc}"], ) - crops={}; page_quality={}; checkbox_meta={}; writing_quality={}; locator=QuestionLocator(self.layout); generator=CropGenerator(locator,self.output/"crops") - by_page = { - page: [question for question in self.questions if question.page == page] - for page in range(1, self.page_count + 1) - } - for spread_number, (left_page, right_page) in self.spread_map.items(): - split = self.splitter.split(self._render_pdf_page(doc[spread_number - 1]), left_page, right_page) - for number, raw_image in ((split.left_page, split.left), (split.right_page, split.right)): - image = Deskewer().correct(NoiseRemover().remove(ShadowRemover().remove(raw_image))) - image = PageAligner().align(image, self.reference_pages[number]) - image = ImageCleaner().clean(image) - page_quality[number]=ImageQuality().score(image) - for q in by_page[number]: - crop,path=generator.crop(image,number,q.id,record_id); crops[q.id]=path - if "choice" in q.type: - marks=CheckboxDetector().detect(crop,len(q.options)); checkbox_meta[q.id]=CheckboxClassifier().classify(marks,q.options) - else: writing_quality[q.id]=HandwritingQuality().score(HandwritingCropper().prepare(crop)) - debug_root = self.output / "debug" - (debug_root / "prompts").mkdir(parents=True, exist_ok=True) - (debug_root / "prompts" / f"{record_id}.txt").write_text(self.prompt, encoding="utf-8") - provider=GeminiProvider(self.settings["gemini_model"],ResponseParser(),Retry(self.settings["max_retries"],self.settings["retry_backoff_seconds"])) - ai={a.question_id:a for a in provider.extract(self.prompt,crops,self.questions,debug_root / "responses" / record_id)} - answers={}; audit=[]; engine=ConfidenceEngine(); rules=RuleEngine() - for q in self.questions: - answer=ai.get(q.id,Answer(q.id,confidence=0,review_required=True,raw_observations="missing AI answer")) - checkbox=0.; handwriting=writing_quality.get(q.id,0.) - if q.id in checkbox_meta: - detected,checkbox,obs=checkbox_meta[q.id] - # Computer vision contributes confidence only. It must never fill - # an answer when Gemini is unavailable or intentionally returns - # an empty response, because checkbox outlines otherwise look - # like selected answers. - answer.raw_observations=f"{answer.raw_observations}; CV cross-check: {obs}" - issues=rules.evaluate(answer,q); answer.final_confidence=engine.calculate(answer.confidence,checkbox,page_quality[q.page],handwriting,not issues) - answer.review_required=answer.review_required or bool(issues) or answer.final_confidence list[str]: + return [ + f"Q{i} — {self.question_catalog[f'Q{i}']}" + if self.include_questions and self.question_catalog.get(f"Q{i}") else f"Q{i}" + for i in range(1, 35) + ] + def write(self, records: list[Record], path: str | Path) -> Path: path=Path(path); path.parent.mkdir(parents=True,exist_ok=True); wb=Workbook(); responses=wb.active; responses.title="Responses" - header=["Record ID","PDF","Confidence","Review"]+[f"Q{i}" for i in range(1,35)]; responses.append(header) - for c in responses[1]: c.font=Font(bold=True); c.fill=PatternFill("solid",fgColor="D9EAF7") + header = [ + "Record ID", + "PDF", + "Student Name", + "Gender", + "School Name", + "School UDISE", + "CRC Name", + "CRC UDISE", + "Block", + "District", + "Grade", + "Meena Manch Participation", + "Confidence", + "Review", + ] + self._question_headers() + responses.append(header) + for c in responses[1]: + c.font=Font(bold=True) + c.fill=PatternFill("solid",fgColor="D9EAF7") + c.alignment=Alignment(wrap_text=True, vertical="top") + responses.row_dimensions[1].height = 90 if self.include_questions else 25 + for column in range(15, 49): + responses.column_dimensions[responses.cell(1, column).column_letter].width = 42 if self.include_questions else 12 review=wb.create_sheet("Review Queue"); review.append(["Record ID","PDF","Question","AI Answer","Confidence","Reason"]) audit=wb.create_sheet("Audit"); audit.append(["Record ID","Event"]) summary=wb.create_sheet("Summary"); summary.append(["Metric","Value"]) - logs=wb.create_sheet("Processing Log"); logs.append(["Record ID","PDF","Confidence","Review"]) + logs=wb.create_sheet("Processing Log"); logs.append(["Record ID","PDF","Student Name","Confidence","Review"]) for record in records: row = [ - record.record_id, - record.pdf, - record.confidence, - record.review, -] + [ - AnswerMapper.value(record.answers.get(f"Q{i}", Answer(question_id=f"Q{i}"))) - for i in range(1, 35) -] - responses.append(row); logs.append([record.record_id,record.pdf,record.confidence,record.review]) + record.record_id, + record.pdf, + record.student.student_name, + record.student.gender, + record.student.school_name, + record.student.school_udise, + record.student.crc_name, + record.student.crc_udise, + record.student.block, + record.student.district, + record.student.grade, + record.student.meena_manch_participation, + record.confidence, + record.review, + ] + [ + AnswerMapper.value( + record.answers.get( + f"Q{i}", + Answer(question_id=f"Q{i}") + ) + ) + for i in range(1, 35) + ] + responses.append(row); logs.append([record.record_id,record.pdf,record.student.student_name,record.confidence,record.review]) for event in record.audit: audit.append([record.record_id,event]) for answer in record.answers.values(): if answer.review_required: review.append([record.record_id,record.pdf,answer.question_id,AnswerMapper.value(answer),answer.final_confidence,answer.raw_observations]) diff --git a/core/models.py b/core/models.py index 65d36f55..4212e3c2 100644 --- a/core/models.py +++ b/core/models.py @@ -14,11 +14,41 @@ class Answer: raw_observations: str = ""; final_confidence: float = 0.0 def to_dict(self) -> dict[str, Any]: return asdict(self) +@dataclass +class StudentInfo: + student_name: str = "" + gender: str = "" + school_name: str = "" + school_udise: str = "" + crc_name: str = "" + crc_udise: str = "" + block: str = "" + district: str = "" + grade: str = "" + meena_manch_participation: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + @dataclass class Record: - record_id: str; pdf: str; answers: dict[str, Answer]; confidence: float - review: bool; audit: list[str] = field(default_factory=list) + record_id: str + pdf: str + answers: dict[str, Answer] = field(default_factory=dict) + confidence: float = 0.0 + review: bool = False + student: StudentInfo = field(default_factory=StudentInfo) + audit: list[str] = field(default_factory=list) + def to_dict(self) -> dict[str, Any]: - return {"record_id": self.record_id, "pdf": self.pdf, "confidence": self.confidence, - "review": self.review, "audit": self.audit, - "answers": {k: v.to_dict() for k, v in self.answers.items()}} + return { + "record_id": self.record_id, + "pdf": self.pdf, + "student": self.student.to_dict(), + "confidence": self.confidence, + "review": self.review, + "audit": self.audit, + "answers": { + k: v.to_dict() for k, v in self.answers.items() + } + } diff --git a/core/questionnaire/question_catalog.py b/core/questionnaire/question_catalog.py new file mode 100644 index 00000000..49fac035 --- /dev/null +++ b/core/questionnaire/question_catalog.py @@ -0,0 +1,85 @@ +"""Load the fixed question-ID-to-text catalogue used in report headers.""" +from __future__ import annotations + +import csv +from pathlib import Path +import re + + +ID_COLUMNS = ("question_id", "id") +TEXT_COLUMNS = ("question", "question_text", "question_name", "label") + + +def load_question_catalog(path: str | Path) -> dict[str, str]: + """Return question text keyed by ID from a CSV or first Excel worksheet.""" + path = Path(path) + if not path.exists(): + return {} + + if path.suffix.lower() == ".txt": + return _load_numbered_text_catalog(path) + if path.suffix.lower() == ".csv": + with path.open(encoding="utf-8-sig", newline="") as source: + rows = list(csv.DictReader(source)) + elif path.suffix.lower() in {".xlsx", ".xlsm"}: + from openpyxl import load_workbook + + worksheet = load_workbook(path, read_only=True, data_only=True).active + values = list(worksheet.values) + if not values: + return {} + headers = [str(value).strip() if value is not None else "" for value in values[0]] + rows = [dict(zip(headers, row)) for row in values[1:]] + else: + raise ValueError(f"Question catalogue must be a TXT, CSV, or Excel file: {path}") + + if not rows: + return {} + headers = {str(header).strip().lower(): header for header in rows[0]} + id_column = next((headers[name] for name in ID_COLUMNS if name in headers), None) + text_column = next((headers[name] for name in TEXT_COLUMNS if name in headers), None) + if not id_column or not text_column: + raise ValueError( + "Question catalogue needs question_id (or id) and question " + "(or question_text, question_name, label) columns" + ) + + catalogue: dict[str, str] = {} + for row in rows: + question_id = str(row.get(id_column) or "").strip() + question_text = str(row.get(text_column) or "").strip() + if not question_id or not question_text: + continue + if question_id in catalogue: + raise ValueError(f"Duplicate question ID in catalogue: {question_id}") + catalogue[question_id] = question_text + return catalogue + + +def _load_numbered_text_catalog(path: Path) -> dict[str, str]: + """Read question blocks in the supplied ``1.`` … ``34.`` text document.""" + catalogue: dict[str, str] = {} + question_number: int | None = None + question_lines: list[str] = [] + + def save_question() -> None: + if question_number is None: + return + text = " ".join(question_lines).strip() + if text: + catalogue[f"Q{question_number}"] = text + + for line in path.read_text(encoding="utf-8-sig").splitlines(): + match = re.match(r"^\s*(\d+)\.\s*$", line) + if match: + save_question() + question_number = int(match.group(1)) + question_lines = [] + continue + # This describes the response format, rather than the question itself. + if line.strip().startswith("खुला प्रश्न"): + continue + if question_number is not None and line.strip(): + question_lines.append(line.strip()) + save_question() + return catalogue diff --git a/core/runtime_settings.py b/core/runtime_settings.py new file mode 100644 index 00000000..48404d06 --- /dev/null +++ b/core/runtime_settings.py @@ -0,0 +1,94 @@ +"""Typed runtime configuration loaded exclusively from environment variables.""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Mapping + + +@dataclass(frozen=True) +class RuntimeSettings: + port: int + include_questions_in_reports: bool + llm_provider: str + openrouter_model: str + gemini_model: str + workers: int + render_dpi: int + source_spread_count: int + logical_page_count: int + reference_master_pdf: str + spread_gutter_fraction: float + min_image_quality: float + review_confidence_threshold: float + max_retries: int + retry_backoff_seconds: float + request_timeout_seconds: float + max_completion_tokens: int + output_workbook: str + question_catalog: str + state_file: str + + +_FIELDS = { + "port": int, + "workers": int, "render_dpi": int, "source_spread_count": int, + "logical_page_count": int, "reference_master_pdf": str, + "spread_gutter_fraction": float, "min_image_quality": float, + "review_confidence_threshold": float, "max_retries": int, + "retry_backoff_seconds": float, "request_timeout_seconds": float, + "max_completion_tokens": int, + "output_workbook": str, + "question_catalog": str, "state_file": str, +} + +def load_runtime_settings(environ: Mapping[str, str] | None = None) -> RuntimeSettings: + """Load all non-secret application configuration from ``.env`` / environment.""" + environment = environ if environ is not None else os.environ + values: dict[str, object] = {} + missing: list[str] = [] + question_headers = environment.get("IS_QUESTIONS_NEED_CSV", "").strip().lower() + if question_headers in {"yes", "true", "1"}: + values["include_questions_in_reports"] = True + elif question_headers in {"no", "false", "0"}: + values["include_questions_in_reports"] = False + elif not question_headers: + missing.append("IS_QUESTIONS_NEED_CSV") + else: + raise ValueError("IS_QUESTIONS_NEED_CSV must be Yes or No") + values["llm_provider"] = environment.get("LLM_PROVIDER", "").strip().lower() + if not values["llm_provider"]: + missing.append("LLM_PROVIDER") + for field, converter in _FIELDS.items(): + variable = "PORT" if field == "port" else f"BIHAR_{field.upper()}" + raw_value = environment.get(variable) + if raw_value is None or not raw_value.strip(): + missing.append(variable) + continue + try: + values[field] = converter(raw_value) + except ValueError as error: + raise ValueError(f"{variable} has an invalid value: {raw_value!r}") from error + + provider = str(values["llm_provider"]) + supported = {"openrouter", "gemini", "direct_gemini", "google"} + # Only validate allowed values when a provider was explicitly set; if + # missing/empty the aggregated `missing` error below should surface it. + if provider and provider not in supported: + raise ValueError( + "LLM_PROVIDER must be one of: " + ", ".join(sorted(supported)) + ) + model_variable = "OPENROUTER_MODEL" if provider == "openrouter" else "GEMINI_MODEL" + model = environment.get(model_variable) + if not model or not model.strip(): + missing.append(model_variable) + values["openrouter_model"] = environment.get("OPENROUTER_MODEL", "") + values["gemini_model"] = environment.get("GEMINI_MODEL", "") + if provider == "openrouter": + if not environment.get("OPENROUTER_API_KEY", "").strip(): + missing.append("OPENROUTER_API_KEY") + elif not (environment.get("GEMINI_API_KEY", "").strip() or environment.get("GOOGLE_API_KEY", "").strip()): + missing.append("GEMINI_API_KEY (or GOOGLE_API_KEY)") + if missing: + raise RuntimeError("Missing required environment variables: " + ", ".join(missing)) + return RuntimeSettings(**values) diff --git a/input_pdfs/20260707020844.pdf b/input_pdfs/20260707020844.pdf deleted file mode 100644 index 3164f866..00000000 Binary files a/input_pdfs/20260707020844.pdf and /dev/null differ diff --git a/input_pdfs/20260707021053.pdf b/input_pdfs/20260707021053.pdf deleted file mode 100644 index 236d7e40..00000000 Binary files a/input_pdfs/20260707021053.pdf and /dev/null differ diff --git a/input_pdfs/20260707021349.pdf b/input_pdfs/20260707021349.pdf deleted file mode 100644 index 01d83e72..00000000 Binary files a/input_pdfs/20260707021349.pdf and /dev/null differ diff --git a/input_pdfs/20260707021512.pdf b/input_pdfs/20260707021512.pdf deleted file mode 100644 index 6473f8d5..00000000 Binary files a/input_pdfs/20260707021512.pdf and /dev/null differ diff --git a/input_pdfs/20260707021644.pdf b/input_pdfs/20260707021644.pdf deleted file mode 100644 index 57b12eec..00000000 Binary files a/input_pdfs/20260707021644.pdf and /dev/null differ diff --git a/main.py b/main.py index f1b3b246..662e7bce 100644 --- a/main.py +++ b/main.py @@ -1,28 +1,106 @@ """Command-line entry point for fixed-booklet batch processing.""" from __future__ import annotations -import argparse, json, shutil, time -import os +import argparse, importlib.util, json, time +from datetime import datetime from pathlib import Path +import re from dotenv import load_dotenv from core.batch.batch_processor import BatchProcessor from core.excel.excel_writer import ExcelWriter from core.analytics.dashboard import Dashboard from core.merger.json_merger import JsonMerger +from core.questionnaire.question_catalog import load_question_catalog +from core.runtime_settings import load_runtime_settings +from core.logging.logger import get_logger + def config_path() -> Path: return Path(__file__).parent / "config" + + +def validate_runtime_dependencies() -> None: + if importlib.util.find_spec("litellm") is None: + raise RuntimeError( + "Missing dependency 'litellm'. Install project dependencies with: " + "pip install -r requirements.txt" + ) + def main() -> None: + parser=argparse.ArgumentParser(description="21stCenturyTool V2") sub=parser.add_subparsers(dest="command",required=True) process=sub.add_parser("process"); process.add_argument("--input",required=True); process.add_argument("--output",required=True) estimate=sub.add_parser("estimate"); estimate.add_argument("--input",required=True); estimate.add_argument("--output",default=".") args=parser.parse_args() load_dotenv() - if args.command == "process" and not os.getenv("GEMINI_API_KEY"): - parser.error("GEMINI_API_KEY is missing. Create .env from .env.example or set it in this terminal before processing.") - processor=BatchProcessor(config_path(),args.output); pdfs=list(Path(args.input).glob("*.pdf")) - if args.command=="estimate": print(json.dumps(processor.estimate(pdfs),indent=2)); return - started=time.monotonic(); records=JsonMerger().merge(processor.process_folder(args.input)) - workbook=ExcelWriter().write(records,Path(args.output)/processor.settings["output_workbook"]) + logger = get_logger("bihar-assessment") + try: + startup_settings = load_runtime_settings() + validate_runtime_dependencies() + except (RuntimeError, ValueError) as error: + parser.error(f"Configuration validation failed: {error}") + logger.info("Starting %s command", args.command) + logger.info("Using LLM provider=%s model=%s", startup_settings.llm_provider, + startup_settings.openrouter_model if startup_settings.llm_provider == "openrouter" else startup_settings.gemini_model) + + + model = (startup_settings.openrouter_model if startup_settings.llm_provider == "openrouter" + else startup_settings.gemini_model).split("/")[-1] + # Sanitize model name for use in filesystem paths (avoid characters like ':' on Windows) + model = re.sub(r"[^A-Za-z0-9._-]", "_", model) + + run_name = datetime.now().strftime("%Y%m%d_%H%M%S") + f"_{model}" + + base_output = Path(args.output) + base_output.mkdir(parents=True, exist_ok=True) + + # Create a timestamped subdirectory for this run's artifacts (workbook, metrics, debug) + run_output = base_output / run_name + run_output.mkdir(parents=True, exist_ok=True) + + # Use the run-specific output directory for processing so each run's + # `processing_state.json`, `debug`, `pages`, and `logs` live inside the + # timestamped run folder and do not mix with other runs. + # Also set `USAGE_LOG_DIR` so API usage logs are written under the run. + import os + os.environ.setdefault("USAGE_LOG_DIR", str(run_output)) + processor = BatchProcessor(config_path(), run_output) + pdfs = list(Path(args.input).glob("*.pdf")) + logger.info("Discovered %d input PDFs", len(pdfs)) + + if args.command=="estimate": + print(json.dumps(processor.estimate(pdfs),indent=2)) + return + + started=time.monotonic() + try: + records=JsonMerger().merge(processor.process_folder(args.input)) + except Exception: + # process_folder() already catches per-PDF failures and keeps going; + # this is a last-resort fallback so a truly unexpected crash still + # produces a report from whatever was saved to processing_state.json + # before the failure, instead of losing the whole run's output. + logger.exception("process_folder failed unexpectedly; falling back to processing_state.json") + state = processor._state() + records = JsonMerger().merge([processor._record_from_dict(x) for x in state["completed"].values()]) + succeeded = sum(1 for r in records if r.answers) + failed = len(records) - succeeded + logger.info( + "Questionnaire processing completed for %d PDFs (%d succeeded, %d failed)", + len(records), succeeded, failed, + ) + if failed: + for record in records: + if not record.answers: + logger.warning("PDF %s failed: %s", record.pdf, "; ".join(record.audit) or "unknown reason") + question_catalog = load_question_catalog(processor.config / processor.settings.question_catalog) \ + if processor.settings.include_questions_in_reports else {} + if processor.settings.include_questions_in_reports and not question_catalog: + logger.warning("Question catalogue is empty; response headers will use question IDs only") + workbook=ExcelWriter(question_catalog, processor.settings.include_questions_in_reports).write( + records, run_output/processor.settings.output_workbook + ) + metrics=Dashboard().metrics(records,time.monotonic()-started) - (Path(args.output)/"metrics.json").write_text(json.dumps(metrics,indent=2),encoding="utf-8") - print(f"Processed {len(records)} questionnaires. Workbook: {workbook}") + + (run_output/"metrics.json").write_text(json.dumps(metrics,indent=2),encoding="utf-8") + logger.info("Processed %d questionnaires; workbook: %s", len(records), workbook) if __name__ == "__main__": main() diff --git a/merge_drop2_results.py b/merge_drop2_results.py new file mode 100644 index 00000000..a374b053 --- /dev/null +++ b/merge_drop2_results.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Merge every result_Drop_2_Batch_* run into one combined Drop 2 report. + +Reads the latest run from each result_Drop_2_Batch_* folder and combines +processing_state.json, debug/ (prompts + raw LLM responses), pages/ +(rendered page images) and logs/ into one result_Drop_2/ folder, plus +writes a combined questionnaire_results.xlsx and metrics.json. Does not +call the LLM or touch the per-batch folders; purely local consolidation +of already-processed results. +""" +from __future__ import annotations +import glob +import json +import os +import shutil +from pathlib import Path + +from dotenv import load_dotenv + +from core.analytics.dashboard import Dashboard +from core.excel.excel_writer import ExcelWriter +from core.merger.json_merger import JsonMerger +from core.models import Answer, Record, StudentInfo +from core.questionnaire.question_catalog import load_question_catalog +from core.runtime_settings import load_runtime_settings +from core.utils import write_json +from core.logging.logger import get_logger + + +def config_path() -> Path: + return Path(__file__).parent / "config" + + +def record_from_dict(data: dict) -> Record: + return Record( + record_id=data["record_id"], + pdf=data["pdf"], + student=StudentInfo(**data.get("student", {})), + answers={k: Answer(**v) for k, v in data.get("answers", {}).items()}, + confidence=data.get("confidence", 0.0), + review=data.get("review", False), + audit=data.get("audit", []), + ) + + +def main() -> None: + load_dotenv() + logger = get_logger("merge-drop2") + settings = load_runtime_settings() + + batch_dirs = sorted(glob.glob("result_Drop_2_Batch_*")) + logger.info("Found %d batch result folders", len(batch_dirs)) + + output_dir = Path("result_Drop_2") + (output_dir / "debug" / "prompts").mkdir(parents=True, exist_ok=True) + (output_dir / "debug" / "responses").mkdir(parents=True, exist_ok=True) + (output_dir / "pages").mkdir(parents=True, exist_ok=True) + (output_dir / "logs").mkdir(parents=True, exist_ok=True) + + merged_state: dict[str, dict] = {} + total_actual_cost = 0.0 + for batch_dir in batch_dirs: + run_dirs = sorted(glob.glob(os.path.join(batch_dir, "*"))) + if not run_dirs: + logger.warning("%s: no run folders found, skipping", batch_dir) + continue + latest_run = Path(run_dirs[-1]) + state_path = latest_run / "processing_state.json" + if not state_path.exists(): + logger.warning("%s: no processing_state.json found, skipping", latest_run) + continue + state = json.loads(state_path.read_text(encoding="utf-8")) + completed = state.get("completed", {}) + logger.info("%s: %d records (from %s)", batch_dir, len(completed), latest_run) + merged_state.update(completed) + + # debug/prompts and debug/responses are keyed by record_id (a content + # hash of the source PDF), so folders never collide across batches. + for prompt_file in (latest_run / "debug" / "prompts").glob("*.txt"): + shutil.copy2(prompt_file, output_dir / "debug" / "prompts" / prompt_file.name) + responses_dir = latest_run / "debug" / "responses" + if responses_dir.is_dir(): + for record_dir in responses_dir.iterdir(): + shutil.copytree(record_dir, output_dir / "debug" / "responses" / record_dir.name, dirs_exist_ok=True) + metadata_path = record_dir / "response_metadata.json" + if metadata_path.exists(): + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + total_actual_cost += metadata.get("usage", {}).get("cost", 0) or 0 + + pages_dir = latest_run / "pages" + if pages_dir.is_dir(): + for record_dir in pages_dir.iterdir(): + shutil.copytree(record_dir, output_dir / "pages" / record_dir.name, dirs_exist_ok=True) + + log_file = latest_run / "logs" / "processing.log" + if log_file.exists(): + shutil.copy2(log_file, output_dir / "logs" / f"{batch_dir}.log") + + records = JsonMerger().merge([record_from_dict(x) for x in merged_state.values()]) + records = sorted(records, key=lambda r: r.pdf) + logger.info("Merged total: %d unique records", len(records)) + + write_json(output_dir / settings.state_file, {"completed": merged_state}) + + question_catalog = ( + load_question_catalog(config_path() / settings.question_catalog) + if settings.include_questions_in_reports else {} + ) + workbook = ExcelWriter(question_catalog, settings.include_questions_in_reports).write( + records, output_dir / settings.output_workbook + ) + logger.info("Combined workbook written: %s", workbook) + + # config/model_pricing.yaml has no entry for the model actually in use, + # so Dashboard's built-in estimated_cost silently returns 0. Use the + # real per-call cost OpenRouter already reported in each + # response_metadata.json instead. + metrics = Dashboard().metrics(records, elapsed_seconds=0, estimated_cost=round(total_actual_cost, 4)) + (output_dir / "metrics.json").write_text(json.dumps(metrics, indent=2), encoding="utf-8") + logger.info("Combined metrics written: %s (actual_cost_usd=%.2f)", output_dir / "metrics.json", total_actual_cost) + + succeeded = sum(1 for r in records if r.answers) + failed = len(records) - succeeded + print(f"Merged {len(records)} records ({succeeded} succeeded, {failed} failed) -> {workbook}") + print(f"Actual OpenRouter spend across merged calls: ${total_actual_cost:.2f}") + + +if __name__ == "__main__": + main() diff --git a/merge_drop3_results.py b/merge_drop3_results.py new file mode 100644 index 00000000..d4f6c570 --- /dev/null +++ b/merge_drop3_results.py @@ -0,0 +1,124 @@ +"""Merge all Drop 3 result folders (original run + network retries) into one. + +Unlike the Drop 2 merge, Drop 3's retries reprocessed PDFs that already had +a (failed) entry from the original run, under a different input folder path. +Since processing_state.json keys by absolute PDF path, the same PDF gets a +different key in each source folder -- so this merges by record_id instead +(a content hash of the PDF, stable across folders) and processes sources in +chronological order so a later retry's outcome always overwrites an earlier +attempt's for the same PDF, whether that's turning a failure into a success +or recording a fresh failure reason. +""" +import glob +import json +import shutil +from pathlib import Path +from dotenv import load_dotenv + +from core.excel.excel_writer import ExcelWriter +from core.analytics.dashboard import Dashboard +from core.merger.json_merger import JsonMerger +from core.models import Answer, Record, StudentInfo +from core.questionnaire.question_catalog import load_question_catalog +from core.runtime_settings import load_runtime_settings +from core.utils import write_json +from core.logging.logger import get_logger + +load_dotenv(".env") +logger = get_logger("merge-drop3-results") +settings = load_runtime_settings() + +# Chronological order matters: each later source's outcome for a given +# record_id overwrites the earlier one. Batch_1/2/3 were an earlier, +# separate set of PDFs (never included in the drop_3/ folder used for the +# big 156-PDF run) -- confirmed no filename overlap between them. +# Each entry is (folder, "latest") to take only that folder's most recent +# run, or (folder, "all") to walk every run subfolder in order -- needed for +# result_drop_3_failed_rerun, which has 22 separate single-PDF runs (one per +# retried PDF), not one run covering all of them. +sources = [ + (Path("result_drop_3_batch_1"), "latest"), + (Path("result_drop_3_batch_2"), "latest"), + (Path("result_drop_3_batch_3"), "latest"), + (Path("result_drop_3"), "latest"), + (Path("result_Drop_3_network_retry"), "latest"), + (Path("result_Drop_3_network_retry_2"), "latest"), + (Path("result_drop_3_failed_rerun"), "all"), +] +output_dir = Path("Drop_3_final") +(output_dir / "debug" / "prompts").mkdir(parents=True, exist_ok=True) +(output_dir / "debug" / "responses").mkdir(parents=True, exist_ok=True) +(output_dir / "pages").mkdir(parents=True, exist_ok=True) +(output_dir / "logs").mkdir(parents=True, exist_ok=True) + +merged_by_record_id: dict[str, dict] = {} +total_actual_cost = 0.0 +for src, mode in sources: + all_run_dirs = sorted(src.glob("*")) + if not all_run_dirs: + logger.warning("%s: no run folders found, skipping", src) + continue + run_dirs = all_run_dirs if mode == "all" else [all_run_dirs[-1]] + + for latest_run in run_dirs: + state = json.loads((latest_run / "processing_state.json").read_text(encoding="utf-8")) + completed = state.get("completed", {}) + logger.info("%s: %d records (from %s)", src, len(completed), latest_run) + for record in completed.values(): + merged_by_record_id[record["record_id"]] = record + + for prompt_file in (latest_run / "debug" / "prompts").glob("*.txt"): + shutil.copy2(prompt_file, output_dir / "debug" / "prompts" / prompt_file.name) + responses_dir = latest_run / "debug" / "responses" + if responses_dir.is_dir(): + for record_dir in responses_dir.iterdir(): + # A PDF retried across sources/runs shares the same + # record_id, so a later run's copytree here overwrites an + # earlier attempt's debug files -- only the winning (most + # recent) attempt's raw response ends up on disk, which is + # fine for browsing. But every attempt was a real billed + # call, so cost must be summed *now*, per run, before it can + # be overwritten. + metadata_path = record_dir / "response_metadata.json" + if metadata_path.exists(): + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + total_actual_cost += metadata.get("usage", {}).get("cost", 0) or 0 + shutil.copytree(record_dir, output_dir / "debug" / "responses" / record_dir.name, dirs_exist_ok=True) + pages_dir = latest_run / "pages" + if pages_dir.is_dir(): + for record_dir in pages_dir.iterdir(): + shutil.copytree(record_dir, output_dir / "pages" / record_dir.name, dirs_exist_ok=True) + log_file = latest_run / "logs" / "processing.log" + if log_file.exists(): + shutil.copy2(log_file, output_dir / "logs" / f"{src.name}_{latest_run.name}.log") + + +def record_from_dict(data: dict) -> Record: + return Record( + record_id=data["record_id"], pdf=data["pdf"], + student=StudentInfo(**data.get("student", {})), + answers={k: Answer(**v) for k, v in data.get("answers", {}).items()}, + confidence=data.get("confidence", 0.0), review=data.get("review", False), + audit=data.get("audit", []), + ) + + +records = JsonMerger().merge([record_from_dict(x) for x in merged_by_record_id.values()]) +records = sorted(records, key=lambda r: r.pdf) +logger.info("Merged total: %d unique PDFs", len(records)) + +write_json(output_dir / settings.state_file, {"completed": {r.record_id: r.to_dict() for r in records}}) +question_catalog = ( + load_question_catalog(Path("config") / settings.question_catalog) + if settings.include_questions_in_reports else {} +) +workbook = ExcelWriter(question_catalog, settings.include_questions_in_reports).write( + records, output_dir / settings.output_workbook +) +metrics = Dashboard().metrics(records, elapsed_seconds=0, estimated_cost=round(total_actual_cost, 4)) +(output_dir / "metrics.json").write_text(json.dumps(metrics, indent=2), encoding="utf-8") + +succeeded = sum(1 for r in records if r.answers) +failed = len(records) - succeeded +print(f"Merged {len(records)} unique PDFs ({succeeded} succeeded, {failed} failed) -> {workbook}") +print(f"Total actual OpenRouter spend across every attempt (incl. overwritten retries): ${total_actual_cost:.4f}") diff --git a/rebuild_drop1_state.py b/rebuild_drop1_state.py new file mode 100644 index 00000000..f0450e8c --- /dev/null +++ b/rebuild_drop1_state.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""Rebuild processing_state.json for the Drop 1 run from saved debug artifacts. + +The Drop 1 run (results/20260819_160942_gemini-3.5-flash) predates the +per-PDF error-handling fix: it crashed partway through with the old code, +so processing_state.json only reflects the 27 PDFs completed before the +crash, even though debug/responses/ and pages/ show all 80 were actually +attempted (77 succeeded, 3 failed). This reconstructs every Record using +the real pipeline scoring logic (ResponseParser + RuleEngine + +ConfidenceEngine) from the saved raw LLM response and rendered page +images -- no LLM calls, no re-billing -- and writes a complete state file +so re-running main.py on input_pdf-drop-1 won't reprocess anything. +""" +from __future__ import annotations +import hashlib +from pathlib import Path + +import cv2 +from dotenv import load_dotenv + +from core.ai.confidence import ConfidenceEngine +from core.ai.response_parser import ResponseParser +from core.models import Answer, Record, StudentInfo +from core.preprocessing.image_quality import ImageQuality +from core.questionnaire.schema_loader import SchemaLoader +from core.rules.rule_engine import RuleEngine +from core.runtime_settings import load_runtime_settings +from core.utils import write_json +from core.logging.logger import get_logger + + +def config_path() -> Path: + return Path(__file__).parent / "config" + + +def build_record(pdf: Path, record_id: str, run_dir: Path, questions, review_confidence_threshold: float) -> Record: + response_dir = run_dir / "debug" / "responses" / record_id + exception_path = response_dir / "exception.txt" + raw_response_path = response_dir / "raw_response.txt" + + if exception_path.exists() or not raw_response_path.exists(): + if exception_path.exists(): + reason = exception_path.read_text(encoding="utf-8").strip().splitlines()[-1] + else: + reason = "No response recorded for this PDF" + return Record( + record_id=record_id, pdf=pdf.name, student=StudentInfo(), answers={}, + confidence=0, review=True, audit=[f"Processing failed: {reason}"], + ) + + text = raw_response_path.read_text(encoding="utf-8") + try: + student, ai_answers = ResponseParser().parse(text, questions) + except Exception as exc: + return Record( + record_id=record_id, pdf=pdf.name, student=StudentInfo(), answers={}, + confidence=0, review=True, audit=[f"Processing failed: {type(exc).__name__}: {exc}"], + ) + + page_quality: dict[int, float] = {} + pages_dir = run_dir / "pages" / record_id + if pages_dir.is_dir(): + for page_file in pages_dir.glob("page_*.png"): + number = int(page_file.stem.split("_")[1]) + image = cv2.imread(str(page_file)) + page_quality[number] = ImageQuality().score(image) if image is not None else 0.5 + + ai = {a.question_id: a for a in ai_answers} + answers: dict[str, Answer] = {} + audit: list[str] = [] + engine = ConfidenceEngine() + rules = RuleEngine() + for question in questions: + answer = ai.get( + question.id, + Answer(question.id, confidence=0, review_required=True, raw_observations="missing AI answer"), + ) + issues = rules.evaluate(answer, question) + page_confidence = page_quality.get(question.page, 0.5) + answer.final_confidence = engine.calculate(answer.confidence, page_confidence, not issues) + answer.review_required = ( + answer.review_required or bool(issues) or answer.final_confidence < review_confidence_threshold + ) + audit.extend(f"{question.id}: {issue}" for issue in issues) + answers[question.id] = answer + + confidence = round(sum(a.final_confidence for a in answers.values()) / len(answers), 3) if answers else 0 + return Record( + record_id=record_id, pdf=pdf.name, student=student, answers=answers, + confidence=confidence, review=any(a.review_required for a in answers.values()), audit=audit, + ) + + +def main() -> None: + load_dotenv() + logger = get_logger("rebuild-drop1-state") + settings = load_runtime_settings() + questions = SchemaLoader(config_path() / "questionnaire.json").load() + + run_dir = Path("results/20260819_160942_gemini-3.5-flash") + input_dir = Path("input_pdf-drop-1") + pdfs = sorted(input_dir.glob("*.pdf")) + logger.info("Rebuilding state for %d input PDFs from %s", len(pdfs), run_dir) + + completed: dict[str, dict] = {} + for pdf in pdfs: + record_id = hashlib.sha256(pdf.read_bytes()).hexdigest()[:16] + record = build_record(pdf, record_id, run_dir, questions, settings.review_confidence_threshold) + completed[str(pdf.resolve())] = record.to_dict() + logger.info("%s -> record_id=%s answers=%d", pdf.name, record_id, len(record.answers)) + + write_json(run_dir / settings.state_file, {"completed": completed}) + succeeded = sum(1 for v in completed.values() if v["answers"]) + failed = len(completed) - succeeded + logger.info("Rebuilt state: %d total (%d succeeded, %d failed)", len(completed), succeeded, failed) + print(f"Rebuilt {run_dir / settings.state_file}: {len(completed)} total ({succeeded} succeeded, {failed} failed)") + + +if __name__ == "__main__": + main() diff --git a/regenerate_excel.py b/regenerate_excel.py new file mode 100644 index 00000000..b70e374e --- /dev/null +++ b/regenerate_excel.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""Regenerate Excel from existing processing_state.json without reprocessing PDFs.""" +from __future__ import annotations +import json, sys, os +from pathlib import Path +from dotenv import load_dotenv +from core.models import Answer, Record, StudentInfo +from core.excel.excel_writer import ExcelWriter +from core.questionnaire.question_catalog import load_question_catalog +from core.logging.logger import get_logger + +def config_path() -> Path: + return Path(__file__).parent / "config" + +def regenerate_excel(results_dir: str | Path, output_workbook_name: str = "questionnaire_results.xlsx") -> Path: + """ + Regenerate Excel workbook from processing_state.json in a results directory. + + Args: + results_dir: Path to the timestamped results folder (e.g., results/20260819_160942_gemini-3.5-flash/) + output_workbook_name: Name of the output Excel file + + Returns: + Path to the generated workbook + """ + load_dotenv() + results_dir = Path(results_dir) + logger = get_logger("regenerate-excel") + + if not results_dir.exists(): + raise FileNotFoundError(f"Results directory not found: {results_dir}") + + state_file = results_dir / "processing_state.json" + if not state_file.exists(): + raise FileNotFoundError(f"No processing_state.json found in: {results_dir}") + + logger.info(f"Reading processing state from {state_file}") + state = json.loads(state_file.read_text(encoding="utf-8")) + + # Reconstruct Record objects from saved state + records = [] + for pdf_path, data in state.get("completed", {}).items(): + record = Record( + record_id=data["record_id"], + pdf=data["pdf"], + student=StudentInfo(**data.get("student", {})), + answers={k: Answer(**v) for k, v in data.get("answers", {}).items()}, + confidence=data.get("confidence", 0.0), + review=data.get("review", False), + audit=data.get("audit", []), + ) + records.append(record) + + logger.info(f"Loaded {len(records)} records from processing state") + + if not records: + logger.warning("No completed records found in processing_state.json") + return None + + # Load question catalog if available + include_questions = os.getenv("IS_QUESTIONS_NEED_CSV", "0") == "1" + question_catalog = {} + if include_questions: + try: + question_catalog = load_question_catalog(config_path() / "questions.txt") + except Exception as e: + logger.warning(f"Could not load question catalog: {e}") + + # Write Excel + workbook_path = results_dir / output_workbook_name + writer = ExcelWriter(question_catalog, include_questions) + excel_file = writer.write(records, workbook_path) + + logger.info(f"Excel workbook generated: {excel_file}") + return excel_file + +def regenerate_excel_from_raw_responses(results_dir: str | Path, output_workbook_name: str = "questionnaire_results.xlsx") -> Path: + """ + Regenerate Excel from ALL raw_response.txt files with proper PDF mapping and metadata. + + Args: + results_dir: Path to the timestamped results folder + output_workbook_name: Name of the output Excel file + + Returns: + Path to the generated workbook + """ + load_dotenv() + results_dir = Path(results_dir) + logger = get_logger("regenerate-excel") + + if not results_dir.exists(): + raise FileNotFoundError(f"Results directory not found: {results_dir}") + + debug_dir = results_dir / "debug" / "responses" + if not debug_dir.exists(): + raise FileNotFoundError(f"No debug/responses directory found in: {results_dir}") + + # Extract PDF names from processing logs + log_file = results_dir / "logs" / "processing.log" + logged_pdfs = set() # Set of PDF names that were logged as completed + record_to_pdf = {} # record_id -> PDF name mapping + + if log_file.exists(): + log_content = log_file.read_text(encoding="utf-8") + import re + # Pattern: "PDF completed:" + pattern = r"PDF\s+([^\s]+\.pdf)\s+completed:" + for match in re.finditer(pattern, log_content): + pdf_name = match.group(1) + logged_pdfs.add(pdf_name) + + logger.info(f"Found {len(logged_pdfs)} PDF names in processing logs") + + # Hash all input PDFs to build record_id -> PDF name mapping + import hashlib + input_dir = Path("input_pdf-drop-1") + if input_dir.exists(): + for pdf_file in sorted(input_dir.glob("*.pdf")): + try: + file_hash = hashlib.sha256(pdf_file.read_bytes()).hexdigest()[:16] + record_to_pdf[file_hash] = pdf_file.name + except Exception as e: + logger.debug(f"Could not hash {pdf_file}: {e}") + + # Find all raw_response.txt files + raw_response_files = sorted(debug_dir.glob("*/raw_response.txt")) + logger.info(f"Found {len(raw_response_files)} raw response files in {debug_dir}") + + if not raw_response_files: + raise FileNotFoundError(f"No raw_response.txt files found in: {debug_dir}") + + + # Identify corrupted PDFs + corrupted_record_ids = set() + all_record_ids = {f.parent.name for f in raw_response_files} + parsed_record_ids = set() + + # Reconstruct Record objects from raw responses + records = [] + skipped = 0 + + for response_file in raw_response_files: + record_id = response_file.parent.name + is_corrupted = False + + try: + response_text = response_file.read_text(encoding="utf-8", errors="replace") + + # Handle markdown-wrapped JSON (```json ... ```) + if response_text.strip().startswith("```json"): + json_start = response_text.find("{") + json_end = response_text.rfind("}") + if json_start >= 0 and json_end > json_start: + response_text = response_text[json_start:json_end+1] + + try: + raw_json = json.loads(response_text) + except json.JSONDecodeError as e: + # Try to recover + json_start = response_text.find("{") + last_answer_end = response_text.rfind(']') + if json_start >= 0 and last_answer_end > json_start: + recovery_text = response_text[json_start:last_answer_end+1] + '}' + try: + raw_json = json.loads(recovery_text) + logger.info(f"Recovered partial JSON for {record_id}") + except: + raise e + else: + raise e + + parsed_record_ids.add(record_id) + + # Get PDF name from mapping or use record_id as fallback + pdf_name = record_to_pdf.get(record_id, f"unknown_{record_id}.pdf") + + # Parse student info + student_data = raw_json.get("student", {}) + student = StudentInfo( + student_name=student_data.get("student_name", ""), + gender=student_data.get("gender", ""), + school_name=student_data.get("school_name", ""), + school_udise=student_data.get("school_udise", ""), + crc_name=student_data.get("crc_name", ""), + crc_udise=student_data.get("crc_udise", ""), + block=student_data.get("block", ""), + district=student_data.get("district", ""), + grade=student_data.get("grade", ""), + meena_manch_participation=student_data.get("meena_manch_participation", ""), + ) + + # Parse answers + answers = {} + confidence_values = [] + review_flag = False + + for answer_data in raw_json.get("answers", []): + q_id = answer_data.get("question_id", "") + answer = Answer( + question_id=q_id, + selected_codes=answer_data.get("selected_codes", []), + answer_text=answer_data.get("answer_text", ""), + confidence=answer_data.get("confidence", 0.0), + review_required=answer_data.get("review_required", False), + raw_observations=answer_data.get("raw_observations", ""), + final_confidence=answer_data.get("confidence", 0.0), + ) + answers[q_id] = answer + confidence_values.append(answer.confidence) + if answer.review_required: + review_flag = True + + # Calculate average confidence + avg_confidence = sum(confidence_values) / len(confidence_values) if confidence_values else 0.0 + + # Create Record + record = Record( + record_id=record_id, + pdf=pdf_name, + student=student, + answers=answers, + confidence=avg_confidence, + review=review_flag, + audit=[], + ) + records.append(record) + + except Exception as e: + logger.warning(f"Failed to parse {response_file}: {e}") + corrupted_record_ids.add(record_id) + skipped += 1 + + logger.info(f"Loaded {len(records)} records, skipped {skipped} (corrupted)") + logger.info(f"Corrupted record_ids: {sorted(corrupted_record_ids)}") + + if corrupted_record_ids: + corrupted_pdfs = [(record_to_pdf.get(rid) or f"unknown_{rid}", rid) for rid in sorted(corrupted_record_ids)] + logger.warning(f"Corrupted PDFs: {corrupted_pdfs}") + + if not records: + raise RuntimeError("No valid records could be parsed from raw responses") + + # Load question catalog if available + include_questions = os.getenv("IS_QUESTIONS_NEED_CSV", "0") == "1" + question_catalog = {} + if include_questions: + try: + question_catalog = load_question_catalog(config_path() / "questions.txt") + except Exception as e: + logger.warning(f"Could not load question catalog: {e}") + + # Write Excel + workbook_path = results_dir / output_workbook_name + writer = ExcelWriter(question_catalog, include_questions) + excel_file = writer.write(records, workbook_path) + + # Post-process Excel to convert boolean review to 0/1 + from openpyxl import load_workbook + wb = load_workbook(excel_file) + responses_sheet = wb["Responses"] + + # Find the "Review" column (13th column, 0-indexed = 12) + review_col_idx = None + for col_idx, cell in enumerate(responses_sheet[1], 1): + if cell.value == "Review": + review_col_idx = col_idx + break + + if review_col_idx: + for row_idx in range(2, responses_sheet.max_row + 1): + cell = responses_sheet.cell(row_idx, review_col_idx) + if cell.value is True: + cell.value = 1 + elif cell.value is False: + cell.value = 0 + + wb.save(excel_file) + logger.info(f"Excel workbook generated: {excel_file}") + + # Print summary for user + print(f"\n{'='*70}") + print("CORRUPTED PDF DETAILS") + print(f"{'='*70}") + print(f"Total corrupted records: {len(corrupted_record_ids)}") + if corrupted_record_ids: + for record_id in sorted(corrupted_record_ids): + pdf_name = record_to_pdf.get(record_id) or f"unknown_{record_id}" + print(f" • {pdf_name} (ID: {record_id})") + print(f"{'='*70}\n") + + return excel_file + +if __name__ == "__main__": + if len(sys.argv) < 2: + print("Usage: python regenerate_excel.py [output_workbook_name]") + print("\nExample:") + print(" python regenerate_excel.py results/20260819_160942_gemini-3.5-flash/") + print(" python regenerate_excel.py results/20260819_160942_gemini-3.5-flash/ my_results.xlsx") + sys.exit(1) + + results_dir = sys.argv[1] + output_name = sys.argv[2] if len(sys.argv) > 2 else "questionnaire_results.xlsx" + + try: + # Use raw responses (more complete) instead of processing_state.json + excel_file = regenerate_excel_from_raw_responses(results_dir, output_name) + print(f"\n✓ Excel workbook created: {excel_file}") + except Exception as e: + print(f"✗ Error: {e}") + sys.exit(1) diff --git a/render_reference.py b/render_reference.py index 937d3fd8..6399e61e 100644 --- a/render_reference.py +++ b/render_reference.py @@ -1,5 +1,8 @@ from pathlib import Path -import fitz +import fitz +from core.logging.logger import get_logger + +logger = get_logger("render-reference") pdf = fitz.open("reference_master.pdf") output = Path("reference_pages") @@ -9,4 +12,4 @@ image = page.get_pixmap(dpi=200, alpha=False) image.save(output / f"page_{page_number}.png") -print(f"Saved {len(pdf)} reference pages to {output.resolve()}") \ No newline at end of file +logger.info("Saved %d reference pages to %s", len(pdf), output.resolve()) diff --git a/requirements.txt b/requirements.txt index 63d635a9..a0c57414 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,10 +3,12 @@ Pillow>=10.0 numpy>=1.24 openpyxl>=3.1 PyMuPDF>=1.23 -google-genai>=1.0 +httpx>=0.28 streamlit>=1.32 PyYAML>=6.0 python-dotenv>=1.0 +litellm>=1.60.0 RapidFuzz>=3.6 tqdm>=4.66 pytest>=8.0 +google-generativeai==0.8.6 \ No newline at end of file diff --git a/test_gemini.py b/test_gemini.py deleted file mode 100644 index cf2e4f3e..00000000 --- a/test_gemini.py +++ /dev/null @@ -1,21 +0,0 @@ -from dotenv import load_dotenv -import os -from google import genai - -load_dotenv() - -client = genai.Client(api_key=os.getenv("GEMINI_API_KEY")) - -response = client.models.generate_content( - model="models/gemini-3.5-flash", - contents="Reply with exactly the word OK" -) - -print("TEXT:") -print(repr(response.text)) - -print("\nFULL RESPONSE:") -try: - print(response.model_dump()) -except Exception: - print(response) \ No newline at end of file diff --git a/test_image.py b/test_image.py deleted file mode 100644 index 119d24ea..00000000 --- a/test_image.py +++ /dev/null @@ -1,25 +0,0 @@ -from dotenv import load_dotenv -import os -from google import genai - -load_dotenv() - -client = genai.Client( - api_key=os.getenv("GEMINI_API_KEY") -) - -with open("reference_pages/page_1.png","rb") as f: - img = f.read() - -response = client.models.generate_content( - model="models/gemini-3.5-flash", - contents=[ - "What is shown in this image?", - { - "mime_type":"image/png", - "data":img - } - ] -) - -print(response.text) \ No newline at end of file diff --git a/test_multimodal.py b/test_multimodal.py deleted file mode 100644 index 4e8b21cb..00000000 --- a/test_multimodal.py +++ /dev/null @@ -1,118 +0,0 @@ -""" -Live compatibility test for google-genai 2.x typed multimodal content. - -Run: - - python test_multimodal.py -""" - -from pathlib import Path -import os - -from dotenv import load_dotenv -from google import genai -from google.genai import types - - -# ---------------------------------------------------- -# Project paths -# ---------------------------------------------------- - -PROJECT_ROOT = Path(__file__).resolve().parent -ENV_FILE = PROJECT_ROOT / ".env" - -TEST_PNG = ( - PROJECT_ROOT - / "output" - / "crops" - / "135c97572a9bdfb6" - / "Q1.png" -) - -# ---------------------------------------------------- -# Load .env -# ---------------------------------------------------- - -print("=" * 60) -print("PROJECT ROOT :", PROJECT_ROOT) -print(".env PATH :", ENV_FILE) -print(".env EXISTS :", ENV_FILE.exists()) - -load_dotenv(ENV_FILE) - -API_KEY = os.getenv("GEMINI_API_KEY") - -print("API KEY LOADED :", bool(API_KEY)) -print("KEY LENGTH :", len(API_KEY) if API_KEY else 0) -print("=" * 60) - -if not API_KEY: - raise RuntimeError("GEMINI_API_KEY was not loaded.") - -# ---------------------------------------------------- -# Check image -# ---------------------------------------------------- - -if not TEST_PNG.exists(): - raise FileNotFoundError(f"Image not found:\n{TEST_PNG}") - -print("IMAGE :", TEST_PNG) -print("SIZE :", TEST_PNG.stat().st_size, "bytes") - -# ---------------------------------------------------- -# Build typed request -# ---------------------------------------------------- - -contents = [ - types.Content( - role="user", - parts=[ - types.Part.from_text( - text=""" -Describe this questionnaire crop. - -Return ONLY this JSON: - -{ - "status":"received" -} -""" - ), - types.Part.from_bytes( - data=TEST_PNG.read_bytes(), - mime_type="image/png", - ), - ], - ) -] - -# ---------------------------------------------------- -# Gemini -# ---------------------------------------------------- - -client = genai.Client(api_key=API_KEY) - -response = client.models.generate_content( - model="models/gemini-3.5-flash", - contents=contents, - config=types.GenerateContentConfig( - temperature=0, - response_mime_type="application/json", - ), -) - -print("\n") -print("=" * 60) -print("TEXT") -print("=" * 60) -print(response.text) - -print("\n") -print("=" * 60) -print("MODEL DUMP") -print("=" * 60) - -try: - print(response.model_dump()) -except Exception: - print(response) \ No newline at end of file diff --git a/tests/test_excel.py b/tests/test_excel.py index d51a3975..a16fe5c2 100644 --- a/tests/test_excel.py +++ b/tests/test_excel.py @@ -6,3 +6,15 @@ def test_writer_creates_required_sheets(tmp_path: Path): answers={f"Q{i}":Answer(f"Q{i}",answer_text="x",final_confidence=.9) for i in range(1,35)} path=ExcelWriter().write([Record("a","a.pdf",answers,.9,False)],tmp_path/"out.xlsx") assert set(load_workbook(path).sheetnames)=={"Responses","Review Queue","Audit","Summary","Processing Log"} + + +def test_writer_uses_question_catalog_in_response_header(tmp_path: Path): + path = ExcelWriter({"Q1": "How do you solve a problem?"}, include_questions=True).write([], tmp_path / "out.xlsx") + + assert load_workbook(path)["Responses"]["O1"].value == "Q1 — How do you solve a problem?" + + +def test_writer_omits_question_text_when_disabled(tmp_path: Path): + path = ExcelWriter({"Q1": "How do you solve a problem?"}, include_questions=False).write([], tmp_path / "out.xlsx") + + assert load_workbook(path)["Responses"]["O1"].value == "Q1" diff --git a/tests/test_litellm_provider.py b/tests/test_litellm_provider.py new file mode 100644 index 00000000..3aeb6ca9 --- /dev/null +++ b/tests/test_litellm_provider.py @@ -0,0 +1,40 @@ +import sys +from types import SimpleNamespace + +from core.ai.litellm_provider import LiteLLMProvider +from core.ai.response_parser import ResponseParser +from core.ai.retry import Retry +from core.models import Question + + +def test_litellm_model_names_are_provider_qualified(): + assert LiteLLMProvider._qualified_model("openrouter", "google/gemini-3.5-flash") == "openrouter/google/gemini-3.5-flash" + assert LiteLLMProvider._qualified_model("gemini", "gemini-2.0-flash") == "gemini/gemini-2.0-flash" + assert LiteLLMProvider._qualified_model("gemini", "gemini/gemini-2.0-flash") == "gemini/gemini-2.0-flash" + + +def test_litellm_sends_question_metadata_immediately_before_each_crop(monkeypatch, tmp_path): + captured = {} + + def completion(**kwargs): + captured.update(kwargs) + return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=''' + {"student":{"student_name":"","gender":"","school_name":"","school_udise":"","crc_name":"","crc_udise":"","block":"","district":"","grade":"","meena_manch_participation":""},"answers":[{"question_id":"Q1","selected_codes":["A"],"answer_text":"","confidence":0.9,"review_required":false,"raw_observations":"clear"}]} + '''))], usage=None) + + monkeypatch.setitem(sys.modules, "litellm", SimpleNamespace(completion=completion)) + student_page = tmp_path / "student.png" + question_crop = tmp_path / "Q1.png" + student_page.write_bytes(b"student") + question_crop.write_bytes(b"question") + provider = LiteLLMProvider("openrouter", "model", ResponseParser(), Retry(attempts=1), 30, 30000) + + provider.extract("extract", {"STUDENT_INFO": student_page, "Q1": question_crop}, [Question("Q1", "single_choice", 1, ["A", "B"])]) + + content = captured["messages"][0]["content"] + assert "question_id=Q1" in content[3]["text"] + assert "type=single_choice" in content[3]["text"] + assert 'allowed_codes=["A", "B"]' in content[3]["text"] + assert content[4]["type"] == "image_url" + assert captured["timeout"] == 30 + assert captured["max_tokens"] == 30000 diff --git a/tests/test_provider.py b/tests/test_provider.py deleted file mode 100644 index 7c8f7977..00000000 --- a/tests/test_provider.py +++ /dev/null @@ -1,41 +0,0 @@ -"""Live provider test using the exact production GeminiProvider request builder.""" -from __future__ import annotations - -import json -from pathlib import Path - -from core.ai.gemini_provider import GeminiProvider -from core.ai.response_parser import ResponseParser -from core.ai.retry import Retry -from core.models import Question - - -class CapturingParser(ResponseParser): - """Captures Gemini's raw response while preserving production parsing.""" - - raw_response: str = "" - - def parse(self, text: str): # type: ignore[override] - self.raw_response = text - return super().parse(text) - - -def main() -> None: - root = Path(__file__).parents[1] - crop = root / "output" / "crops" / "135c97572a9bdfb6" / "Q1.png" - parser = CapturingParser() - provider = GeminiProvider("gemini-3.5-flash", parser, Retry(attempts=1)) - question = Question("Q1", "single_choice", 3, ["A", "B", "C", "D"], True) - prompt = ( - "Return a JSON array with exactly one object containing question_id, " - "selected_codes, answer_text, confidence, review_required, and " - "raw_observations. The question_id is Q1 and allowed options are A, B, C, D." - ) - provider.extract(prompt, {"Q1": crop}, [question]) - parsed = json.loads(parser.raw_response) - assert isinstance(parsed, list) and parsed, "Gemini response is not a non-empty JSON array" - print(parser.raw_response) - - -if __name__ == "__main__": - main() diff --git a/tests/test_provider_selection.py b/tests/test_provider_selection.py new file mode 100644 index 00000000..ceb3b55b --- /dev/null +++ b/tests/test_provider_selection.py @@ -0,0 +1,10 @@ +from core.runtime_settings import RuntimeSettings + + +def test_runtime_settings_exposes_provider_and_models(): + settings = RuntimeSettings( + 8501, True, "gemini", "openrouter/model", "gemini-2.0-flash", 2, 200, 6, 12, + "../reference_master.pdf", .018, .35, .72, 3, 1.5, 120, 30000, "out.xlsx", "questions.txt", "state.json" + ) + assert settings.llm_provider == "gemini" + assert settings.gemini_model == "gemini-2.0-flash" diff --git a/tests/test_question_catalog.py b/tests/test_question_catalog.py new file mode 100644 index 00000000..f159a3c1 --- /dev/null +++ b/tests/test_question_catalog.py @@ -0,0 +1,26 @@ +import csv + +from core.questionnaire.question_catalog import load_question_catalog + + +def test_load_question_catalog_from_csv(tmp_path): + path = tmp_path / "questions.csv" + with path.open("w", encoding="utf-8", newline="") as output: + writer = csv.DictWriter(output, fieldnames=["question_id", "question"]) + writer.writeheader() + writer.writerow({"question_id": "Q1", "question": "How do you solve a problem?"}) + + assert load_question_catalog(path) == {"Q1": "How do you solve a problem?"} + + +def test_load_question_catalog_from_numbered_text_file(tmp_path): + path = tmp_path / "questions.txt" + path.write_text( + "1.\nHow do you solve a problem?\n\n2.\nWrite an answer.\nखुला प्रश्न — लिखित उत्तर।\n", + encoding="utf-8", + ) + + assert load_question_catalog(path) == { + "Q1": "How do you solve a problem?", + "Q2": "Write an answer.", + } diff --git a/tests/test_response_parser.py b/tests/test_response_parser.py index 69c6e2d0..e1ca0d34 100644 --- a/tests/test_response_parser.py +++ b/tests/test_response_parser.py @@ -1,6 +1,23 @@ +import pytest + from core.ai.response_parser import ResponseParser -def test_parser_rejects_extra_fields(): - raw='[{"question_id":"Q1","selected_codes":[],"answer_text":"","confidence":0.5,"review_required":false,"raw_observations":"x","unexpected":1}]' - try: ResponseParser().parse(raw) - except ValueError: return - assert False, "Unexpected response field must be rejected" +from core.models import Question + + +def test_parser_ignores_extra_fields_with_warning(): + raw='{"student":{"student_name":"","gender":"","school_name":"","school_udise":"","crc_name":"","crc_udise":"","block":"","district":"","grade":"","meena_manch_participation":""},"answers":[{"question_id":"Q1","selected_codes":[],"answer_text":"","confidence":0.5,"review_required":false,"raw_observations":"x","unexpected":1}]}' + with pytest.warns(RuntimeWarning, match="Ignoring unexpected answer fields"): + _, answers = ResponseParser().parse(raw) + assert answers[0].question_id == "Q1" + + +def test_parser_rejects_unknown_question_ids_and_option_codes(): + raw='{"student":{"student_name":"","gender":"","school_name":"","school_udise":"","crc_name":"","crc_udise":"","block":"","district":"","grade":"","meena_manch_participation":""},"answers":[{"question_id":"Q2","selected_codes":["Z"],"answer_text":"","confidence":0.5,"review_required":false,"raw_observations":"x"}]}' + questions = [Question("Q1", "single_choice", 1, ["A", "B"])] + + with pytest.raises(ValueError, match="unknown question_id"): + ResponseParser().parse(raw, questions) + + invalid_code = raw.replace('"Q2","selected_codes":["Z"]', '"Q1","selected_codes":["Z"]') + with pytest.raises(ValueError, match="invalid codes"): + ResponseParser().parse(invalid_code, questions) diff --git a/tests/test_runtime_settings.py b/tests/test_runtime_settings.py new file mode 100644 index 00000000..b0b53aeb --- /dev/null +++ b/tests/test_runtime_settings.py @@ -0,0 +1,68 @@ +import pytest + +from core.runtime_settings import load_runtime_settings + + +def environment() -> dict[str, str]: + return { + "LLM_PROVIDER": "openrouter", "OPENROUTER_API_KEY": "test-key", + "OPENROUTER_MODEL": "google/gemini-3.5-flash", "GEMINI_MODEL": "gemini-2.0-flash", + "PORT": "8501", + "IS_QUESTIONS_NEED_CSV": "Yes", + "BIHAR_WORKERS": "2", "BIHAR_RENDER_DPI": "200", + "BIHAR_SOURCE_SPREAD_COUNT": "6", "BIHAR_LOGICAL_PAGE_COUNT": "12", + "BIHAR_REFERENCE_MASTER_PDF": "../reference_master.pdf", + "BIHAR_SPREAD_GUTTER_FRACTION": "0.018", "BIHAR_MIN_IMAGE_QUALITY": "0.35", + "BIHAR_REVIEW_CONFIDENCE_THRESHOLD": "0.72", "BIHAR_MAX_RETRIES": "3", + "BIHAR_RETRY_BACKOFF_SECONDS": "1.5", "BIHAR_REQUEST_TIMEOUT_SECONDS": "120", + "BIHAR_MAX_COMPLETION_TOKENS": "30000", + "BIHAR_OUTPUT_WORKBOOK": "results.xlsx", + "BIHAR_QUESTION_CATALOG": "questions.txt", "BIHAR_STATE_FILE": "state.json", + } + + +def test_loads_typed_settings_from_environment(): + settings = load_runtime_settings(environment()) + + assert settings.workers == 2 + assert settings.retry_backoff_seconds == 1.5 + assert settings.request_timeout_seconds == 120 + assert settings.output_workbook == "results.xlsx" + assert settings.include_questions_in_reports is True + + +def test_missing_setting_is_reported(): + values = environment() + del values["BIHAR_RENDER_DPI"] + + with pytest.raises(RuntimeError, match="BIHAR_RENDER_DPI"): + load_runtime_settings(values) + + +def test_provider_specific_credentials_are_required(): + values = environment() + del values["OPENROUTER_API_KEY"] + + with pytest.raises(RuntimeError, match="OPENROUTER_API_KEY"): + load_runtime_settings(values) + + +def test_gemini_requires_gemini_credentials(): + values = environment() + values["LLM_PROVIDER"] = "gemini" + del values["OPENROUTER_API_KEY"] + values.pop("GEMINI_API_KEY", None) + values.pop("GOOGLE_API_KEY", None) + + with pytest.raises(RuntimeError, match="GEMINI_API_KEY"): + load_runtime_settings(values) + + +def test_question_header_setting_accepts_no_and_rejects_invalid_values(): + values = environment() + values["IS_QUESTIONS_NEED_CSV"] = "No" + assert load_runtime_settings(values).include_questions_in_reports is False + + values["IS_QUESTIONS_NEED_CSV"] = "sometimes" + with pytest.raises(ValueError, match="IS_QUESTIONS_NEED_CSV"): + load_runtime_settings(values)