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