import os import json import time from hashlib import sha256 from typing import Any, Dict, List, Tuple import gradio as gr from modules.ocr import extract_text from modules.llm_provider import extract_form_data, ask_llm from modules.overview import generate_overview from modules.checklist import generate_checklist from modules.eligibility import check_eligibility from modules.timeline import extract_timeline from modules.risk_detector import generate_risks from modules.verifier import verify_documents, verification_summary def _read_css() -> str: css_content = "" css_path = os.path.join("assets", "style.css") if os.path.exists(css_path): with open(css_path, "r", encoding="utf-8") as f: css_content = f.read() return css_content def _file_fingerprint(file_obj) -> str: """Lightweight key for caching: file name + size + mtime if available.""" if file_obj is None: return "" name = getattr(file_obj, "name", "") or "" size = getattr(file_obj, "size", None) mtime = getattr(file_obj, "last_modified", None) raw = f"{name}|{size}|{mtime}".encode("utf-8", errors="ignore") return sha256(raw).hexdigest() # In-memory cache (server-side) _CACHE: Dict[str, Dict[str, Any]] = {} def analyze_document(file): """Analyze uploaded form and return all outputs for the dashboard.""" if file is None: return ( {}, gr.update(value="ℹ Upload a form first.", visible=True), gr.update(visible=False), gr.update(value="ℹ Analyze a form to see results.", visible=True), gr.update(visible=False), gr.update(value="ℹ No timeline information found in this form.", visible=True), gr.update(visible=False), "ℹ No risks detected.", gr.update(value="ℹ Upload supporting documents to verify.", visible=True), gr.update(visible=False), gr.update(value="", visible=False), gr.update(value="ℹ Ready", visible=False), gr.update(value=[], visible=False), ) key = _file_fingerprint(file) cached = _CACHE.get(key) if cached: master_json = cached["master_json"] overview = cached["overview"] checklist = cached["checklist"] timeline = cached["timeline"] risks = cached["risks"] status = f"✅ Loaded from cache ({time.strftime('%H:%M:%S')})" else: extracted_text = extract_text(file.name) master_json = extract_form_data(extracted_text) overview = generate_overview(master_json) checklist = generate_checklist(master_json) timeline = extract_timeline(extracted_text) risks = generate_risks(master_json) status = f"✅ Analysis complete ({time.strftime('%H:%M:%S')})" _CACHE[key] = { "master_json": master_json, "overview": overview, "checklist": checklist, "timeline": timeline, "risks": risks, } # Checklist empty state if not checklist: checklist_empty_update = gr.update(value="ℹ Analyze a form to see results.", visible=True) checklist_output_update = gr.update(visible=False) else: checklist_empty_update = gr.update(visible=False) checklist_output_update = gr.update(value=checklist, visible=True) # Timeline empty state if not timeline: timeline_empty_update = gr.update(value="ℹ No timeline information found in this form.", visible=True) timeline_output_update = gr.update(visible=False) else: timeline_empty_update = gr.update(visible=False) timeline_output_update = gr.update(value=timeline, visible=True) # Risks display if not risks or (isinstance(risks, str) and risks.strip() == "✅ No major issues detected."): risks_display = "ℹ No risks detected." else: risks_display = risks # Reset verification outputs when a new form is analyzed verification_empty_update = gr.update(value="ℹ Upload supporting documents to verify.", visible=True) verification_output_update = gr.update(visible=False) verification_summary_update = gr.update(value="", visible=False) # Overview empty state if not overview: overview_empty_update = gr.update(value="ℹ Upload a form first.", visible=True) overview_output_update = gr.update(visible=False) else: overview_empty_update = gr.update(value="", visible=False) overview_output_update = gr.update(value=overview, visible=True) completeness = 0 if master_json: completeness += 25 if checklist: completeness += 25 if timeline: completeness += 25 if risks_display and "No risks" not in str(risks_display): completeness += 25 return ( master_json, overview_empty_update, overview_output_update, checklist_empty_update, checklist_output_update, timeline_empty_update, timeline_output_update, risks_display, verification_empty_update, verification_output_update, verification_summary_update, gr.update(value=f"✅ Ready · Completeness {completeness}%", visible=True), gr.update(value=[], visible=False), ) def run_eligibility_check(income, master_json): if not master_json: return "ℹ Upload a form first." return check_eligibility(master_json, income) def run_verification(files, master_json): # Gradio may return a File object or list depending on browser upload. # Normalize to list for reliable truthiness checks. if not master_json: return ( gr.update(value="ℹ Upload supporting documents to verify.", visible=True), gr.update(visible=False), gr.update(value="", visible=False), ) if files is None: files_list = [] elif isinstance(files, list): files_list = files else: files_list = [files] if len(files_list) == 0: return ( gr.update(value="ℹ Upload supporting documents to verify.", visible=True), gr.update(visible=False), gr.update(value="", visible=False), ) if not files: return ( gr.update(value="ℹ Upload supporting documents to verify.", visible=True), gr.update(visible=False), gr.update(value="", visible=False), ) results = verify_documents(files_list, master_json) summary = verification_summary(results) return ( gr.update(visible=False), gr.update(value=results, visible=True), gr.update(value=summary, visible=True), ) def run_qa(question: str, master_json: dict, chat_history): """Chat handler for gr.Chatbot. Gradio 6.x expects Chatbot value to be either: - list[dict{"role": "user"|"assistant", "content": "..."}] - or list[gr.ChatMessage] We normalize to dict format to avoid "messages format" errors. """ if not master_json: assistant = "ℹ Upload a form first." else: prompt = json.dumps({"question": question, "master_json": master_json}) assistant = ask_llm(prompt) # Normalize incoming history history = chat_history if isinstance(chat_history, list) else [] def _to_msg(msg): if isinstance(msg, dict) and "role" in msg and "content" in msg: return msg # Some older formats may come as (user, assistant) tuples. if isinstance(msg, (list, tuple)) and len(msg) == 2: return {"role": "user", "content": str(msg[0])} return None norm_history: List[Dict[str, str]] = [] for m in history: if isinstance(m, dict) and "role" in m and "content" in m: norm_history.append({"role": m["role"], "content": m["content"]}) # Append new turn norm_history.append({"role": "user", "content": str(question)}) norm_history.append({"role": "assistant", "content": str(assistant)}) # Second output clears the textbox. return norm_history, "" with gr.Blocks(title="PaperPilot") as demo: master_json_state = gr.State() with gr.Group(elem_id="appbar"): gr.HTML( """
""" ) with gr.Tabs(elem_id="pp-tabs"): with gr.Tab("1) Analyze"): with gr.Row(equal_height=True): with gr.Column(scale=1): with gr.Group(elem_id="card-upload", elem_classes=["pp-card"]): gr.HTML( """