File size: 4,212 Bytes
447450c | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
Document Processor — parses uploaded files (PDF, DOCX, TXT, MD, PY)
into LangChain Documents ready for embedding and indexing.
"""
from __future__ import annotations
import io
import re
from pathlib import Path
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from app.config import get_settings
from app.observability.logger import logger
SUPPORTED_EXTENSIONS = {".pdf", ".docx", ".txt", ".md", ".py", ".rst"}
def _extract_pdf(content: bytes, filename: str) -> str:
"""Extract text from PDF bytes using pypdf."""
try:
import pypdf
reader = pypdf.PdfReader(io.BytesIO(content))
return "\n\n".join(
page.extract_text() or "" for page in reader.pages
).strip()
except ImportError:
logger.warning("pypdf not installed — PDF extraction unavailable")
return ""
except Exception as exc:
logger.error(f"PDF extraction failed for {filename}: {exc}")
return ""
def _extract_docx(content: bytes, filename: str) -> str:
"""Extract text from DOCX bytes using python-docx."""
try:
import docx
doc = docx.Document(io.BytesIO(content))
return "\n\n".join(p.text for p in doc.paragraphs if p.text.strip())
except ImportError:
logger.warning("python-docx not installed — DOCX extraction unavailable")
return ""
except Exception as exc:
logger.error(f"DOCX extraction failed for {filename}: {exc}")
return ""
def _clean_text(text: str) -> str:
"""Normalize whitespace and remove null bytes."""
text = text.replace("\x00", "")
text = re.sub(r"\r\n", "\n", text)
text = re.sub(r"\n{4,}", "\n\n\n", text)
return text.strip()
def process_file(
content: bytes,
filename: str,
user_id: str,
doc_id: str,
) -> list[Document]:
"""
Parse uploaded file bytes into chunked LangChain Documents.
Args:
content: Raw file bytes.
filename: Original filename (used for extension detection).
user_id: Owner's user identifier.
doc_id: UUID assigned to this upload (for deletion tracking).
Returns:
List of Document objects ready for embedding.
"""
settings = get_settings()
suffix = Path(filename).suffix.lower()
if suffix not in SUPPORTED_EXTENSIONS:
raise ValueError(
f"Unsupported file type '{suffix}'. "
f"Supported: {', '.join(SUPPORTED_EXTENSIONS)}"
)
# ── Extract raw text ──────────────────────────────────────────────────
if suffix == ".pdf":
raw_text = _extract_pdf(content, filename)
elif suffix == ".docx":
raw_text = _extract_docx(content, filename)
else:
# TXT, MD, PY, RST — decode as UTF-8
try:
raw_text = content.decode("utf-8", errors="replace")
except Exception:
raw_text = content.decode("latin-1", errors="replace")
raw_text = _clean_text(raw_text)
if not raw_text:
raise ValueError(f"No extractable text found in '{filename}'")
# ── Chunk ─────────────────────────────────────────────────────────────
splitter = RecursiveCharacterTextSplitter(
chunk_size=settings.CHUNK_SIZE,
chunk_overlap=settings.CHUNK_OVERLAP,
separators=["\n\n\n", "\n\n", "\n", ". ", " "],
)
base_doc = Document(
page_content=raw_text,
metadata={
"source": "user_doc",
"filename": filename,
"user_id": user_id,
"doc_id": doc_id,
"file_type": suffix.lstrip("."),
"question_id": 0,
"q_score": 0,
},
)
chunks = splitter.split_documents([base_doc])
logger.info_data(
"Document processed",
filename=filename,
user_id=user_id,
doc_id=doc_id,
raw_chars=len(raw_text),
chunks=len(chunks),
)
return chunks
|