Spaces:
Running
Running
File size: 3,595 Bytes
fc2fe9f 167c070 fc2fe9f 167c070 fc2fe9f 247a9a5 a965d17 247a9a5 fc2fe9f a965d17 fc2fe9f 247a9a5 167c070 fc2fe9f 167c070 fc2fe9f 167c070 3e047b8 167c070 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Document parsing for PDF, DOCX, PPTX, XLSX, and plain text — powered by MarkItDown."""
import io
import logging
from typing import Optional
logger = logging.getLogger("rga_auditor.parser")
MAX_FILE_SIZE = 50 * 1024 * 1024 # 50 MB
_md = None
def _get_markitdown():
global _md
if _md is None:
from markitdown import MarkItDown
_md = MarkItDown()
return _md
def _parse_with_markitdown(content: bytes, ext: str = "") -> str:
md = _get_markitdown()
result = md.convert_stream(io.BytesIO(content), file_extension=ext or None)
return result.text_content
def parse_pdf(content: bytes) -> str:
return _parse_with_markitdown(content, ".pdf")
def parse_pdf_with_pages(content: bytes) -> tuple[str, list[dict]]:
"""Parse PDF and return (full_text, page_ranges) where page_ranges
contains {start, end, page} character-offset-to-page mappings.
Uses pypdf (faster than pdfplumber for text extraction)."""
from pypdf import PdfReader
pages: list[dict] = []
texts: list[str] = []
offset = 0
reader = PdfReader(io.BytesIO(content))
for i, page in enumerate(reader.pages):
t = (page.extract_text() or "").strip()
if t:
texts.append(t)
start = offset
offset += len(t)
pages.append({"start": start, "end": offset, "page": i + 1})
offset += 2 # separator
full_text = "\n\n".join(texts)
return full_text, pages
def parse_docx(content: bytes) -> str:
return _parse_with_markitdown(content, ".docx")
def parse_pptx(content: bytes) -> str:
return _parse_with_markitdown(content, ".pptx")
def parse_xlsx(content: bytes) -> str:
return _parse_with_markitdown(content, ".xlsx")
def parse_text(content: bytes) -> str:
return content.decode("utf-8", errors="replace")
def parse_document(filename: str, content: bytes) -> str:
"""Dispatch by extension. Raises ValueError for unsupported types."""
if len(content) > MAX_FILE_SIZE:
raise ValueError(f"file too large: {len(content)} bytes (max {MAX_FILE_SIZE})")
name = (filename or "").lower()
if name.endswith(".pdf"):
return parse_pdf(content)
if name.endswith(".docx"):
return parse_docx(content)
if name.endswith(".pptx"):
return parse_pptx(content)
if name.endswith(".xlsx"):
return parse_xlsx(content)
if name.endswith((".txt", ".md", ".markdown")):
return parse_text(content)
raise ValueError(f"unsupported file type: {filename}")
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 100) -> list[str]:
"""Section-aware chunking: prefer splitting on headers, fall back to char windows."""
from langchain_text_splitters import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter
# Try markdown header-aware splitting first (works for MD, DOCX via MarkItDown)
header_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
)
sections = header_splitter.split_text(text)
if len(sections) > 1:
# Has headers — chunk within each section
splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=overlap)
chunks: list[str] = []
for sec in sections:
chunks.extend(splitter.split_text(sec.page_content))
return chunks
# No headers — fall back to recursive character splitting
splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=overlap)
return splitter.split_text(text)
|