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