feat(documents): add document parser, service, and API routes
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
This commit is contained in:
361
src/documents/parser.py
Normal file
361
src/documents/parser.py
Normal file
@@ -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),
|
||||
)
|
||||
Reference in New Issue
Block a user