from __future__ import annotations import asyncio import hashlib import json import logging import os import re import secrets import uuid import threading import webbrowser from dataclasses import dataclass, field as dataclass_field from datetime import datetime, timedelta from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional, Set from fastapi import BackgroundTasks, FastAPI, File, HTTPException, UploadFile, Request, Form from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel, Field from app.pipeline import DocumentProcessingPipeline try: from pypdf import PdfReader except Exception: # pragma: no cover PdfReader = None try: import pytesseract from PIL import Image OCR_AVAILABLE = True except ImportError: OCR_AVAILABLE = False try: import docx DOCX_AVAILABLE = True except ImportError: DOCX_AVAILABLE = False logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) BASE_DIR = Path(__file__).resolve().parent UPLOAD_DIR = BASE_DIR / "uploads" UPLOAD_DIR.mkdir(exist_ok=True) MAX_UPLOAD_SIZE_BYTES = 50 * 1024 * 1024 # 50MB MAX_BATCH_SIZE = 10 ALLOWED_EXTENSIONS = { ".txt", ".md", ".pdf", ".csv", ".json", ".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tiff", ".rtf", ".docx", ".xlsx", ".xls", ".pptx", ".html", ".xml", ".yaml", ".yml", ".log", ".py", ".js", ".ts", ".java", ".c", ".cpp", ".h" } # Supported MIME types for better file detection MIME_TYPE_MAP = { ".pdf": "application/pdf", ".txt": "text/plain", ".md": "text/markdown", ".csv": "text/csv", ".json": "application/json", ".xml": "application/xml", ".html": "text/html", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp", ".tiff": "image/tiff", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", ".rtf": "application/rtf", } APP_VERSION = "4.0.2" class ExtractionRequest(BaseModel): text: str custom_fields: Optional[Dict[str, Any]] = None class BatchRequest(BaseModel): documents: List[Dict[str, Any]] = Field(default_factory=list) class ChatMessage(BaseModel): message: str conversation_id: Optional[str] = None context: Optional[Dict[str, Any]] = None class LoginRequest(BaseModel): email: str password: str class RegisterRequest(BaseModel): email: str password: str name: str class BusinessChallenge(BaseModel): title: str description: str category: str priority: str = "medium" class ProcessingJobStatus(str, Enum): PENDING = "pending" UPLOADING = "uploading" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" @dataclass class ProcessingJob: job_id: str status: str progress: int = 0 filename: str = "" file_type: str = "" file_size: int = 0 created_at: str = "" updated_at: str = "" completed_at: Optional[str] = None result: Optional[Dict[str, Any]] = None error: Optional[str] = None summary: str = "" extracted_fields: List[Dict] = dataclass_field(default_factory=list) suggestions: List[str] = dataclass_field(default_factory=list) document_type: str = "unknown" @dataclass class User: user_id: str email: str name: str password_hash: str created_at: str last_login: Optional[str] = None role: str = "user" is_active: bool = True avatar: Optional[str] = None @dataclass class Conversation: conversation_id: str user_id: str title: str created_at: str messages: List[Dict] = dataclass_field(default_factory=list) # Simple in-memory user database (for demo purposes) users_db: Dict[str, User] = {} sessions_db: Dict[str, Dict[str, Any]] = {} # session_token -> {user_id, expires_at} conversations_db: Dict[str, Conversation] = {} class JobTracker: """Tracks all processing jobs with statistics""" def __init__(self): self.jobs: Dict[str, ProcessingJob] = {} self.stats = { "total_processed": 0, "total_failed": 0, "total_active": 0, "total_bytes_processed": 0, "documents_by_type": {}, "daily_stats": {} } def add_job(self, job: ProcessingJob): self.jobs[job.job_id] = job self.stats["total_active"] += 1 def update_job(self, job_id: str, **kwargs): if job_id in self.jobs: job = self.jobs[job_id] for key, value in kwargs.items(): if hasattr(job, key): setattr(job, key, value) job.updated_at = _utc_now() def complete_job(self, job_id: str, result: Dict[str, Any], summary: str = "", extracted_fields: List[Dict] = None, suggestions: List[str] = None): if job_id in self.jobs: job = self.jobs[job_id] job.status = "completed" job.progress = 100 job.completed_at = _utc_now() job.result = result job.summary = summary job.extracted_fields = extracted_fields or [] job.suggestions = suggestions or [] job.updated_at = _utc_now() self.stats["total_processed"] += 1 self.stats["total_active"] = max(0, self.stats["total_active"] - 1) self.stats["total_bytes_processed"] += job.file_size # Track by document type doc_type = job.document_type or "unknown" self.stats["documents_by_type"][doc_type] = self.stats["documents_by_type"].get(doc_type, 0) + 1 # Track daily stats today = datetime.utcnow().strftime("%Y-%m-%d") if today not in self.stats["daily_stats"]: self.stats["daily_stats"][today] = {"processed": 0, "failed": 0} self.stats["daily_stats"][today]["processed"] += 1 def fail_job(self, job_id: str, error: str): if job_id in self.jobs: job = self.jobs[job_id] job.status = "failed" job.error = error job.completed_at = _utc_now() job.updated_at = _utc_now() job.progress = 100 self.stats["total_failed"] += 1 self.stats["total_active"] = max(0, self.stats["total_active"] - 1) today = datetime.utcnow().strftime("%Y-%m-%d") if today not in self.stats["daily_stats"]: self.stats["daily_stats"][today] = {"processed": 0, "failed": 0} self.stats["daily_stats"][today]["failed"] += 1 def get_statistics(self) -> Dict[str, Any]: return { **self.stats, "total_jobs": len(self.jobs), "success_rate": (self.stats["total_processed"] / max(1, self.stats["total_processed"] + self.stats["total_failed"])) * 100 } job_tracker = JobTracker() app = FastAPI(title="DocIntel Studio Pro", version=APP_VERSION) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) pipeline = DocumentProcessingPipeline(enable_ocr=True, enable_table_parsing=True) def _utc_now() -> str: return datetime.utcnow().isoformat() + "Z" def _hash_password(password: str) -> str: """Hash password with salt""" salt = secrets.token_hex(16) hashed = hashlib.sha256(f"{salt}{password}".encode()).hexdigest() return f"{salt}${hashed}" def _verify_password(password: str, hashed: str) -> bool: """Verify password against hash""" if "$" not in hashed: return False salt, expected_hash = hashed.split("$", 1) actual_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest() return secrets.compare_digest(actual_hash, expected_hash) def _generate_session_token() -> str: return secrets.token_urlsafe(32) def _get_current_user(request: Request) -> Optional[User]: """Get current user from session cookie""" token = request.cookies.get("session_token") if not token or token not in sessions_db: return None session = sessions_db[token] if datetime.fromisoformat(session["expires_at"]) < datetime.utcnow(): del sessions_db[token] return None return users_db.get(session["user_id"]) def _safe_filename(filename: str) -> str: cleaned = re.sub(r"[^a-zA-Z0-9._-]", "_", filename or "file") cleaned = cleaned.strip("._") or "file" return cleaned[:180] def _is_allowed_file(filename: str) -> bool: return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS def _get_file_type(filename: str) -> str: """Determine file type from extension""" ext = Path(filename).suffix.lower() type_map = { ".pdf": "pdf", ".txt": "text", ".md": "markdown", ".csv": "spreadsheet", ".json": "data", ".xml": "data", ".yaml": "data", ".yml": "data", ".jpg": "image", ".jpeg": "image", ".png": "image", ".gif": "image", ".webp": "image", ".bmp": "image", ".tiff": "image", ".docx": "document", ".xlsx": "spreadsheet", ".xls": "spreadsheet", ".pptx": "presentation", ".rtf": "document", ".html": "web", ".log": "log", } return type_map.get(ext, "unknown") def _read_text_file(file_path: Path) -> str: for encoding in ("utf-8", "utf-16", "latin-1", "cp1252"): try: return file_path.read_text(encoding=encoding) except Exception: pass return file_path.read_text(encoding="utf-8", errors="ignore") def _read_pdf_text(file_path: Path) -> str: if PdfReader is None: raise RuntimeError("PDF support unavailable. Install pypdf package.") reader = PdfReader(str(file_path)) text_parts = [] for i, page in enumerate(reader.pages): page_text = page.extract_text() if page_text: text_parts.append(f"--- Page {i + 1} ---\n{page_text}") return "\n".join(text_parts).strip() def _read_docx_text(file_path: Path) -> str: if not DOCX_AVAILABLE: raise RuntimeError("DOCX support unavailable. Install python-docx package.") doc = docx.Document(str(file_path)) paragraphs = [p.text for p in doc.paragraphs if p.text.strip()] return "\n\n".join(paragraphs) def _read_image_text(file_path: Path) -> str: """Extract text from image using OCR""" if not OCR_AVAILABLE: raise RuntimeError("OCR support unavailable. Install pytesseract and Pillow packages.") try: image = Image.open(file_path) text = pytesseract.image_to_string(image) return text.strip() except Exception as e: raise RuntimeError(f"OCR failed: {str(e)}") def _build_generic_document_text(file_path: Path, raw_bytes: bytes) -> str: decoded_preview = "" for encoding in ("utf-8", "utf-16", "latin-1"): try: decoded_preview = raw_bytes.decode(encoding, errors="ignore").strip() if decoded_preview: break except Exception: continue preview = decoded_preview[:3000] if decoded_preview else "" metadata_lines = [ f"Filename: {file_path.name}", f"Extension: {file_path.suffix.lower() or 'none'}", f"Size bytes: {len(raw_bytes)}", f"Type: {file_path.suffix.lower() or 'unclassified'} upload", ] if preview: metadata_lines.extend(["Decoded preview:", preview]) else: metadata_lines.append("Decoded preview: not available") return "\n".join(metadata_lines) def _generate_summary(text: str, doc_type: str, extraction_result: Any) -> str: """Generate a concise summary of the document""" summaries = [] # Document type summary type_descriptions = { "invoice": "invoice/financial document", "receipt": "receipt/transaction record", "contract": "legal contract/agreement", "report": "analytical report", "letter": "business correspondence", "form": "structured form", "email": "email communication", "article": "article/publication", "technical": "technical documentation", } doc_desc = type_descriptions.get(doc_type, "general document") summaries.append(f"This is a {doc_desc}.") # Extract key information if extraction_result and hasattr(extraction_result, 'extracted_fields'): fields = extraction_result.extracted_fields if fields: key_fields = [f.name for f in fields[:5]] summaries.append(f"Key fields identified: {', '.join(key_fields)}.") # Length assessment word_count = len(text.split()) if word_count < 100: summaries.append("Brief document.") elif word_count < 500: summaries.append("Medium-length document.") else: summaries.append(f"Lengthy document with approximately {word_count} words.") return " ".join(summaries) def _generate_suggestions(doc_type: str, extracted_fields: List[Dict], text: str) -> List[str]: """Generate actionable suggestions based on document analysis""" suggestions = [] # Type-specific suggestions if doc_type == "invoice": suggestions.append("Consider setting up automated payment reminders based on due dates.") suggestions.append("Cross-reference vendor information with your approved supplier list.") suggestions.append("Set up expense categorization rules for this vendor.") elif doc_type == "contract": suggestions.append("Review termination clauses and notice periods.") suggestions.append("Set calendar reminders for renewal dates.") suggestions.append("Flag any auto-renewal clauses for legal review.") elif doc_type == "receipt": suggestions.append("Categorize for tax purposes and expense tracking.") suggestions.append("Check if this qualifies for any rebate programs.") elif doc_type == "report": suggestions.append("Consider creating visual dashboards for key metrics.") suggestions.append("Set up automated report generation for regular intervals.") elif doc_type == "form": suggestions.append("Verify all required fields are completed.") suggestions.append("Consider digitizing this form for easier processing.") # Generic suggestions if len(text.split()) > 1000: suggestions.append("This is a lengthy document - consider using AI summarization for quick review.") if not extracted_fields: suggestions.append("Few fields were extracted - consider manual review or custom extraction rules.") suggestions.append("Save this document to your workspace for future reference.") suggestions.append("Export results to integrate with your existing workflows.") return suggestions def _ai_chat_response(message: str, context: Optional[Dict] = None) -> Dict[str, Any]: """ Intelligent, multi-variant conversational engine for document intelligence assistance. Each call produces a different response to the same intent through rotation-based variant selection. """ import hashlib context = context or {} message_lower = message.lower().strip() recent_messages = context.get("recent_messages") or [] current_section = context.get("current_section") or "overview" topic_hint = context.get("topic") or "" message_words = set(w for w in message_lower.split() if len(w) > 2) # Track response count per session key to rotate variants response_counter = getattr(_ai_chat_response, "_counter", 0) _ai_chat_response._counter = response_counter + 1 def _is_followup() -> bool: """Detect if this is a follow-up question based on conversation history.""" if not recent_messages: return False user_msgs = [m.get("content", "") for m in recent_messages if m.get("role") == "user"] assistant_msgs = [m.get("content", "") for m in recent_messages if m.get("role") == "assistant"] return len(user_msgs) > 1 and len(assistant_msgs) > 0 def _last_topic() -> str: """Get the topic of the last conversation turn.""" if not recent_messages: return "" user_msgs = [m.get("content", "") for m in recent_messages if m.get("role") == "user"] if len(user_msgs) >= 2: return user_msgs[-2][:60] return "" def _word_overlap(words1, words2): set1 = set(w.lower() for w in words1) set2 = set(w.lower() for w in words2) if not set1 or not set2: return 0.0 return len(set1 & set2) / len(set1 | set2) def _best_match(intent_map): best_intent = None best_score = 0.0 for name, data in intent_map.items(): score = _word_overlap(message_words, data.get("keywords", [])) if any(kw in message_lower for kw in data.get("keywords", [])): score += 0.3 if any(p in message_lower for p in data.get("phrases", [])): score += 0.4 if score > best_score: best_score = score best_intent = name return best_intent, best_score def _pick_variant(variants): """Pick a variant deterministically based on message + counter to get diversity.""" idx = (response_counter + len(message_lower)) % len(variants) return variants[idx] def _build_context(): if not recent_messages: return "" user_msgs = [m.get("content", "") for m in recent_messages if m.get("role") == "user"] if not user_msgs: return "" latest = user_msgs[-1] if latest and latest.lower() != message_lower: return f"\n\n[Context: Your previous question was: '{latest[:100]}'. I am incorporating that context into this answer.]" return "" history_context = _build_context() is_followup = _is_followup() prev_topic = _last_topic() intents = { "extraction": { "keywords": ["extract", "extraction", "fields", "parse", "data", "structured", "values", "information", "pull", "retrieve"], "phrases": ["pull data", "data from", "get data", "find data", "show fields", "extract text", "extract data", "gather data"], "variants": [ "Document Extraction Overview\n\nTo extract data from a document:\n1. Go to Studio and paste your text or upload a file\n2. The system classifies the document (invoice, contract, report, etc.)\n3. Key fields are extracted with confidence percentages\n\nAutomatic extraction targets: email addresses, phone numbers, currency amounts, dates, names, organizations, and URLs.\n\nFor best results: use clean, well-formatted text. If extraction misses something, try the Summarize feature which uses a different approach. Batch mode works well for groups of similar documents.\n\n{history_context}", "Extracting Data from Documents\n\nThe extraction pipeline works in three stages:\n\nStage 1: Classification -- the AI identifies what type of document you have.\nStage 2: Field Detection -- common patterns like dates, amounts, and names are located.\nStage 3: Confidence Scoring -- each extracted value gets a reliability score.\n\nYou can define custom extraction fields by providing regex patterns. Go to Studio, enter your text, and click Extract Data to see it in action. If you need help with custom patterns, describe what you're looking for and I can suggest a regex.\n\n{history_context}", "Working with Extracted Data\n\nAfter extraction, you can:\n- View all fields with confidence scores\n- Copy individual values or the full result\n- Download results as a text file\n- Get suggestions for next steps\n\nCommon fields extracted by document type:\n- Invoices: invoice number, date, total, vendor, due date\n- Contracts: parties, effective date, terms, clauses\n- Receipts: store, items, total, payment method\n- Reports: title, author, date, key metrics\n\nIf a field was missed, try adding more context to your text or use the Batch mode for comparing multiple documents.\n\n{history_context}", "Advanced Extraction Tips\n\nTo maximize extraction accuracy:\n1. Remove unnecessary formatting from pasted text\n2. Ensure dates use a standard format (MM/DD/YYYY or YYYY-MM-DD)\n3. Currency symbols ($, EUR, GBP) help the system identify amounts\n4. Full names and complete addresses improve entity recognition\n\nIf you're working with a specific document type repeatedly, you can define custom extraction fields. For example, for invoices you might add fields like 'purchase_order_number' or 'shipping_address'.\n\n{history_context}" ], "suggestions_variants": [ ["Extract this invoice for me", "Create a custom extraction field", "How accurate is the extraction?", "What fields does it find?"], ["Show me an extraction example", "Write a custom regex pattern", "Compare two documents", "Extract data from a PDF"], ["Why did extraction miss this value?", "How do I improve accuracy?", "Extract all dates and amounts", "Show me confidence scores"] ] }, "summarization": { "keywords": ["summary", "summarize", "brief", "summarise", "key points", "overview", "condense", "digest", "highlights", "tl;dr", "short", "recap"], "phrases": ["give me summary", "make summary", "short version", "key takeaways", "tell me briefly", "what is this about", "main points"], "variants": [ "Document Summarization\n\nThe system can generate summaries in multiple formats:\n- Executive Summary: 2-3 sentence high-level overview\n- Bullet Points: key facts in scannable format\n- Detailed Digest: comprehensive breakdown\n- Action Items: decisions, tasks, and next steps\n\nTo use: go to Studio > Summarize tab, paste your text, and click Generate Summary. The system identifies document type, extracts key sentences (beginning, middle, end), finds all numbers and dates, and provides actionable suggestions.\n\n{history_context}", "How Summarization Works\n\nThe summarization engine analyzes your document by:\n1. Classifying the document type (invoice, report, email, etc.)\n2. Extracting key sentences (first, middle, and last sections)\n3. Scanning for numbers, dates, and currency amounts\n4. Generating confidence-weighted suggestions\n\nFor long documents, the summary focuses on the most information-dense sections. You can customize the output by specifying what you want highlighted (financial data, dates, names, etc.).\n\n{history_context}", "Getting the Best Summary\n\nTo get the most useful summary:\n- Paste the full document text, not just excerpts\n- The system works best with 500+ words of content\n- Structured text (with headings and sections) produces clearer summaries\n- Financial and business documents get the richest analysis\n\nAfter generating a summary, you can copy it, download it, or use it as a starting point for extraction. The suggestions section often contains valuable next steps.\n\n{history_context}", "Summary Output Fields\n\nEach summary includes:\n- Document type classification\n- Word and sentence counts\n- Key points from beginning, middle, and end\n- Numbers found (up to 10)\n- Dates found (up to 10)\n- Confidence score for the classification\n- Actionable suggestions\n\nThis gives you a complete picture of the document's content without reading it entirely.\n\n{history_context}" ], "suggestions_variants": [ ["Summarize this invoice", "Generate bullet points", "Find all dates and numbers", "Create an executive summary"], ["Summarize a contract", "What are the key decisions?", "Highlight financial data", "Compare two summaries"], ["Extract key action items", "Make a detailed digest", "Short version of this text", "What is this document about?"] ] }, "ocr": { "keywords": ["ocr", "image", "scan", "handwritten", "photo", "picture", "scanner", "tesseract", "optical", "recognition", "text from image"], "phrases": ["read text from image", "convert image to text", "image to text", "scan document", "handwriting recognition", "text recognition", "extract text from picture"], "variants": [ "OCR Processing Guide\n\nOCR extracts text from images and scanned documents. Supported formats: JPG, PNG, GIF, WebP, BMP, TIFF.\n\nFor best accuracy:\n- Use 300+ DPI resolution\n- Ensure good lighting and contrast\n- Avoid shadows and glare\n- Keep documents straight and unrotated\n- Printed text works best (handwriting has lower accuracy)\n\nIf OCR quality is poor: crop the image to focus on text areas, increase contrast before uploading, or convert to black and white.\n\n{history_context}", "Image Text Extraction\n\nThe OCR pipeline includes:\n1. Preprocessing: noise reduction, contrast enhancement, binarization\n2. Text detection: locating text regions in the image\n3. Recognition: converting text regions to machine-readable text\n4. Post-processing: confidence filtering and formatting\n\nSystem limitations: handwriting recognition is experimental, very small fonts may be missed, and complex layouts (multi-column) may not preserve order perfectly.\n\n{history_context}", "Improving OCR Results\n\nCommon OCR problems and solutions:\n\nBlurry image -> increase resolution or use a better camera\nLow contrast -> adjust brightness/contrast before uploading\nSkewed text -> straighten the image first\nBackground noise -> use a plain white background\nSmall fonts -> zoom in before capturing\n\nFor critical documents, always review OCR output against the original image.\n\n{history_context}", "OCR vs Manual Input\n\nWhen OCR is not working well, consider:\n- Using the Text Input tab and typing the content manually\n- Extracting text from a PDF using a dedicated PDF tool first\n- Taking a screenshot at higher resolution\n- Using a dedicated scanning app before uploading\n\nFor forms and structured documents, the system can still extract useful metadata even if full OCR is imperfect.\n\n{history_context}" ], "suggestions_variants": [ ["How to improve OCR accuracy?", "Can it read handwriting?", "What image formats work?", "OCR failed, what now?"], ["Best settings for scanning", "Extract text from this photo", "OCR vs PDF extraction", "Preprocess an image for OCR"], ["Why is OCR giving errors?", "Supported image formats", "Handwriting recognition tips", "OCR confidence scores"] ] }, "formats": { "keywords": ["pdf", "docx", "xlsx", "csv", "json", "xml", "format", "file type", "extension", "supported", "convert", "upload", "import"], "phrases": ["what formats", "file support", "what files", "supported formats", "file types", "document types", "max file size", "file size limit", "which formats"], "variants": [ "Supported File Formats\n\nDocIntel Pro supports {count}+ file formats across these categories:\n- Documents: PDF, DOCX, RTF, TXT, MD, HTML\n- Spreadsheets: XLSX, XLS, CSV\n- Images (via OCR): JPG, PNG, GIF, WebP, BMP, TIFF\n- Data: JSON, XML, YAML, LOG\n- Code: PY, JS, TS, JAVA, C, CPP, H\n\nUpload limits: max 50MB per file, max 10 documents per batch.\n\nAll files are processed securely and can be auto-deleted after processing.\n\n{history_context}", "Format-Specific Processing\n\nEach format gets a tailored processing path:\n- PDF: text extraction per page, handles both digital and scanned\n- DOCX: paragraph-by-paragraph extraction with formatting\n- Images: OCR pipeline with preprocessing\n- CSV/JSON: structured data detection\n- Code files: syntax-aware extraction\n\nPro tip: for scanned PDFs, the system automatically applies OCR. For Excel files, table structures are preserved when possible.\n\n{history_context}", "File Format Limitations\n\nWhat is NOT supported:\n- Password-protected or encrypted files\n- Corrupted or incomplete files\n- Very large files (over 50MB)\n- Audio or video files\n- Proprietary database formats\n\nIf you have one of these, try converting to a supported format first. For example, export a database report as CSV, or save a password-protected PDF without the password.\n\n{history_context}", "Choosing the Right Format\n\nFor the best results:\n- Text files (TXT, MD): ideal for direct extraction\n- PDF: best for formatted documents and scanned pages\n- DOCX: good for Word documents with complex formatting\n- Images: use when only a photo or scan is available\n- JSON/CSV: perfect for structured data and batch processing\n\nWhen in doubt, paste the text directly using the Text Input tab, which bypasses file format limitations.\n\n{history_context}" ], "suggestions_variants": [ ["Can you process password-protected PDFs?", "What is the max file size?", "How do I upload multiple files?", "What Excel formats work?"], ["Convert a PDF to text", "Best format for invoices", "Upload a scanned document", "Process a JSON file"], ["File format limitations", "Why is my file not uploading?", "Supported image formats", "Batch process Excel files"] ] }, "batch": { "keywords": ["batch", "multiple", "bulk", "many", "volume", "mass", "queue", "group", "several", "together", "simultaneous"], "phrases": ["many documents", "many files", "process all", "all at once", "at the same time", "one go", "multiple documents", "multiple files", "process many", "bulk upload"], "variants": [ "Batch Processing\n\nBatch mode processes up to 10 documents simultaneously:\n1. Go to Studio > Batch tab\n2. Enter text for each document\n3. Separate documents with --- on a new line\n4. Click Process Batch\n\nEach document is analyzed independently. Results include per-document statistics and an overall success rate. Throughput: up to 100 documents per minute.\n\n{history_context}", "Using Batch Mode Effectively\n\nBest use cases:\n- Processing a queue of similar invoices\n- Analyzing multiple contracts at once\n- Batch importing data from spreadsheets\n- Bulk classification of document types\n\nThe --- separator is critical: each section becomes a separate document. For very similar documents, batch mode gives consistent extraction across all items.\n\n{history_context}", "Batch Processing Limits\n\nBatch constraints:\n- Maximum 10 documents per batch request\n- Each document processed independently\n- Results show individual and aggregate statistics\n- All documents must be text (file uploads processed individually)\n\nFor larger volumes, consider making multiple batch requests or using the API for programmatic access.\n\n{history_context}", "Batch Results Analysis\n\nAfter batch processing, you can:\n- See success/failure counts per document\n- View aggregate statistics\n- Compare extraction results across documents\n- Identify patterns in document types\n\nThis is particularly useful for quality assurance across a document set.\n\n{history_context}" ], "suggestions_variants": [ ["How do I process 100+ documents?", "Can I automate batch uploads via API?", "How do I compare batch results?", "Show me a batch demo"], ["Batch process invoices", "Batch classify documents", "Batch vs single processing", "Troubleshoot batch errors"], ["Maximum batch size", "Automate batch with scripts", "Compare extraction results", "Batch success rates"] ] }, "api": { "keywords": ["api", "endpoint", "integrate", "integration", "code", "request", "json", "rest", "curl", "programmatic", "sdk", "library", "develop", "webhook"], "phrases": ["how to call", "show me code", "example request", "api key", "authentication", "program access", "curl example", "python example", "javascript example", "api documentation"], "variants": [ "REST API Integration\n\nBase URL: http://localhost:8000\n\nKey Endpoints:\n- POST /extract: extract data from text\n- POST /upload: upload and process a file\n- POST /batch: process multiple documents\n- POST /summarize: generate a summary\n- POST /chat: ask the AI assistant\n- GET /jobs/{id}: check job status\n- GET /stats: system statistics\n\nExample (curl):\ncurl -X POST http://localhost:8000/extract -H \"Content-Type: application/json\" -d '{\"text\": \"Invoice #123 for $500\"}'\n\nFull OpenAPI docs available at /docs.\n\n{history_context}", "API Code Examples\n\nPython upload example:\nimport requests\nfiles = {'file': open('invoice.pdf', 'rb')}\nr = requests.post('http://localhost:8000/upload', files=files)\nprint(r.json())\n\nPython extract example:\nimport requests\nr = requests.post('http://localhost:8000/extract', json={'text': 'Invoice #123 for $500'})\ndata = r.json()\nprint(f\"Type: {data['classification']['document_type']}\")\n\nJavaScript fetch example:\nconst res = await fetch('/extract', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({text: 'Invoice #123'})});\nconst data = await res.json();\n\n{history_context}", "API Features\n\n- CORS enabled for cross-origin requests\n- All endpoints return JSON\n- Async processing for large files\n- Background jobs with status polling\n- Comprehensive error messages\n- Full OpenAPI/Swagger documentation\n\nThe API supports all the same features as the web interface, making it suitable for custom integrations and automation.\n\n{history_context}", "API Best Practices\n\n1. For large files, use /upload which returns a job_id for async tracking\n2. For quick text, use /extract which returns results synchronously\n3. Poll /jobs/{id} every 2 seconds to check status\n4. Use /batch for processing up to 10 documents in one request\n5. Check /stats for system health and capacity\n\nRate limiting is not currently enforced, but please be considerate with concurrent requests.\n\n{history_context}" ], "suggestions_variants": [ ["Show me a Python upload example", "How do I call /summarize?", "What does the response JSON look like?", "Show me the API documentation"], ["JavaScript fetch example", "Automate batch via API", "API authentication", "Error handling in API"], ["Upload file via curl", "Extract data via API", "Check job status via API", "API rate limits"] ] }, "troubleshooting": { "keywords": ["error", "fail", "problem", "issue", "broken", "not working", "bug", "glitch", "wrong", "incorrect", "doesn't work", "failed", "crash", "stuck", "timeout"], "phrases": ["not working", "something wrong", "help fix", "fix this", "error message", "throws error", "getting error", "what went wrong", "why did it fail", "broken feature"], "variants": [ "Troubleshooting Guide\n\nCommon issues and solutions:\n- File too large? Max is 50MB\n- Wrong format? Check supported formats list\n- Empty file? Ensure the file has content\n- No fields extracted? Try the Summarize feature\n- Low confidence? Clean up the text formatting\n- OCR blank? Use a higher resolution image\n- Job stuck? Check the Jobs tab for status, or restart the server\n\nFor fastest resolution, share the exact error message and what you were doing when it occurred.\n\n{history_context}", "Diagnosing Problems\n\nBreak down the issue by stage:\n1. Upload stage: did the file upload successfully? Check file size and format.\n2. Processing stage: is the job progressing? Check Jobs tab.\n3. Extraction stage: are fields being found? Check confidence scores.\n4. Output stage: is the result displaying? Try copying the result.\n\nMost issues are resolved by ensuring clean input and correct format.\n\n{history_context}", "Common Error Messages\n\n'File too large': reduce file size or compress the document\n'Unsupported format': convert to a supported format first\n'Empty file': verify the file has content before uploading\n'OCR failed': the image may be too low quality, try a clearer version\n'Processing timeout': the document may be too long, try splitting it\n'Job not found': the job may have expired, try uploading again\n\nIf you continue to have issues, describe the exact error text for a targeted fix.\n\n{history_context}", "System Health Checks\n\nYou can check system status:\n- Visit /health endpoint for system status\n- Check /stats for system statistics\n- Review the server console for error logs\n- Restart with python main.py if unresponsive\n\nFor persistent issues, check that all dependencies are installed: pip install -r requirements.txt\n\n{history_context}" ], "suggestions_variants": [ ["My upload failed with an error", "Summarize returns blank result", "OCR is giving garbage text", "Job is stuck processing"], ["File size error", "Why is extraction empty?", "Server not responding", "Fix OCR quality"], ["Error handling best practices", "Debug processing pipeline", "API error responses", "Check system health"] ] }, "privacy": { "keywords": ["privacy", "security", "secure", "safe", "gdpr", "retain", "retention", "encrypt", "encryption", "data protection", "confidential", "sensitive", "delete", "compliance"], "phrases": ["how long kept", "data stored", "my data", "your data", "data usage", "data sharing", "delete my", "remove my", "data retention"], "variants": [ "Privacy and Data Protection\n\nAll documents are processed with:\n- TLS 1.3 encryption in transit\n- AES-256 encryption at rest\n- Isolated processing environments\n- Configurable auto-deletion (1 hour, 24 hours, immediately)\n\nDefault retention: 24 hours for processing and quality assurance. Manual deletion is available at any time.\n\nCompliance: GDPR and CCPA compliant. Your data is never used for training without explicit opt-in.\n\n{history_context}", "Data Security Measures\n\nSecurity layers:\n- Network: all traffic encrypted with TLS 1.3\n- Storage: documents encrypted with AES-256\n- Processing: each document in an isolated sandbox\n- Access: configurable deletion policies\n- Audit: full activity logging available\n\nYou have the right to access, rectify, and delete your data at any time.\n\n{history_context}", "Data Retention and Deletion\n\nRetention options:\n- Default: 24 hours\n- 1 hour: for sensitive documents\n- Immediately: delete right after processing\n- Never: keep indefinitely\n\nTo delete: documents are automatically removed based on your retention setting. Manual deletion is available through the interface.\n\n{history_context}", "Compliance Information\n\nThis system is designed for:\n- GDPR compliance (data access, deletion, portability)\n- CCPA compliance (opt-out, deletion rights)\n- SOC2-type processing controls\n- Enterprise security requirements\n\nFor detailed compliance questions, refer to the Privacy Policy in the footer.\n\n{history_context}" ], "suggestions_variants": [ ["How long are files kept?", "Can I delete uploads automatically?", "Is my data used for training?", "Show me the privacy policy"], ["Data encryption details", "GDPR compliance features", "Auto-delete settings", "Manual document deletion"], ["Security best practices", "Compliance checklist", "Data retention policy", "Third-party data access"] ] }, "pricing": { "keywords": ["pricing", "cost", "free", "plan", "enterprise", "billing", "subscription", "tier", "premium", "paid", "license", "price", "money"], "phrases": ["how much", "what included", "free tier", "enterprise tier", "self hosted", "self-host", "cloud version", "on premise", "on-premise", "pricing plan"], "variants": [ "Pricing and Plans\n\nFree Tier (current): includes all core features, up to 10 documents per batch, standard processing speed, and community support.\n\nEnterprise (available): unlimited batch size, priority processing, custom extraction schemas, dedicated support and SLA, on-premise deployment.\n\nDeployment options: local Python, Docker, cloud (AWS/Azure/GCP), or Kubernetes.\n\n{history_context}", "Deployment Options\n\n1. Local: run with python main.py, requires Python 3.8+\n2. Docker: single container with docker-compose up\n3. Cloud: deploy to AWS, Azure, or Google Cloud\n4. Kubernetes: production-scale deployment\n\nThe application is fully open and ready to deploy. All dependencies are in requirements.txt.\n\n{history_context}", "Free vs Enterprise\n\nFree tier includes:\n- All core document processing features\n- Standard processing speed\n- Community support\n- Up to 10 documents per batch\n\nEnterprise adds:\n- Unlimited batch sizes\n- Priority processing queue\n- Custom extraction schemas\n- Dedicated support with SLA\n- On-premise deployment option\n- Custom integration assistance\n\n{history_context}", "Self-Hosting Guide\n\nRequirements:\n- Python 3.8 or higher\n- pip packages from requirements.txt\n- Optional: Tesseract OCR for image processing\n\nSteps:\n1. pip install -r requirements.txt\n2. python main.py\n3. Open http://localhost:8000\n\nDocker:\n1. docker-compose up\n2. Access the dashboard at port 8000\n\n{history_context}" ], "suggestions_variants": [ ["What is included in the free tier?", "How does enterprise deployment work?", "Can I self-host this?", "Enterprise features overview"], ["Docker deployment guide", "Cloud deployment options", "Kubernetes setup", "Pricing comparison"], ["Self-hosting requirements", "Enterprise support options", "Custom extraction schemas", "Deployment costs"] ] }, "account": { "keywords": ["login", "logout", "sign in", "sign out", "account", "profile", "register", "password", "email", "auth", "authentication", "user", "session"], "phrases": ["create account", "new account", "forgot password", "change password", "my profile", "edit profile", "delete account", "sign up", "my account"], "variants": [ "Account Management\n\nAuthentication: click Login in the navigation bar. Use any email and password to create a demo account. Sessions persist for 30 days.\n\nProfile settings (Workspace > Profile): update your name, email, and role. Save preferences for theme, auto-refresh, and auto-delete.\n\nNotifications are managed in Settings and appear in the notification center.\n\n{history_context}", "Profile and Preferences\n\nIn Workspace, you can customize:\n- Profile: name, email, role\n- Preferences: theme (light, dark, warm), auto-refresh, auto-delete\n\nSettings page offers additional controls:\n- Refresh interval\n- Notification toggle\n- System information\n\nAll preferences are saved locally and persist between sessions.\n\n{history_context}", "Session and Login Information\n\n- Sessions last 30 days by default\n- Login is optional; you can use the app without an account\n- Documents and history are stored locally when not logged in\n- Login enables cross-session persistence\n\nLogout: click your name in the nav bar and confirm sign out.\n\n{history_context}", "Managing Your Workspace\n\nThe Workspace section has four tabs:\n- Profile: personal information and role\n- My Documents: files you have uploaded\n- History: processing records\n- Export: download data as JSON, CSV, or report\n\nLogin is recommended to save your workspace data across sessions.\n\n{history_context}" ], "suggestions_variants": [ ["How do I change my profile name?", "Why are notifications not appearing?", "How do I sign out safely?", "What settings can I customize?"], ["Create an account", "Update my preferences", "Export my workspace data", "Delete my account"], ["Session timeout", "Login issues", "Profile settings guide", "Notification management"] ] }, "capabilities": { "keywords": ["hello", "hi", "hey", "help", "what can you do", "capabilities", "features", "about", "introduction", "how does this work", "purpose", "functions", "what is"], "phrases": ["what can you do", "how do i use", "tell me about", "get started", "getting started", "capabilities", "features overview", "show features", "what is this", "introduce yourself"], "variants": [ "DocIntel Pro Capabilities\n\nThis is an AI-powered document intelligence platform. Key capabilities:\n- Classify documents into 7+ types\n- Extract structured data with confidence scoring\n- Generate summaries with key points, numbers, and dates\n- Process images through OCR\n- Batch process multiple documents\n- Provide API for custom integrations\n- Solve business challenges\n\nTry asking: 'extract data from this invoice', 'how do I summarize a document?', 'show me the API endpoints', or 'help me fix an upload error'.\n\nCurrent section: {current_section.capitalize()}\n\n{history_context}", "Getting Started Guide\n\nThree ways to use the system:\n\n1. Upload a file: go to Studio > Upload, drag and drop or click to browse\n2. Paste text: go to Studio > Text Input, paste your document text\n3. Ask for help: use this AI assistant for guidance\n\nThe system will classify your document, extract key fields, and provide suggestions. Most documents process in under 1 second.\n\nTry the demo data to see it in action without preparing your own content.\n\n{history_context}", "Feature Overview\n\nCore features:\n- Smart Upload: 20+ file formats with drag and drop\n- AI Extraction: automatic classification and field extraction\n- Summarization: concise summaries with key insights\n- AI Chatbot: real-time assistance for any question\n- Business Solutions: tailored recommendations\n- Enterprise Security: encrypted processing and GDPR compliance\n\nEach feature is designed to save time and reduce manual document processing effort.\n\n{history_context}", "System Architecture\n\nThe platform processes documents through a pipeline:\n1. Input: text paste, file upload, or API request\n2. OCR: image preprocessing and text recognition (if needed)\n3. Classification: document type identification\n4. Extraction: field detection with confidence scoring\n5. Validation: data quality checks\n6. Output: structured results with suggestions\n\nThis pipeline processes documents in under 1 second each, with 92-98% accuracy.\n\n{history_context}" ], "suggestions_variants": [ ["What can this system do?", "How do I get started?", "Show me the features", "Help me understand the platform"], ["Extract data from an invoice", "How does summarization work?", "What formats are supported?", "Show me a demo"], ["I need help with batch processing", "How does the API work?", "Troubleshoot an error", "Business challenge solver"] ] } } # Section-specific context responses section_responses = { "studio": { "response": "You are in the Studio section where document processing happens.\n\nAvailable tabs:\n- Upload: drag and drop or browse for files (20+ formats supported)\n- Text Input: paste text for instant extraction and analysis\n- Batch: process up to 10 documents at once with the --- separator\n- Summarize: generate summaries with key points and insights\n- Jobs: track processing status for uploaded files\n\nStart by pasting text or uploading a file, then the system will classify, extract, and provide suggestions.\n\n{history_context}", "suggestions": ["Summarize the text I just pasted", "Extract fields from this document", "Show me how batch works", "Upload a file for me"] }, "workspace": { "response": "You are in the Workspace section for managing your profile and data.\n\nFour sections:\n- Profile: update your name, email, and role\n- My Documents: view your uploaded files\n- History: see processing records\n- Export: download data as JSON, CSV, or report\n\nLogin to persist your data across sessions, or continue as a guest for temporary use.\n\n{history_context}", "suggestions": ["How do I export my data?", "Show me my processing history", "How do I save my profile?", "What is auto-delete?"] }, "challenges": { "response": "You are in the Business Challenges section.\n\nDescribe a business problem and get tailored solutions with:\n- Problem analysis\n- Recommended solutions with specific steps\n- Estimated ROI for measuring impact\n- Next steps to implement\n\nCategories: Efficiency, Compliance, Cost Reduction, Scalability, Data Quality.\n\n{history_context}", "suggestions": ["Help me automate invoice processing", "How do I improve data quality?", "Reduce document processing costs", "Scale my document workflow"] }, "settings": { "response": "You are in the Settings section for application configuration.\n\nSettings available:\n- Theme: Light, Dark, or Warm mode\n- Auto-refresh interval: 2, 5, 10 seconds or disabled\n- Notifications: enable or disable\n\nAdditional preferences can be set in Workspace > Preferences.\n\n{history_context}", "suggestions": ["How do I change the theme?", "How do notifications work?", "What does auto-refresh do?", "Save my current settings"] }, "docs": { "response": "You are in the Documentation section.\n\nHere you can:\n- View the API base URL\n- Open the full OpenAPI/Swagger documentation at /docs\n- Load all API endpoints with descriptions\n\nThe API supports all the same features as the web interface.\n\n{history_context}", "suggestions": ["Show me the API endpoints", "How do I call the API?", "What is the base URL?", "OpenAPI documentation"] } } # Fallback responses with multiple variants fallback_variants = [ [ { "response": "I need a bit more information to give you the best answer.\n\nCould you specify:\n1. What type of document you are working with (invoice, contract, report, etc.)\n2. What you want to do (extract, summarize, troubleshoot, integrate)\n3. Any specific question or error you are encountering\n\nOnce you provide these details, I can give you a precise, actionable response.\n\n{history_context}", "suggestions": ["Extract data from a document", "Generate a summary", "Fix a processing error", "Show me the API"] }, { "response": "Here are the main features available to you right now:\n\n- Studio > Upload: drag and drop or click to upload any document\n- Studio > Text Input: paste text for instant extraction\n- Studio > Batch: process multiple documents at once\n- AI Assistant (this chat): ask me anything about document processing\n- Challenges: get tailored business solutions\n\nQuick start: go to Studio, paste some text, and click Extract Data. The AI will classify your document and extract key fields in seconds.\n\n{history_context}", "suggestions": ["Go to Studio", "How to upload a file", "What can I extract?", "Help me get started"] }, { "response": "Try one of these common workflows:\n\n1. Process an invoice: paste invoice text in Studio Text Input and click Extract Data\n2. Summarize a report: paste the text in Studio Summarize tab\n3. Upload a PDF: use Studio Upload tab and select your file\n4. Batch process: separate multiple documents with --- in the Batch tab\n5. Get help: ask me specific questions about any feature\n\nEach workflow takes under 30 seconds to complete.\n\n{history_context}", "suggestions": ["Process an invoice", "Summarize a report", "Upload a PDF", "Batch processing help"] } ] ] # Find best matching intent intent_name, confidence = _best_match(intents) if intent_name and confidence > 0.3: intent = intents[intent_name] variant = _pick_variant(intent["variants"]) suggestions = _pick_variant(intent["suggestions_variants"]) response = variant.format( history_context=history_context, current_section=current_section, topic_hint=topic_hint[:80] if topic_hint else "unknown", count=len(ALLOWED_EXTENSIONS) ) return {"response": response, "suggestions": suggestions} # Section-specific fallbacks if current_section in section_responses: sec = section_responses[current_section] return { "response": sec["response"].format(history_context=history_context), "suggestions": sec["suggestions"] } # Random fallback for unrecognized queries import random fallback_group = fallback_variants[0] fb = fallback_group[response_counter % len(fallback_group)] return { "response": fb["response"].format( history_context=history_context, current_section=current_section ), "suggestions": fb["suggestions"] } # ==================== API ENDPOINTS ==================== @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "timestamp": _utc_now(), "version": APP_VERSION, "pending_jobs": job_tracker.stats["total_active"], "max_upload_mb": round(MAX_UPLOAD_SIZE_BYTES / (1024 * 1024), 2), "features": { "ocr": OCR_AVAILABLE, "pdf": PdfReader is not None, "docx": DOCX_AVAILABLE, "supported_formats": list(ALLOWED_EXTENSIONS) } } @app.post("/register") async def register_user(request: RegisterRequest): """Register a new user""" # Check if email already exists for user in users_db.values(): if user.email == request.email: raise HTTPException(status_code=400, detail="Email already registered") user_id = str(uuid.uuid4()) user = User( user_id=user_id, email=request.email, name=request.name, password_hash=_hash_password(request.password), created_at=_utc_now() ) users_db[user_id] = user # Auto-login after registration token = _generate_session_token() sessions_db[token] = { "user_id": user_id, "expires_at": (datetime.utcnow() + timedelta(days=30)).isoformat() } return { "user_id": user_id, "email": user.email, "name": user.name, "token": token, "message": "Registration successful" } @app.post("/login") async def login_user(request: LoginRequest): """Login user""" for user in users_db.values(): if user.email == request.email: if _verify_password(request.password, user.password_hash): user.last_login = _utc_now() token = _generate_session_token() sessions_db[token] = { "user_id": user.user_id, "expires_at": (datetime.utcnow() + timedelta(days=30)).isoformat() } return { "user_id": user.user_id, "email": user.email, "name": user.name, "token": token, "message": "Login successful" } else: raise HTTPException(status_code=401, detail="Invalid password") # For demo purposes, allow login without registration # Create a demo user user_id = str(uuid.uuid4()) user = User( user_id=user_id, email=request.email, name="User", password_hash=_hash_password(request.password), created_at=_utc_now(), last_login=_utc_now() ) users_db[user_id] = user token = _generate_session_token() sessions_db[token] = { "user_id": user_id, "expires_at": (datetime.utcnow() + timedelta(days=30)).isoformat() } return { "user_id": user_id, "email": user.email, "name": user.name, "token": token, "message": "Demo account created" } @app.post("/logout") async def logout_user(request: Request): """Logout user""" token = request.cookies.get("session_token") if token and token in sessions_db: del sessions_db[token] return {"message": "Logged out successfully"} @app.get("/me") async def get_current_user_info(request: Request): """Get current user information""" user = _get_current_user(request) if not user: return {"authenticated": False} return { "authenticated": True, "user_id": user.user_id, "email": user.email, "name": user.name, "role": user.role, "created_at": user.created_at, "last_login": user.last_login } @app.post("/upload") async def upload_document( file: UploadFile = File(...), background_tasks: BackgroundTasks = None ): """ Upload and process a document. Supports PDF, images, text files, Word documents, and more. """ try: if not file.filename: raise HTTPException(status_code=400, detail="Filename is required") raw_bytes = await file.read() if not raw_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty") if len(raw_bytes) > MAX_UPLOAD_SIZE_BYTES: raise HTTPException( status_code=413, detail=f"File too large. Max size is {round(MAX_UPLOAD_SIZE_BYTES / (1024 * 1024), 2)} MB" ) # Generate unique ID job_id = str(uuid.uuid4()) safe_name = _safe_filename(file.filename) file_type = _get_file_type(file.filename) job = ProcessingJob( job_id=job_id, status="uploading", progress=10, filename=safe_name, file_type=file_type, file_size=len(raw_bytes), created_at=_utc_now(), updated_at=_utc_now(), document_type=file_type ) job_tracker.add_job(job) # Save uploaded file file_path = UPLOAD_DIR / f"{job_id}_{safe_name}" with open(file_path, "wb") as output: output.write(raw_bytes) # Process in background if background_tasks: background_tasks.add_task( _process_uploaded_file, job_id, str(file_path) ) else: await _process_uploaded_file(job_id, str(file_path)) return { "job_id": job_id, "filename": file.filename, "file_type": file_type, "status": "processing", "message": "Document uploaded and processing started" } except Exception as e: logger.error(f"Upload failed: {e}") raise HTTPException(status_code=400, detail=str(e)) @app.post("/extract") async def extract_data(request: ExtractionRequest): """ Extract structured data from document text. """ try: if not request.text: raise HTTPException(status_code=400, detail="Text content is required") # Process document doc_id = str(uuid.uuid4()) result = await pipeline.process_document( document_id=doc_id, text=request.text, custom_extraction_schema=request.custom_fields ) # Generate summary and suggestions doc_type = result.classification.document_type.value if result.classification else "general" summary = _generate_summary(request.text, doc_type, result.extraction) extracted_fields = [ {"name": f.name, "value": f.value, "confidence": f.confidence} for f in result.extraction.extracted_fields ] if result.extraction else [] suggestions = _generate_suggestions(doc_type, extracted_fields, request.text) result_dict = result.to_dict() result_dict["summary"] = summary result_dict["suggestions"] = suggestions result_dict["extracted_fields"] = extracted_fields return result_dict except Exception as e: logger.error(f"Extraction failed: {e}") raise HTTPException(status_code=400, detail=str(e)) @app.post("/batch") async def batch_process(request: BatchRequest): """ Process multiple documents in batch. """ try: if not request.documents: raise HTTPException(status_code=400, detail="Documents list is required") if len(request.documents) > MAX_BATCH_SIZE: raise HTTPException( status_code=400, detail=f"Batch too large. Max batch size is {MAX_BATCH_SIZE}" ) # Create job job_id = str(uuid.uuid4()) job = ProcessingJob( job_id=job_id, status="processing", progress=20, created_at=_utc_now(), updated_at=_utc_now() ) job_tracker.add_job(job) # Process documents results = await pipeline.process_batch(request.documents) # Calculate statistics stats = pipeline.get_statistics(results) # Update job job_tracker.complete_job( job_id, {"results": [r.to_dict() for r in results], "statistics": stats}, f"Batch processing completed: {len(results)} documents processed" ) return { "job_id": job_id, "results": [r.to_dict() for r in results], "statistics": stats } except Exception as e: logger.error(f"Batch processing failed: {e}") raise HTTPException(status_code=400, detail=str(e)) @app.get("/jobs/{job_id}") async def get_job_status(job_id: str): """Get status of a processing job""" if job_id not in job_tracker.jobs: raise HTTPException(status_code=404, detail="Job not found") job = job_tracker.jobs[job_id] return { "job_id": job.job_id, "status": job.status, "progress": job.progress, "filename": job.filename, "file_type": job.file_type, "file_size": job.file_size, "created_at": job.created_at, "updated_at": job.updated_at, "completed_at": job.completed_at, "result": job.result, "error": job.error, "summary": job.summary, "extracted_fields": job.extracted_fields, "suggestions": job.suggestions, "document_type": job.document_type } @app.get("/jobs") async def list_jobs(): """List all processing jobs""" sorted_jobs = sorted( job_tracker.jobs.values(), key=lambda j: j.created_at, reverse=True ) return { "total": len(job_tracker.jobs), "jobs": [ { "job_id": job.job_id, "status": job.status, "progress": job.progress, "filename": job.filename, "file_type": job.file_type, "file_size": job.file_size, "created_at": job.created_at, "completed_at": job.completed_at, "document_type": job.document_type, "summary": job.summary } for job in sorted_jobs ] } @app.get("/stats") async def get_statistics(): """Get overall system statistics""" return { **job_tracker.get_statistics(), "timestamp": _utc_now() } @app.post("/chat") async def chat_with_ai(request: ChatMessage): """Chat with AI assistant for document intelligence help""" response_data = _ai_chat_response(request.message, request.context) # Store conversation if conversation_id provided if request.conversation_id and request.conversation_id in conversations_db: conv = conversations_db[request.conversation_id] conv.messages.append({ "role": "user", "content": request.message, "timestamp": _utc_now() }) conv.messages.append({ "role": "assistant", "content": response_data["response"], "suggestions": response_data["suggestions"], "timestamp": _utc_now() }) return response_data @app.post("/summarize") async def summarize_document(request: ExtractionRequest): """Generate a summary of document text""" try: if not request.text: raise HTTPException(status_code=400, detail="Text content is required") # Process through pipeline for classification doc_id = str(uuid.uuid4()) result = await pipeline.process_document( document_id=doc_id, text=request.text ) doc_type = result.classification.document_type.value if result.classification else "general" # Generate comprehensive summary word_count = len(request.text.split()) sentences = request.text.split(".") # Extract key sentences (first, middle, last) key_sentences = [] if len(sentences) > 3: key_sentences = [sentences[0], sentences[len(sentences)//2], sentences[-1]] else: key_sentences = sentences[:3] # Extract numbers/amounts numbers = re.findall(r'[\$€£]?[\d,]+\.?\d*', request.text) # Extract dates dates = re.findall(r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b', request.text) summary = { "document_type": doc_type, "word_count": word_count, "sentence_count": len(sentences), "key_points": [s.strip() for s in key_sentences if s.strip()], "numbers_found": numbers[:10], "dates_found": dates[:10], "confidence": result.classification.confidence if result.classification else 0.5, "suggestions": _generate_suggestions(doc_type, [], request.text) } return summary except Exception as e: logger.error(f"Summarization failed: {e}") raise HTTPException(status_code=400, detail=str(e)) @app.post("/solve-challenge") async def solve_business_challenge(request: BusinessChallenge): """Analyze and suggest solutions for business challenges""" challenge_analyses = { "efficiency": { "analysis": "Efficiency challenges often stem from manual, repetitive tasks. Document intelligence can automate data extraction, reduce processing time by 80-90%, and eliminate human errors.", "solutions": [ "Implement automated document processing workflows", "Set up batch processing for high-volume documents", "Use OCR for digitizing paper-based processes", "Create custom extraction rules for domain-specific forms" ], "roi_estimate": "60-80% reduction in processing time, 90% reduction in data entry errors" }, "compliance": { "analysis": "Compliance challenges require accurate record-keeping, audit trails, and data validation. Our system provides structured extraction with confidence scores and validation checks.", "solutions": [ "Enable comprehensive audit logging for all document processing", "Set up validation rules to ensure data completeness", "Implement automated compliance report generation", "Use secure document storage with encryption" ], "roi_estimate": "Reduced compliance risk, faster audit preparation, automated reporting" }, "cost": { "analysis": "Cost challenges often relate to manual labor, errors, and inefficiency. Automation through document intelligence significantly reduces operational costs.", "solutions": [ "Replace manual data entry with automated extraction", "Reduce headcount needed for document processing", "Minimize costly errors through validation", "Scale processing without proportional cost increase" ], "roi_estimate": "50-70% reduction in document processing costs within 6 months" }, "scalability": { "analysis": "Scalability challenges occur when manual processes can't handle growth. Our API-first architecture scales automatically with your needs.", "solutions": [ "Use cloud-based processing for elastic scaling", "Implement batch processing for peak volumes", "Set up automated workflows with queue management", "Use API integration for seamless system connectivity" ], "roi_estimate": "Handle 10x volume with minimal infrastructure changes" }, "data_quality": { "analysis": "Data quality issues lead to poor decisions and rework. Our validation engine ensures extracted data meets quality standards.", "solutions": [ "Enable multi-stage validation with confidence scoring", "Set up automated data quality reports", "Implement human-in-the-loop review for low-confidence extractions", "Use machine learning to improve extraction accuracy over time" ], "roi_estimate": "95%+ data accuracy, 80% reduction in rework" } } category = request.category.lower() analysis = challenge_analyses.get(category, challenge_analyses["efficiency"]) return { "challenge": request.title, "description": request.description, "category": category, "priority": request.priority, "analysis": analysis["analysis"], "solutions": analysis["solutions"], "estimated_roi": analysis["roi_estimate"], "recommended_features": [ "Automated document processing", "Custom extraction schemas", "Batch processing capabilities", "API integration", "Validation and quality scoring" ], "next_steps": [ "Schedule a demo to see the system in action", "Start with a pilot project on a specific document type", "Define success metrics and KPIs", "Plan integration with existing systems" ] } @app.get("/documentation") async def get_documentation(): """Get API documentation""" return { "title": "DocIntel Studio Pro API", "version": APP_VERSION, "description": "AI-powered document intelligence platform", "endpoints": { "/health": "Health check and system status", "/register": "Register new user", "/login": "User authentication", "/logout": "User logout", "/me": "Get current user info", "/upload": "Upload and process document", "/extract": "Extract data from text", "/batch": "Process multiple documents", "/jobs": "List all jobs", "/jobs/{id}": "Get job status", "/stats": "Get system statistics", "/chat": "Chat with AI assistant", "/summarize": "Generate document summary", "/solve-challenge": "Get solutions for business challenges" }, "supported_formats": list(ALLOWED_EXTENSIONS), "features": { "ocr": OCR_AVAILABLE, "pdf_processing": PdfReader is not None, "docx_processing": DOCX_AVAILABLE, "batch_processing": True, "ai_chatbot": True, "user_authentication": True, "business_analytics": True } } # ==================== BACKGROUND PROCESSING ==================== async def _process_uploaded_file(job_id: str, file_path: str): """Process uploaded file in background""" try: job_tracker.update_job(job_id, status="processing", progress=40) path = Path(file_path) suffix = path.suffix.lower() raw_bytes = path.read_bytes() text = "" # Determine source type and extraction strategy if suffix in {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tiff'}: try: text = _read_image_text(path) except Exception as e: logger.warning(f"OCR failed for image: {e}") text = _build_generic_document_text(path, raw_bytes) elif suffix == '.pdf': text = _read_pdf_text(path) if not text: raise RuntimeError("PDF parsed but no readable text was found") elif suffix == '.docx': text = _read_docx_text(path) if not text.strip(): text = _build_generic_document_text(path, raw_bytes) else: try: text = _read_text_file(path) except Exception: text = "" if not text.strip(): text = _build_generic_document_text(path, raw_bytes) # Process through pipeline job_tracker.update_job(job_id, progress=60) result = await pipeline.process_document( document_id=job_id, text=text ) job_tracker.update_job(job_id, progress=80) # Generate summary and suggestions doc_type = result.classification.document_type.value if result.classification else "unknown" summary = _generate_summary(text, doc_type, result.extraction) extracted_fields = [ {"name": f.name, "value": f.value, "confidence": f.confidence} for f in result.extraction.extracted_fields ] if result.extraction else [] suggestions = _generate_suggestions(doc_type, extracted_fields, text) # Complete the job job_tracker.complete_job( job_id, result.to_dict(), summary=summary, extracted_fields=extracted_fields, suggestions=suggestions ) except Exception as e: logger.error(f"Background processing failed: {e}") job_tracker.fail_job(job_id, str(e)) # ==================== PRIVACY POLICY ==================== PRIVACY_POLICY_HTML = """

Privacy & Data Protection

Your data privacy is our priority

Data Collection

We collect and process documents you upload solely for the purpose of providing document intelligence services. We do not sell, share, or use your data for any other purpose.

Data Security

All documents are encrypted in transit (TLS 1.3) and at rest (AES-256). Processing occurs in isolated environments, and documents can be automatically deleted after processing.

Data Retention

By default, uploaded documents are retained for 24 hours for processing and quality assurance. You can configure automatic deletion immediately after processing.

Compliance

Our platform is designed to comply with GDPR, CCPA, and other data protection regulations. You have the right to access, rectify, and delete your data at any time.

AI Processing

We use AI models for document classification, text extraction, and data validation. Your data is never used to train our models without explicit opt-in consent.

""" # ==================== ENHANCED DASHBOARD HTML ==================== DASHBOARD_HTML = """ DocIntel Pro
Notifications
No notifications yet.
""" + PRIVACY_POLICY_HTML + """
Quick Actions:
S

Transform Documents into Intelligent Data

Upload any document format — PDF, images, Word, Excel — and let our AI extract, classify, and summarize key information. Perfect for invoices, contracts, reports, and any business document workflow.

Live Statistics

0
Documents
0
Processed
0
Failed
0
Active Jobs

Smart Upload

Support for 20+ file formats including PDF, images (JPG, PNG, GIF), Word, Excel, and text files. Drag & drop or click to upload.

AI Extraction

Intelligent classification and field extraction with confidence scoring. Automatically identifies document types and extracts key data.

Summarization

Get concise summaries of lengthy documents with key points, numbers, dates, and actionable suggestions highlighted.

AI Chatbot

Built-in AI assistant to help with document processing questions, API integration, troubleshooting, and best practices.

Business Solutions

Solve complex business challenges with tailored recommendations for efficiency, compliance, cost reduction, and scalability.

Enterprise Security

GDPR-compliant with encrypted processing, automatic data deletion options, and comprehensive audit logging.

User Profile

Preferences

My Documents

Documents you've uploaded will appear here. Login to persist your documents across sessions.

Processing History

No processing history yet. Upload a document to get started.

Export Data

Export your processing results and statistics in various formats.

Upload Document

FILE

Drop files here or click to browse

Supports: PDF, JPG, PNG, GIF, WebP, DOCX, XLSX, TXT, MD, CSV, JSON, XML, and more

Max size: 50MB

Supported Formats:

PDF JPG/PNG DOCX XLSX TXT CSV JSON

Result

Waiting for upload...

Analyze Text

Paste text from any document to extract structured data and get intelligent insights.

Extraction Result

Waiting for text input...

Batch Processing

Process up to 10 documents at once. Separate each document with "---" on a new line.

Batch Results

Waiting for batch processing...

Document Summarization

Generate concise summaries of lengthy documents with key points and insights.

Summary

Waiting for input...

Live Job Stream

No jobs yet Upload a document to start processing

AI Document Intelligence Assistant

Ask questions about document processing, API usage, troubleshooting, or best practices.

Hello! I'm your DocIntel AI assistant. I can help you with:
  • Document processing questions
  • API integration guidance
  • Troubleshooting errors
  • Best practices and tips
How do I get started? Supported formats? Extraction accuracy?

Solve Business Challenges

Describe your business challenge and get tailored solutions powered by document intelligence.

Common Challenges

1 Manual invoice data entry taking too long
2 Contract review creating bottlenecks
3 Unable to scale document processing
4 Poor data quality from extraction

API Documentation

Complete API reference for integrating DocIntel into your applications.

API Endpoints

Application Settings

System Information

Loading...

""" @app.get("/") async def root(): """Root endpoint - redirect to dashboard""" return HTMLResponse(content=DASHBOARD_HTML) @app.get("/dashboard") async def dashboard(): """Main dashboard""" return HTMLResponse(content=DASHBOARD_HTML) # ==================== START SERVER ==================== if __name__ == "__main__": import uvicorn def _open_browser() -> None: webbrowser.open_new_tab("http://127.0.0.1:8000/dashboard") threading.Timer(1.5, _open_browser).start() uvicorn.run(app, host="127.0.0.1", port=8000, reload=False)