""" File parsing utilities for documents (TXT, PDF, DOCX) Returns extracted UTF-8 text and basic metadata """ from __future__ import annotations import os from typing import Dict, Any, List def parse_text_bytes(content: bytes) -> str: """Decode bytes to text using utf-8 with fallback.""" try: return content.decode("utf-8") except Exception: try: return content.decode("latin-1", errors="ignore") except Exception: return content.decode(errors="ignore") def parse_pdf(path: str) -> str: if not _fitz_available: raise ImportError("PyMuPDF (fitz) is not installed. Install with: pip install PyMuPDF") doc = fitz.open(path) try: texts = [] for page in doc: texts.append(page.get_text()) return "\n".join(texts) finally: doc.close() def parse_docx(path: str) -> str: if not _docx_available: raise ImportError("python-docx is not installed. Install with: pip install python-docx") from docx import Document # type: ignore d = Document(path) return "\n".join(p.text for p in d.paragraphs) def extract_text_from_file(temp_path: str, original_filename: str | None, content: bytes) -> Dict[str, Any]: """ Extract text from a temp file based on extension. Supported: .txt, .pdf, .docx """ suffix = os.path.splitext(original_filename or "")[1].lower() if suffix in (".txt", ""): text = parse_text_bytes(content) elif suffix == ".pdf": text = parse_pdf(temp_path) elif suffix == ".docx": text = parse_docx(temp_path) else: raise ValueError(f"Unsupported document type: {suffix}") return { "text": text, "metadata": { "filename": original_filename, "extension": suffix, "size": len(content) } } def chunk_text(text: str, max_chars: int = 2000) -> List[str]: """Split text into chunks at whitespace boundaries up to max_chars.""" if not text: return [""] chunks: List[str] = [] start = 0 n = len(text) while start < n: end = min(start + max_chars, n) if end < n: # try to break on whitespace ws = text.rfind(" ", start, end) if ws != -1 and ws > start + max_chars * 0.6: end = ws chunks.append(text[start:end]) start = end return chunks import os import logging from typing import Optional, Tuple from pathlib import Path # Optional imports with graceful fallback try: import fitz # type: ignore # PyMuPDF _fitz_available = True except ImportError: _fitz_available = False fitz = None try: from docx import Document # type: ignore _docx_available = True except ImportError: _docx_available = False Document = None try: import chardet # type: ignore _chardet_available = True except ImportError: _chardet_available = False chardet = None from config.settings import settings logger = logging.getLogger(__name__) class FileParser: """Utility class for parsing various file formats""" def __init__(self): self.max_file_size = settings.max_file_size self.text_extensions = settings.allowed_text_extensions self.supported_encodings = ['utf-8', 'latin-1', 'ascii', 'cp1252'] def parse_file(self, file_path: str | Path) -> Tuple[str, dict]: """ Parse a file and extract text content Args: file_path: Path to the file to parse Returns: Tuple of (extracted_text, metadata) """ file_path = Path(file_path) if not file_path.exists(): raise FileNotFoundError(f"File not found: {file_path}") # Check file size file_size = file_path.stat().st_size if file_size > self.max_file_size: raise ValueError(f"File too large. Maximum size: {self.max_file_size} bytes") # Get file extension extension = file_path.suffix.lower() if extension == '.pdf': return self._parse_pdf(file_path) elif extension == '.docx': return self._parse_docx(file_path) elif extension == '.txt': return self._parse_txt(file_path) else: raise ValueError(f"Unsupported file type: {extension}") def _parse_pdf(self, file_path: Path) -> Tuple[str, dict]: """Extract text from PDF file using PyMuPDF""" if not _fitz_available: raise ImportError("PyMuPDF (fitz) is not installed. Install with: pip install PyMuPDF") try: text_content = [] metadata = { 'file_type': 'pdf', 'pages': 0, 'extracted_pages': [] } with fitz.open(file_path) as pdf: metadata['pages'] = len(pdf) metadata['title'] = pdf.metadata.get('title', '') metadata['author'] = pdf.metadata.get('author', '') for page_num, page in enumerate(pdf): page_text = page.get_text() if page_text.strip(): text_content.append(page_text) metadata['extracted_pages'].append(page_num + 1) full_text = '\n\n'.join(text_content) logger.info(f"Successfully parsed PDF: {file_path.name}, " f"extracted {len(metadata['extracted_pages'])} pages") return full_text, metadata except Exception as e: logger.error(f"Error parsing PDF file {file_path}: {e}") raise ValueError(f"Failed to parse PDF: {str(e)}") def _parse_docx(self, file_path: Path) -> Tuple[str, dict]: """Extract text from DOCX file""" if not _docx_available: raise ImportError("python-docx is not installed. Install with: pip install python-docx") try: doc = Document(file_path) text_content = [] metadata = { 'file_type': 'docx', 'paragraphs': 0, 'tables': 0 } # Extract text from paragraphs for para in doc.paragraphs: if para.text.strip(): text_content.append(para.text) metadata['paragraphs'] += 1 # Extract text from tables for table in doc.tables: metadata['tables'] += 1 for row in table.rows: row_text = [] for cell in row.cells: if cell.text.strip(): row_text.append(cell.text) if row_text: text_content.append('\t'.join(row_text)) # Extract document properties core_props = doc.core_properties metadata['title'] = core_props.title or '' metadata['author'] = core_props.author or '' metadata['created'] = str(core_props.created) if core_props.created else '' full_text = '\n\n'.join(text_content) logger.info(f"Successfully parsed DOCX: {file_path.name}, " f"extracted {metadata['paragraphs']} paragraphs") return full_text, metadata except Exception as e: logger.error(f"Error parsing DOCX file {file_path}: {e}") raise ValueError(f"Failed to parse DOCX: {str(e)}") def _parse_txt(self, file_path: Path) -> Tuple[str, dict]: """Extract text from TXT file with encoding detection""" try: # Detect encoding with open(file_path, 'rb') as file: raw_data = file.read() if _chardet_available: result = chardet.detect(raw_data) encoding = result['encoding'] or 'utf-8' confidence = result['confidence'] or 0.0 else: # Fallback without chardet encoding = 'utf-8' confidence = 0.0 # Try to decode with detected encoding try: text = raw_data.decode(encoding) except UnicodeDecodeError: # Fallback to common encodings for enc in self.supported_encodings: try: text = raw_data.decode(enc) encoding = enc break except UnicodeDecodeError: continue else: # If all fails, decode with errors='ignore' text = raw_data.decode('utf-8', errors='ignore') encoding = 'utf-8 (with errors ignored)' metadata = { 'file_type': 'txt', 'encoding': encoding, 'encoding_confidence': confidence, 'lines': len(text.splitlines()), 'characters': len(text) } logger.info(f"Successfully parsed TXT: {file_path.name}, " f"encoding: {encoding}, lines: {metadata['lines']}") return text, metadata except Exception as e: logger.error(f"Error parsing TXT file {file_path}: {e}") raise ValueError(f"Failed to parse TXT: {str(e)}") def validate_file(self, file_path: str, allowed_extensions: list) -> bool: """ Validate if file has allowed extension Args: file_path: Path to the file allowed_extensions: List of allowed extensions Returns: True if file is valid, False otherwise """ file_ext = Path(file_path).suffix.lower() return file_ext in allowed_extensions def get_file_info(self, file_path: str) -> dict: """ Get basic information about a file Args: file_path: Path to the file Returns: Dictionary with file information """ path = Path(file_path) if not path.exists(): raise FileNotFoundError(f"File not found: {file_path}") stat = path.stat() return { 'name': path.name, 'extension': path.suffix.lower(), 'size': stat.st_size, 'size_mb': round(stat.st_size / (1024 * 1024), 2), 'created': stat.st_ctime, 'modified': stat.st_mtime, 'is_valid_text': self.validate_file(file_path, self.text_extensions) } # Create a singleton instance file_parser = FileParser()