""" SAAP Document Parser Service Extracts text content from PDF, DOCX, and TXT files for AI agent processing Supported formats: - PDF (.pdf) - Microsoft Word (.docx, .doc) - Plain Text (.txt) """ import io import logging import mimetypes from typing import Dict, Optional, Tuple from pathlib import Path logger = logging.getLogger(__name__) class DocumentParser: """Parse documents and extract text content""" # Maximum file size: 10 MB MAX_FILE_SIZE = 10 * 1024 * 1024 # Supported MIME types SUPPORTED_TYPES = { 'application/pdf': 'pdf', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document': 'docx', 'application/msword': 'doc', 'text/plain': 'txt', } def __init__(self): """Initialize document parser with optional dependencies""" self.pdf_available = False self.docx_available = False # Try to import PDF parser try: import PyPDF2 self.PyPDF2 = PyPDF2 self.pdf_available = True logger.info("✅ PyPDF2 available for PDF parsing") except ImportError: logger.warning("⚠️ PyPDF2 not available - PDF parsing disabled") # Try to import DOCX parser try: import docx self.docx = docx self.docx_available = True logger.info("✅ python-docx available for DOCX parsing") except ImportError: logger.warning("⚠️ python-docx not available - DOCX parsing disabled") def validate_file(self, file_data: bytes, filename: str, mime_type: Optional[str] = None) -> Tuple[bool, str]: """ Validate uploaded file Returns: (is_valid, error_message) """ # Check file size if len(file_data) > self.MAX_FILE_SIZE: return False, f"File too large: {len(file_data)} bytes (max {self.MAX_FILE_SIZE} bytes)" # Check if empty if len(file_data) == 0: return False, "File is empty" # Detect MIME type if not provided if not mime_type: mime_type, _ = mimetypes.guess_type(filename) # Check if supported if mime_type not in self.SUPPORTED_TYPES: supported = ', '.join(self.SUPPORTED_TYPES.values()) return False, f"Unsupported file type: {mime_type}. Supported: {supported}" # Check parser availability file_type = self.SUPPORTED_TYPES[mime_type] if file_type == 'pdf' and not self.pdf_available: return False, "PDF parsing not available (PyPDF2 not installed)" if file_type in ['docx', 'doc'] and not self.docx_available: return False, "DOCX parsing not available (python-docx not installed)" return True, "" def parse_pdf(self, file_data: bytes) -> str: """Extract text from PDF file""" if not self.pdf_available: raise ValueError("PDF parsing not available") try: pdf_file = io.BytesIO(file_data) pdf_reader = self.PyPDF2.PdfReader(pdf_file) text_parts = [] for page_num in range(len(pdf_reader.pages)): page = pdf_reader.pages[page_num] text = page.extract_text() if text.strip(): text_parts.append(f"--- Page {page_num + 1} ---\n{text}") full_text = "\n\n".join(text_parts) logger.info(f"✅ Extracted {len(full_text)} characters from PDF ({len(pdf_reader.pages)} pages)") return full_text except Exception as e: logger.error(f"❌ PDF parsing error: {e}") raise ValueError(f"Failed to parse PDF: {str(e)}") def parse_docx(self, file_data: bytes) -> str: """Extract text from DOCX file""" if not self.docx_available: raise ValueError("DOCX parsing not available") try: docx_file = io.BytesIO(file_data) doc = self.docx.Document(docx_file) text_parts = [] for para in doc.paragraphs: if para.text.strip(): text_parts.append(para.text) # Extract text from tables for table in doc.tables: for row in table.rows: row_text = " | ".join(cell.text.strip() for cell in row.cells if cell.text.strip()) if row_text: text_parts.append(row_text) full_text = "\n".join(text_parts) logger.info(f"✅ Extracted {len(full_text)} characters from DOCX ({len(doc.paragraphs)} paragraphs)") return full_text except Exception as e: logger.error(f"❌ DOCX parsing error: {e}") raise ValueError(f"Failed to parse DOCX: {str(e)}") def parse_txt(self, file_data: bytes) -> str: """Extract text from TXT file""" try: # Try UTF-8 first try: text = file_data.decode('utf-8') except UnicodeDecodeError: # Fallback to latin-1 text = file_data.decode('latin-1') logger.info(f"✅ Extracted {len(text)} characters from TXT file") return text except Exception as e: logger.error(f"❌ TXT parsing error: {e}") raise ValueError(f"Failed to parse TXT: {str(e)}") def parse_document(self, file_data: bytes, filename: str, mime_type: Optional[str] = None) -> Dict: """ Parse document and extract text content Args: file_data: Raw file bytes filename: Original filename mime_type: Optional MIME type Returns: { "success": bool, "content": str, # Extracted text "metadata": { "filename": str, "file_type": str, "file_size": int, "char_count": int }, "error": str # Only if success=False } """ # Validate file is_valid, error_msg = self.validate_file(file_data, filename, mime_type) if not is_valid: return { "success": False, "content": "", "metadata": {}, "error": error_msg } # Detect MIME type if not mime_type: mime_type, _ = mimetypes.guess_type(filename) file_type = self.SUPPORTED_TYPES[mime_type] # Parse based on file type try: if file_type == 'pdf': content = self.parse_pdf(file_data) elif file_type in ['docx', 'doc']: content = self.parse_docx(file_data) elif file_type == 'txt': content = self.parse_txt(file_data) else: return { "success": False, "content": "", "metadata": {}, "error": f"Unsupported file type: {file_type}" } return { "success": True, "content": content, "metadata": { "filename": filename, "file_type": file_type, "file_size": len(file_data), "char_count": len(content) }, "error": "" } except ValueError as e: return { "success": False, "content": "", "metadata": { "filename": filename, "file_type": file_type, "file_size": len(file_data) }, "error": str(e) } # Global instance document_parser = DocumentParser()