utils.py

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")