From fdc21e3316781228428b2e4db7420d0c7039c633 Mon Sep 17 00:00:00 2001 From: Mahdi Bazrafshan Date: Sun, 26 Jul 2026 09:37:40 +0330 Subject: [PATCH] feat(documents): add document parser, service, and API routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: - Need to parse .docx/.doc files into DocumentTree + markdown - Need API endpoints for document upload, processing, and deletion Changes: - Parser: .doc→.docx conversion via LibreOffice, XML-level extraction for table-heavy docs, heuristic heading detection - Service: upload, process (delegates to chunking), delete orchestration - Routes: POST /documents, POST /documents/{id}/process, DELETE /documents/{id}, GET /strategies --- src/documents/__init__.py | 1 + src/documents/models.py | 70 ++++++++ src/documents/parser.py | 361 ++++++++++++++++++++++++++++++++++++++ src/documents/routes.py | 71 ++++++++ src/documents/service.py | 151 ++++++++++++++++ 5 files changed, 654 insertions(+) create mode 100644 src/documents/__init__.py create mode 100644 src/documents/models.py create mode 100644 src/documents/parser.py create mode 100644 src/documents/routes.py create mode 100644 src/documents/service.py diff --git a/src/documents/__init__.py b/src/documents/__init__.py new file mode 100644 index 0000000..50125a4 --- /dev/null +++ b/src/documents/__init__.py @@ -0,0 +1 @@ +"""Document parsing and processing.""" diff --git a/src/documents/models.py b/src/documents/models.py new file mode 100644 index 0000000..4578800 --- /dev/null +++ b/src/documents/models.py @@ -0,0 +1,70 @@ +"""Request/response schemas for the Documents API.""" + +from typing import Optional + +from pydantic import BaseModel, Field + +from src.core.models import StrategyName + + +# ── Responses ────────────────────────────────────────────────────── + +class DocumentResponse(BaseModel): + """Returned after upload or GET.""" + + id: str + filename: str + paragraph_count: int + chunk_counts: dict[str, int] = Field(default_factory=dict) + created_at: str + + +class DocumentDetailResponse(DocumentResponse): + """Full document detail including tree and text preview.""" + + parsed_text_preview: str # first 500 chars + document_tree: dict # JSON-serialised DocumentTree + + +class ProcessRequest(BaseModel): + """Body for POST /documents/{id}/process.""" + + strategies: list[StrategyName] = Field( + description="Which chunking strategies to run", + min_length=1, + ) + + +class StrategyResult(BaseModel): + """Outcome for a single strategy in a processing run.""" + + strategy: StrategyName + status: str # "completed" | "failed" + chunks_produced: int = 0 + error: Optional[str] = None + + +class ProcessResponse(BaseModel): + """Returned after processing a document with selected strategies.""" + + document_id: str + strategies_completed: list[StrategyResult] + strategies_failed: list[StrategyResult] + + +class DeleteResponse(BaseModel): + """Returned after deleting a document.""" + + deleted: bool + document_id: str + + +# ── Strategies list ──────────────────────────────────────────────── + +class StrategyInfo(BaseModel): + name: StrategyName + description: str + + +class StrategiesResponse(BaseModel): + strategies: list[StrategyInfo] diff --git a/src/documents/parser.py b/src/documents/parser.py new file mode 100644 index 0000000..4650ce3 --- /dev/null +++ b/src/documents/parser.py @@ -0,0 +1,361 @@ +"""DOCX parser: extracts document tree + flat markdown. + +Uses python-docx to read paragraph styles and build a hierarchical +DocumentTree (Document > Section > Article > Paragraph). Also produces +a markdown representation consumed by chunking strategies. + +Supports both .docx and .doc formats. .doc files are converted to +.docx via LibreOffice headless mode before parsing. + +Handles documents where content is in tables (not just paragraphs). +""" + +from __future__ import annotations + +import re +import subprocess +import tempfile +from pathlib import Path + +from docx import Document as DocxDocument +from docx.document import Document as DocxDocumentType +from docx.text.paragraph import Paragraph + +from src.core.models import ( + DocumentTree, + DocumentTreeNode, + NodeType, +) +from src.core.exceptions import DocumentProcessingError + + +# ── Heading level → node type mapping ────────────────────────────── + +_HEADING_MAP: dict[int, NodeType] = { + 1: NodeType.SECTION, + 2: NodeType.ARTICLE, + # 3+ also ARTICLE (nesting depth determines hierarchy) +} + + +def _heading_level(style_name: str) -> int | None: + """Return the heading level from a style name, or None if not a heading. + + Handles both "Heading 1" (display name) and "Heading1" (style ID from XML). + """ + m = re.match(r"^heading\s*(\d+)$", style_name.strip(), re.IGNORECASE) + if m: + return int(m.group(1)) + return None + + +def _node_type_for_level(level: int) -> NodeType: + return _HEADING_MAP.get(level, NodeType.ARTICLE) + + +# ── Heuristic heading detection for table-heavy documents ────────── + +# Patterns that look like section/article headings in Farsi/English docs +_HEADING_PATTERNS: list[tuple[str, int]] = [ + # Farsi section markers: "بخش اول", "بخش دوم", etc. + (re.compile(r"^بخش\s+"), 1), + # Numbered sections: "1-1", "2-1", "10-1", "1-1-8" + (re.compile(r"^\d+[\-\.]\d+"), 2), + # Starred sections: "*بخش اول", "*تعریف" + (re.compile(r"^\*\s*بخش\s+"), 1), + (re.compile(r"^\*\s*\S"), 2), +] + + +def _is_likely_heading(text: str) -> int | None: + """Detect heading-like patterns in table-extracted text. + + Returns the heading level (1 or 2) if detected, else None. + Used when the document has no real heading styles (table-only content). + """ + text = text.strip() + if len(text) > 150: # headings are short + return None + for pattern, level in _HEADING_PATTERNS: + if pattern.match(text): + return level + return None + + +def _detect_heading_blocks(blocks: list[_TextBlock]) -> bool: + """Upgrade paragraph blocks to heading blocks based on content patterns. + + Returns True if any blocks were upgraded. + Only activates when no real headings exist in the document. + """ + # Check if there are already real headings + has_headings = any(_heading_level(b.style_name) is not None for b in blocks) + if has_headings: + return False + + upgraded = False + for block in blocks: + if block.style_name != "Normal": + continue + level = _is_likely_heading(block.text) + if level is not None: + block.style_name = f"Heading{level}" + upgraded = True + + return upgraded + + +# ── Text block extraction (paragraphs + tables) ─────────────────── + +class _TextBlock: + """A unit of text extracted from the document, preserving reading order.""" + __slots__ = ("style_name", "text") + + def __init__(self, style_name: str, text: str) -> None: + self.style_name = style_name + self.text = text + + +def _extract_text_blocks(doc: DocxDocumentType) -> list[_TextBlock]: + """Walk the document body in reading order, extracting paragraphs and tables. + + This handles documents where content lives inside table cells + (common in Farsi/Arabic .doc files exported from older Word versions). + """ + from docx.oxml.ns import qn + + blocks: list[_TextBlock] = [] + + for element in doc.element.body: + tag = element.tag.split("}")[-1] if "}" in element.tag else element.tag + + if tag == "p": + # Paragraph — extract text and style + text = element.text or "" + # Also check for runs (text split across formatting) + if not text.strip(): + runs = element.findall(qn("w:r")) + text = "".join(r.text or "" for r in runs) + + # Get style name + ppr = element.find(qn("w:pPr")) + style_name = "Normal" + if ppr is not None: + pstyle = ppr.find(qn("w:pStyle")) + if pstyle is not None: + style_name = pstyle.get(qn("w:val"), "Normal") + + if text.strip(): + blocks.append(_TextBlock(style_name, text.strip())) + + elif tag == "tbl": + # Table — extract all cell text as paragraph blocks + for row in element.findall(qn("w:tr")): + for cell in row.findall(qn("w:tc")): + for para in cell.findall(qn("w:p")): + # Get paragraph text + text = "" + runs = para.findall(qn("w:r")) + text = "".join(r.text or "" for r in runs) + + # Get style + ppr = para.find(qn("w:pPr")) + style_name = "Normal" + if ppr is not None: + pstyle = ppr.find(qn("w:pStyle")) + if pstyle is not None: + style_name = pstyle.get(qn("w:val"), "Normal") + + if text.strip(): + blocks.append(_TextBlock(style_name, text.strip())) + + return blocks + + +# ── Tree builder ─────────────────────────────────────────────────── + +def build_document_tree(paragraphs: list[Paragraph] | list[_TextBlock]) -> DocumentTreeNode: + """Build a DocumentTreeNode tree from a list of text blocks. + + Accepts either python-docx Paragraph objects or _TextBlock objects. + """ + root = DocumentTreeNode(node_type=NodeType.DOCUMENT, text="", heading=None, heading_level=None) + + # Stack of (level, node) for current nesting. level 0 = root. + stack: list[tuple[int, DocumentTreeNode]] = [(0, root)] + + for block in paragraphs: + # Get style name and text from either type + if isinstance(block, _TextBlock): + style_name = block.style_name + text = block.text + else: + style_name = block.style.name + text = block.text.strip() + + level = _heading_level(style_name) + + if not text: + continue # skip blank paragraphs + + if level is not None: + # Pop back to parent level + while len(stack) > 1 and stack[-1][0] >= level: + stack.pop() + + node = DocumentTreeNode( + node_type=_node_type_for_level(level), + text="", + heading=text, + heading_level=level, + ) + stack[-1][1].children.append(node) + stack.append((level, node)) + else: + # Body text — add as paragraph child of current heading + node = DocumentTreeNode( + node_type=NodeType.PARAGRAPH, + text=text, + ) + stack[-1][1].children.append(node) + + return root + + +# ── Markdown renderer ────────────────────────────────────────────── + +def tree_to_markdown(node: DocumentTreeNode, depth: int = 0) -> str: + """Render a DocumentTreeNode tree to markdown text.""" + parts: list[str] = [] + + if node.heading: + prefix = "#" * (node.heading_level or 1) + parts.append(f"{prefix} {node.heading}") + + if node.text: + parts.append(node.text) + + for child in node.children: + parts.append(tree_to_markdown(child, depth + 1)) + + return "\n\n".join(parts) + + +# ── Public API ───────────────────────────────────────────────────── + +class ParseResult: + """Output of parse_docx(): tree + markdown + raw text.""" + + __slots__ = ("tree", "markdown", "plain_text", "paragraph_count") + + def __init__( + self, + tree: DocumentTree, + markdown: str, + plain_text: str, + paragraph_count: int, + ) -> None: + self.tree = tree + self.markdown = markdown + self.plain_text = plain_text + self.paragraph_count = paragraph_count + + +# ── .doc → .docx conversion ─────────────────────────────────────── + +def _convert_doc_to_docx(doc_path: Path) -> Path: + """Convert a .doc file to .docx using LibreOffice headless mode. + + Returns the path to the converted .docx file (in a temp directory). + The caller is responsible for cleanup. + + Raises: + DocumentProcessingError: If conversion fails. + """ + out_dir = Path(tempfile.mkdtemp(prefix="docconv_")) + try: + result = subprocess.run( + [ + "libreoffice", + "--headless", + "--convert-to", "docx", + "--outdir", str(out_dir), + str(doc_path), + ], + capture_output=True, + text=True, + timeout=60, + ) + if result.returncode != 0: + raise DocumentProcessingError( + f"LibreOffice conversion failed: {result.stderr}" + ) + + # Find the converted file + converted = out_dir / doc_path.with_suffix(".docx").name + if not converted.exists(): + raise DocumentProcessingError( + f"Converted file not found: {converted}" + ) + return converted + except subprocess.TimeoutExpired: + raise DocumentProcessingError("LibreOffice conversion timed out") + except DocumentProcessingError: + raise + except Exception as exc: + raise DocumentProcessingError( + f"Failed to convert .doc to .docx: {exc}" + ) from exc + + +def parse_docx(file_path: str | Path) -> ParseResult: + """Parse a .docx or .doc file into a DocumentTree + markdown. + + .doc files are automatically converted to .docx via LibreOffice. + + Args: + file_path: Path to the .docx or .doc file. + + Returns: + ParseResult with tree, markdown, plain_text, and paragraph_count. + + Raises: + DocumentProcessingError: If the file cannot be parsed. + """ + path = Path(file_path) + if not path.exists(): + raise DocumentProcessingError(f"File not found: {path}") + + suffix = path.suffix.lower() + if suffix not in (".docx", ".doc"): + raise DocumentProcessingError(f"Not a supported file format: {suffix} (expected .docx or .doc)") + + # Convert .doc to .docx if needed + if suffix == ".doc": + path = _convert_doc_to_docx(path) + + try: + doc: DocxDocumentType = DocxDocument(str(path)) + except Exception as exc: + raise DocumentProcessingError(f"Failed to open DOCX: {exc}") from exc + + # Extract text blocks from paragraphs + tables (preserves reading order) + blocks = _extract_text_blocks(doc) + + if not blocks: + raise DocumentProcessingError("Document contains no text content") + + # Detect heading patterns in table-only documents + _detect_heading_blocks(blocks) + + root = build_document_tree(blocks) + tree = DocumentTree(root=root) + markdown = tree_to_markdown(root) + plain_text = "\n".join(b.text for b in blocks if b.text) + + return ParseResult( + tree=tree, + markdown=markdown, + plain_text=plain_text, + paragraph_count=len(blocks), + ) diff --git a/src/documents/routes.py b/src/documents/routes.py new file mode 100644 index 0000000..1425c77 --- /dev/null +++ b/src/documents/routes.py @@ -0,0 +1,71 @@ +"""Document and strategy API routes. + +Endpoints: + POST /documents Upload a .docx file + POST /documents/{id}/process Run chunking strategies (stub until Phase 2) + DELETE /documents/{id} Remove document + vectors + GET /strategies List available chunking strategies +""" + +from fastapi import APIRouter, File, UploadFile + +from src.core.exceptions import DocumentProcessingError +from src.core.models import PaginatedResponse, StrategyName +from src.documents.models import ( + DeleteResponse, + DocumentResponse, + ProcessRequest, + ProcessResponse, + StrategiesResponse, + StrategyInfo, +) +from src.documents import service + +router = APIRouter() + + +@router.post("/documents", response_model=DocumentResponse, status_code=201) +async def upload_document(file: UploadFile = File(...)): + """Upload a .docx file. Parses it, stores the document tree in SQLite.""" + if not file.filename: + raise DocumentProcessingError("No filename provided") + if not file.filename.lower().endswith((".docx", ".doc")): + raise DocumentProcessingError("Only .docx and .doc files are supported") + + content = await file.read() + if not content: + raise DocumentProcessingError("Empty file") + + doc = service.upload_document(file.filename, content) + return DocumentResponse( + id=doc["id"], + filename=doc["filename"], + paragraph_count=doc.get("paragraph_count", 0), + chunk_counts=doc.get("chunk_counts", {}), + created_at=doc["created_at"], + ) + + +@router.post("/documents/{doc_id}/process", response_model=ProcessResponse) +async def process_document(doc_id: str, request: ProcessRequest): + """Run selected chunking strategies on an uploaded document. + + Currently returns stub results until Phase 2 implements the strategies. + """ + return service.process_document(doc_id, request) + + +@router.delete("/documents/{doc_id}", response_model=DeleteResponse) +async def delete_document(doc_id: str): + """Delete a document and all its chunk vectors from Qdrant.""" + deleted = service.delete_document(doc_id) + return DeleteResponse(deleted=deleted, document_id=doc_id) + + +@router.get("/strategies", response_model=StrategiesResponse) +async def list_strategies(): + """List available chunking strategies with descriptions.""" + strategies = service.list_strategies() + return StrategiesResponse( + strategies=[StrategyInfo(**s) for s in strategies] + ) diff --git a/src/documents/service.py b/src/documents/service.py new file mode 100644 index 0000000..4e98033 --- /dev/null +++ b/src/documents/service.py @@ -0,0 +1,151 @@ +"""Document service: upload, parse, process, delete. + +Orchestrates the document lifecycle. Strategy processing delegates +to src/chunking/service.py for the actual chunking pipeline. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from src.core.exceptions import DocumentProcessingError +from src.core.models import StrategyName +from src.documents.models import ( + ProcessRequest, + ProcessResponse, + StrategyResult, +) +from src.documents.parser import parse_docx +from src.storage import sqlite as db +from src.storage import qdrant as qdr + +logger = logging.getLogger(__name__) + +STRATEGY_DESCRIPTIONS: dict[StrategyName, str] = { + StrategyName.RECURSIVE: ( + "Cascade splitting: headers > double newline > newline > punctuation > space. " + "Stops when chunks reach target size." + ), + StrategyName.FIXED_SIZE: ( + "Fixed-size token splitting with overlap. Simple baseline for comparison. " + "If smarter strategies can't beat this, they're not worth the complexity." + ), + StrategyName.SEMANTIC: ( + "Sentence-level embeddings → cosine similarity → dynamic chunk boundaries. " + "Chunks split when topic similarity drops below threshold." + ), + StrategyName.CONTEXTUAL_RETRIEVAL: ( + "Each chunk gets a short context summary from LLM prepended before embedding. " + "Anthropic's 2024 research showed 49% retrieval improvement." + ), + StrategyName.SEMANTIC_PARENT_CHILD: ( + "Paragraphs grouped into semantic clusters. Each cluster is a parent; " + "each paragraph is a child. Uses meaning, not headings." + ), +} + + +# ── Upload ───────────────────────────────────────────────────────── + +def upload_document(filename: str, file_bytes: bytes) -> dict[str, Any]: + """Parse a .docx upload, store in SQLite, return document record.""" + import tempfile + from pathlib import Path + + suffix = Path(filename).suffix or ".docx" + with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: + tmp.write(file_bytes) + tmp_path = tmp.name + + try: + result = parse_docx(tmp_path) + finally: + Path(tmp_path).unlink(missing_ok=True) + + doc = db.save_document( + filename=filename, + parsed_text=result.markdown, + document_tree=result.tree.model_dump_json(), + ) + + logger.info( + "Uploaded '%s': %d blocks, %d chars", + filename, result.paragraph_count, len(result.markdown), + ) + doc["paragraph_count"] = result.paragraph_count + return doc + + +# ── Process (strategies) ────────────────────────────────────────── + +def process_document( + doc_id: str, request: ProcessRequest +) -> ProcessResponse: + """Run selected chunking strategies on a stored document. + + Delegates to src/chunking/service.run_strategies() which handles: + 1. Loading DocumentTree + markdown from SQLite. + 2. Running each strategy (chunk → embed → Qdrant). + 3. Per-strategy failure isolation (ADR 0003). + """ + doc = db.get_document(doc_id) + if doc is None: + raise DocumentProcessingError(f"Document not found: {doc_id}") + + # Import here to avoid circular imports at module level + from src.chunking.service import run_strategies + + completed_raw, failed_raw = run_strategies(doc_id, request.strategies) + + completed = [ + StrategyResult( + strategy=r["strategy"], + status="completed", + chunks_produced=r["chunks_produced"], + ) + for r in completed_raw + ] + failed = [ + StrategyResult( + strategy=r["strategy"], + status="failed", + error=r.get("error"), + ) + for r in failed_raw + ] + + return ProcessResponse( + document_id=doc_id, + strategies_completed=completed, + strategies_failed=failed, + ) + + +# ── Delete ───────────────────────────────────────────────────────── + +def delete_document(doc_id: str) -> bool: + """Delete a document and all its Qdrant vectors.""" + doc = db.get_document(doc_id) + if doc is None: + return False + + for strategy_name, count in doc.get("chunk_counts", {}).items(): + if count > 0: + try: + strategy = StrategyName(strategy_name) + qdr.delete_document_chunks(strategy, doc["filename"]) + except Exception as exc: + logger.warning("Failed to delete Qdrant vectors for %s: %s", strategy_name, exc) + + return db.delete_document(doc_id) + + +# ── List strategies ──────────────────────────────────────────────── + +def list_strategies() -> list[dict[str, str]]: + """Return all available strategies with descriptions.""" + return [ + {"name": s.value, "description": STRATEGY_DESCRIPTIONS[s]} + for s in StrategyName + ]