app/services/parser.py

"""Author RAG Chatbot SaaS — Document Parser.

Converts uploaded files (PDF, EPUB, DOCX, TXT) to plain text.
RULE: Always detect file type by magic bytes before parsing.
RULE: Return structured result with page count and extracted text.
"""

import re
from dataclasses import dataclass

import structlog

from app.exceptions.ingestion import CorruptedFileError, ParseError

logger = structlog.get_logger(__name__)


@dataclass
class ParseResult:
    """Result of parsing a document."""

    text: str         # Full extracted plain text
    page_count: int   # Number of pages (0 for plain text)
    char_count: int   # Total character count


def parse_document(file_path: str, extension: str) -> ParseResult:
    """Parse a document file into plain text.

    Dispatches to the appropriate parser based on file extension.

    Args:
        file_path: Absolute path to the file.
        extension: File type: 'pdf', 'epub', 'docx', or 'txt'.

    Returns:
        ParseResult with extracted text and metadata.

    Raises:
        ParseError: If the document cannot be parsed.
        CorruptedFileError: If the file is corrupted.
    """
    parsers = {
        "pdf": _parse_pdf,
        "epub": _parse_epub,
        "docx": _parse_docx,
        "txt": _parse_txt,
    }
    parser = parsers.get(extension)
    if not parser:
        raise ParseError(file_path, f"No parser for extension '{extension}'")

    logger.debug("Parsing document", path=file_path, extension=extension)
    result = parser(file_path)
    logger.info("Parsed document", extension=extension, pages=result.page_count, chars=result.char_count)
    return result


def _parse_pdf(file_path: str) -> ParseResult:
    """Extract text from a PDF file using PyPDF2.

    Args:
        file_path: Absolute path to the PDF.

    Returns:
        ParseResult with extracted text.
    """
    try:
        import PyPDF2
        pages_text = []
        with open(file_path, "rb") as f:
            reader = PyPDF2.PdfReader(f)
            if reader.is_encrypted:
                raise ParseError(file_path, "PDF is password-protected or DRM-encrypted")
            for page in reader.pages:
                text = page.extract_text() or ""
                pages_text.append(text)

        full_text = "\n\n".join(pages_text)
        full_text = _clean_text(full_text)

        if not full_text.strip():
            raise ParseError(
                file_path,
                "No text could be extracted. This may be a scanned image PDF. OCR is not supported."
            )
        return ParseResult(
            text=full_text,
            page_count=len(pages_text),
            char_count=len(full_text),
        )
    except ParseError:
        raise
    except Exception as e:
        raise CorruptedFileError(file_path) from e


def _parse_epub(file_path: str) -> ParseResult:
    """Extract text from an EPUB file.

    Args:
        file_path: Absolute path to the EPUB.

    Returns:
        ParseResult with extracted text.
    """
    try:
        import ebooklib
        from ebooklib import epub
        from html.parser import HTMLParser

        class _TextExtractor(HTMLParser):
            def __init__(self):
                super().__init__()
                self.texts = []

            def handle_data(self, data):
                stripped = data.strip()
                if stripped:
                    self.texts.append(stripped)

        book = epub.read_epub(file_path)
        chapters = []
        for item in book.get_items_of_type(ebooklib.ITEM_DOCUMENT):
            content = item.get_content().decode("utf-8", errors="ignore")
            extractor = _TextExtractor()
            extractor.feed(content)
            chapter_text = " ".join(extractor.texts)
            if chapter_text.strip():
                chapters.append(chapter_text)

        full_text = "\n\n".join(chapters)
        full_text = _clean_text(full_text)
        if not full_text.strip():
            raise ParseError(file_path, "No text content found in EPUB")

        return ParseResult(text=full_text, page_count=len(chapters), char_count=len(full_text))
    except ParseError:
        raise
    except Exception as e:
        raise CorruptedFileError(file_path) from e


def _parse_docx(file_path: str) -> ParseResult:
    """Extract text from a DOCX file.

    Args:
        file_path: Absolute path to the DOCX.

    Returns:
        ParseResult with extracted text.
    """
    try:
        from docx import Document
        doc = Document(file_path)
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        full_text = "\n\n".join(paragraphs)
        full_text = _clean_text(full_text)
        if not full_text.strip():
            raise ParseError(file_path, "No text content found in DOCX")
        return ParseResult(text=full_text, page_count=0, char_count=len(full_text))
    except ParseError:
        raise
    except Exception as e:
        raise CorruptedFileError(file_path) from e


def _parse_txt(file_path: str) -> ParseResult:
    """Read plain text file.

    Args:
        file_path: Absolute path to the text file.

    Returns:
        ParseResult with file content.
    """
    try:
        with open(file_path, "r", encoding="utf-8", errors="replace") as f:
            text = f.read()
        text = _clean_text(text)
        if not text.strip():
            raise ParseError(file_path, "Text file is empty")
        return ParseResult(text=text, page_count=0, char_count=len(text))
    except ParseError:
        raise
    except Exception as e:
        raise ParseError(file_path, str(e)) from e


def _clean_text(text: str) -> str:
    """Normalize extracted text — remove excessive whitespace and control chars.

    Args:
        text: Raw extracted text.

    Returns:
        Cleaned text string.
    """
    # Remove null bytes and control characters (except newlines/tabs)
    text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
    # Normalize multiple whitespace to single space
    text = re.sub(r"[ \t]+", " ", text)
    # Normalize multiple newlines to max 2
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()