import os import re import logging from typing import List logger = logging.getLogger(__name__) def load_text_file(filepath: str) -> str: """Load plain text file and return content.""" with open(filepath, "r", encoding="utf-8", errors="ignore") as f: return f.read() def load_pdf_file(filepath: str) -> str: """Extract text from PDF using pypdf.""" try: from pypdf import PdfReader reader = PdfReader(filepath) return "\n".join(page.extract_text() or "" for page in reader.pages) except Exception as e: logger.error(f"PDF extraction failed: {e}") raise RuntimeError(f"Could not read PDF: {e}") def load_document(filepath: str) -> str: """Load a .txt or .pdf file and return raw text.""" ext = os.path.splitext(filepath)[1].lower() if ext == ".txt": return load_text_file(filepath) elif ext == ".pdf": return load_pdf_file(filepath) else: raise ValueError(f"Unsupported file format: {ext}") def clean_text(text: str) -> str: """Normalize whitespace and remove non-printable characters.""" text = re.sub(r'\s+', ' ', text) text = re.sub(r'[^\x20-\x7E\n]', '', text) return text.strip() def chunk_text(text: str, chunk_size: int, chunk_overlap: int, min_chunk_size: int) -> List[str]: """ Split text into overlapping chunks. Tries to split on sentence boundaries first, falls back to hard split. """ sentences = re.split(r'(?<=[.!?])\s+', text) chunks = [] current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) + 1 <= chunk_size: current_chunk += (" " if current_chunk else "") + sentence else: if len(current_chunk.strip()) >= min_chunk_size: chunks.append(current_chunk.strip()) overlap_text = current_chunk[-chunk_overlap:] if len(current_chunk) > chunk_overlap else current_chunk current_chunk = overlap_text + (" " if overlap_text else "") + sentence if len(current_chunk.strip()) >= min_chunk_size: chunks.append(current_chunk.strip()) logger.info(f"Text split into {len(chunks)} chunks") return chunks def format_context(chunks: List[str]) -> str: """Format retrieved chunks into a single context string for the LLM.""" return "\n\n---\n\n".join( f"[Excerpt {i+1}]:\n{chunk}" for i, chunk in enumerate(chunks) ) def validate_file(filepath: str, max_size_mb: float) -> None: """Validate file exists, has a supported extension, and is within size limit.""" from config import SUPPORTED_FORMATS if not os.path.exists(filepath): raise FileNotFoundError(f"File not found: {filepath}") ext = os.path.splitext(filepath)[1].lower() if ext not in SUPPORTED_FORMATS: raise ValueError(f"Unsupported format '{ext}'. Supported: {SUPPORTED_FORMATS}") size_mb = os.path.getsize(filepath) / (1024 * 1024) if size_mb > max_size_mb: raise ValueError(f"File too large ({size_mb:.1f} MB). Limit: {max_size_mb} MB")