leocrm/app/plugins/builtins/unified_search/text_extraction.py

"""Text extraction from various file formats for DMS indexing."""

from __future__ import annotations

import logging
from typing import Any

logger = logging.getLogger(__name__)

MAX_CHARS = 50_000


async def extract_text_from_file(file_path: str, mime_type: str) -> str:
    """Extract text content from a file based on its MIME type.

    Supports:
    - PDF (via PyMuPDF/fitz)
    - DOCX (via python-docx)
    - XLSX (via openpyxl)
    - PPTX (via python-pptx)
    - text/* (via aiofiles)
    - Images: returns empty (OCR not implemented)

    Args:
        file_path: Absolute path to the file.
        mime_type: MIME type of the file.

    Returns:
        Extracted text, truncated to MAX_CHARS. Empty string on error.
    """
    try:
        if mime_type == "application/pdf":
            return await _extract_pdf(file_path)
        elif mime_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
            return await _extract_docx(file_path)
        elif mime_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":
            return await _extract_xlsx(file_path)
        elif mime_type == "application/vnd.openxmlformats-officedocument.presentationml.presentation":
            return await _extract_pptx(file_path)
        elif mime_type.startswith("text/"):
            return await _extract_text(file_path)
        elif mime_type.startswith("image/"):
            return ""
        else:
            logger.debug("Unsupported mime_type=%s for text extraction", mime_type)
            return ""
    except Exception:
        logger.warning("Failed to extract text from %s (mime=%s)", file_path, mime_type)
        return ""


def _truncate(text: str) -> str:
    """Truncate text to MAX_CHARS with indicator."""
    if len(text) > MAX_CHARS:
        return text[:MAX_CHARS] + "... [truncated]"
    return text


async def _extract_pdf(file_path: str) -> str:
    """Extract text from PDF using PyMuPDF (fitz)."""
    import fitz  # PyMuPDF

    doc = fitz.open(file_path)
    parts: list[str] = []
    for page in doc:
        parts.append(page.get_text())
    doc.close()
    return _truncate("\n".join(parts))


async def _extract_docx(file_path: str) -> str:
    """Extract text from DOCX using python-docx."""
    from docx import Document

    doc = Document(file_path)
    parts: list[str] = []
    for para in doc.paragraphs:
        if para.text.strip():
            parts.append(para.text)
    # Also extract table text
    for table in doc.tables:
        for row in table.rows:
            for cell in row.cells:
                if cell.text.strip():
                    parts.append(cell.text)
    return _truncate("\n".join(parts))


async def _extract_xlsx(file_path: str) -> str:
    """Extract text from XLSX using openpyxl."""
    from openpyxl import load_workbook

    wb = load_workbook(file_path, read_only=True, data_only=True)
    parts: list[str] = []
    for ws in wb.worksheets:
        for row in ws.iter_rows(values_only=True):
            row_text = " ".join(str(c) for c in row if c is not None)
            if row_text.strip():
                parts.append(row_text)
    wb.close()
    return _truncate("\n".join(parts))


async def _extract_pptx(file_path: str) -> str:
    """Extract text from PPTX using python-pptx."""
    from pptx import Presentation

    prs = Presentation(file_path)
    parts: list[str] = []
    for slide in prs.slides:
        for shape in slide.shapes:
            if shape.has_text_frame:
                for para in shape.text_frame.paragraphs:
                    text = para.text.strip()
                    if text:
                        parts.append(text)
    return _truncate("\n".join(parts))


async def _extract_text(file_path: str) -> str:
    """Read plain text file using aiofiles."""
    import aiofiles

    async with aiofiles.open(file_path, mode="r", encoding="utf-8", errors="replace") as f:
        content = await f.read()
    return _truncate(content)