| """ |
| 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""" |
| |
| |
| MAX_FILE_SIZE = 10 * 1024 * 1024 |
| |
| |
| 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: |
| 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: |
| 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) |
| """ |
| |
| if len(file_data) > self.MAX_FILE_SIZE: |
| return False, f"File too large: {len(file_data)} bytes (max {self.MAX_FILE_SIZE} bytes)" |
| |
| |
| if len(file_data) == 0: |
| return False, "File is empty" |
| |
| |
| if not mime_type: |
| mime_type, _ = mimetypes.guess_type(filename) |
| |
| |
| if mime_type not in self.SUPPORTED_TYPES: |
| supported = ', '.join(self.SUPPORTED_TYPES.values()) |
| return False, f"Unsupported file type: {mime_type}. Supported: {supported}" |
| |
| |
| 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) |
| |
| |
| 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: |
| text = file_data.decode('utf-8') |
| except UnicodeDecodeError: |
| |
| 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 |
| } |
| """ |
| |
| is_valid, error_msg = self.validate_file(file_data, filename, mime_type) |
| |
| if not is_valid: |
| return { |
| "success": False, |
| "content": "", |
| "metadata": {}, |
| "error": error_msg |
| } |
| |
| |
| if not mime_type: |
| mime_type, _ = mimetypes.guess_type(filename) |
| |
| file_type = self.SUPPORTED_TYPES[mime_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) |
| } |
|
|
| |
| document_parser = DocumentParser() |
|
|
