| """
|
| DOCX Parser
|
| Part of SOVEREIGN PYTHON LLM ENGINE
|
|
|
| Parse Microsoft Word documents (.docx).
|
| """
|
|
|
| from typing import Any
|
| from pathlib import Path
|
| from dataclasses import dataclass
|
| import io
|
|
|
| from ...core.evidence import WORMLedger
|
|
|
|
|
| @dataclass
|
| class DOCXDocument:
|
| """Parsed DOCX document"""
|
| text: str
|
| paragraphs: list[str]
|
| tables: list[list[list[str]]]
|
| metadata: dict[str, Any]
|
|
|
|
|
| class DOCXParser:
|
| """
|
| DOCX document parser using python-docx.
|
| """
|
|
|
| def __init__(self, worm_ledger: WORMLedger | None = None):
|
| """
|
| Initialize DOCX parser.
|
|
|
| Args:
|
| worm_ledger: Optional WORM ledger
|
| """
|
| self.worm_ledger = worm_ledger
|
|
|
| async def parse(self, file_path: Path) -> DOCXDocument:
|
| """
|
| Parse DOCX file.
|
|
|
| Args:
|
| file_path: Path to DOCX file
|
|
|
| Returns:
|
| DOCXDocument with extracted content
|
| """
|
| try:
|
| from docx import Document
|
| except ImportError:
|
| raise ImportError("python-docx is required. Install: pip install python-docx")
|
|
|
| import asyncio
|
|
|
| def _sync_parse():
|
| doc = Document(file_path)
|
|
|
|
|
| paragraphs = []
|
| for para in doc.paragraphs:
|
| if para.text.strip():
|
| paragraphs.append(para.text)
|
|
|
|
|
| text = "\n\n".join(paragraphs)
|
|
|
|
|
| tables = []
|
| for table in doc.tables:
|
| table_data = []
|
| for row in table.rows:
|
| row_data = [cell.text for cell in row.cells]
|
| table_data.append(row_data)
|
| tables.append(table_data)
|
|
|
|
|
| metadata = {}
|
| if doc.core_properties:
|
| props = doc.core_properties
|
| metadata = {
|
| "title": props.title or "",
|
| "author": props.author or "",
|
| "subject": props.subject or "",
|
| "keywords": props.keywords or "",
|
| "created": props.created.isoformat() if props.created else None,
|
| "modified": props.modified.isoformat() if props.modified else None,
|
| }
|
|
|
| return DOCXDocument(
|
| text=text,
|
| paragraphs=paragraphs,
|
| tables=tables,
|
| metadata=metadata
|
| )
|
|
|
| return await asyncio.to_thread(_sync_parse)
|
|
|
| async def parse_bytes(self, docx_bytes: bytes) -> DOCXDocument:
|
| """
|
| Parse DOCX from bytes.
|
|
|
| Args:
|
| docx_bytes: DOCX file bytes
|
|
|
| Returns:
|
| DOCXDocument
|
| """
|
| try:
|
| from docx import Document
|
| except ImportError:
|
| raise ImportError("python-docx is required")
|
|
|
| import asyncio
|
|
|
| def _sync_parse():
|
| doc = Document(io.BytesIO(docx_bytes))
|
|
|
| paragraphs = []
|
| for para in doc.paragraphs:
|
| if para.text.strip():
|
| paragraphs.append(para.text)
|
|
|
| text = "\n\n".join(paragraphs)
|
|
|
| tables = []
|
| for table in doc.tables:
|
| table_data = []
|
| for row in table.rows:
|
| row_data = [cell.text for cell in row.cells]
|
| table_data.append(row_data)
|
| tables.append(table_data)
|
|
|
| metadata = {}
|
| if doc.core_properties:
|
| props = doc.core_properties
|
| metadata = {
|
| "title": props.title or "",
|
| "author": props.author or "",
|
| }
|
|
|
| return DOCXDocument(
|
| text=text,
|
| paragraphs=paragraphs,
|
| tables=tables,
|
| metadata=metadata
|
| )
|
|
|
| return await asyncio.to_thread(_sync_parse)
|
|
|
|
|
|
|
| async def parse_docx_tool(file_path: str) -> dict:
|
| """
|
| Tool wrapper for DOCX parsing.
|
|
|
| Args:
|
| file_path: Path to DOCX file
|
|
|
| Returns:
|
| Dictionary with parsed content
|
| """
|
| parser = DOCXParser()
|
| result = await parser.parse(Path(file_path))
|
|
|
| return {
|
| "text": result.text,
|
| "paragraphs_count": len(result.paragraphs),
|
| "tables_count": len(result.tables),
|
| "metadata": result.metadata
|
| }
|
|
|
