Spaces:
Runtime error
Runtime error
| """Document extraction for DOCX / PDF / legacy DOC. | |
| Produces an ordered list of :class:`Block` objects (text + heading flag + level) | |
| so the template discoverer can infer section structure, and a plain-text helper | |
| for reference ingest. Legacy ``.doc`` files are converted to ``.docx`` first using | |
| the first available tool: pandoc, LibreOffice, or Microsoft Word via pywin32. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| import re | |
| import shutil | |
| import subprocess | |
| import sys | |
| import tempfile | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| logger = logging.getLogger(__name__) | |
| # Word VBA: wdFormatXMLDocument — Office Open XML (.docx) without macros. | |
| _WD_FORMAT_DOCX = 12 | |
| class Block: | |
| """One extracted paragraph/line.""" | |
| text: str | |
| is_heading: bool = False | |
| level: int = 0 | |
| def _soffice_executable() -> str | None: | |
| for name in ("soffice", "libreoffice"): | |
| p = shutil.which(name) | |
| if p: | |
| return p | |
| if sys.platform == "win32": | |
| for pf in ( | |
| os.environ.get("ProgramFiles", r"C:\Program Files"), | |
| os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"), | |
| ): | |
| cand = Path(pf) / "LibreOffice" / "program" / "soffice.exe" | |
| if cand.is_file(): | |
| return str(cand) | |
| return None | |
| def convert_doc_to_docx(source: Path) -> tuple[Path | None, Path | None]: | |
| """Convert a legacy ``.doc`` to ``.docx`` in a temp dir. | |
| Returns ``(docx_path, temp_dir)``; the caller must ``shutil.rmtree(temp_dir)``. | |
| Returns ``(None, None)`` on failure. | |
| """ | |
| out_dir = Path(tempfile.mkdtemp(prefix="rics_v2_doc_")) | |
| out = out_dir / f"{source.stem}.docx" | |
| pandoc = shutil.which("pandoc") | |
| if pandoc: | |
| try: | |
| subprocess.run( | |
| [pandoc, str(source.resolve()), "-o", str(out.resolve())], | |
| check=True, capture_output=True, text=True, timeout=180, | |
| ) | |
| if out.is_file() and out.stat().st_size > 0: | |
| logger.info("Converted %s via pandoc", source.name) | |
| return out, out_dir | |
| except (subprocess.CalledProcessError, OSError, subprocess.TimeoutExpired) as exc: | |
| logger.debug("pandoc .doc->.docx failed for %s: %s", source, exc) | |
| soffice = _soffice_executable() | |
| if soffice: | |
| try: | |
| subprocess.run( | |
| [soffice, "--headless", "--convert-to", "docx", "--outdir", | |
| str(out_dir), str(source.resolve())], | |
| check=True, capture_output=True, text=True, timeout=240, | |
| ) | |
| if out.is_file() and out.stat().st_size > 0: | |
| logger.info("Converted %s via LibreOffice", source.name) | |
| return out, out_dir | |
| except (subprocess.CalledProcessError, OSError, subprocess.TimeoutExpired) as exc: | |
| logger.debug("LibreOffice .doc->.docx failed for %s: %s", source, exc) | |
| try: | |
| import win32com.client # type: ignore[import-untyped] | |
| except ImportError: | |
| pass | |
| else: | |
| word = doc = None | |
| try: | |
| word = win32com.client.Dispatch("Word.Application") | |
| word.Visible = False | |
| word.DisplayAlerts = 0 | |
| doc = word.Documents.Open(str(source.resolve()), ReadOnly=True) | |
| doc.SaveAs2(str(out.resolve()), FileFormat=_WD_FORMAT_DOCX) | |
| doc.Close(SaveChanges=False) | |
| doc = None | |
| word.Quit() | |
| word = None | |
| if out.is_file() and out.stat().st_size > 0: | |
| logger.info("Converted %s via Microsoft Word (COM)", source.name) | |
| return out, out_dir | |
| except Exception as exc: # noqa: BLE001 | |
| logger.debug("Word COM .doc->.docx failed for %s: %s", source, exc) | |
| finally: | |
| if doc is not None: | |
| try: | |
| doc.Close(SaveChanges=False) | |
| except Exception: # noqa: BLE001 | |
| pass | |
| if word is not None: | |
| try: | |
| word.Quit() | |
| except Exception: # noqa: BLE001 | |
| pass | |
| shutil.rmtree(out_dir, ignore_errors=True) | |
| logger.warning( | |
| "Could not convert legacy .doc %s. Install pandoc, LibreOffice, or " | |
| "Microsoft Word + pywin32, or provide a .docx instead.", | |
| source, | |
| ) | |
| return None, None | |
| def _docx_blocks(path: Path) -> list[Block]: | |
| from docx import Document # type: ignore[import-untyped] | |
| doc = Document(str(path)) | |
| blocks: list[Block] = [] | |
| for para in doc.paragraphs: | |
| text = (para.text or "").strip() | |
| if not text: | |
| continue | |
| style = (para.style.name or "") if para.style else "" | |
| is_heading = style.lower().startswith("heading") or style.lower() == "title" | |
| level = 0 | |
| if is_heading: | |
| m = re.search(r"(\d+)", style) | |
| level = int(m.group(1)) if m else 1 | |
| blocks.append(Block(text=text, is_heading=is_heading, level=level)) | |
| return blocks | |
| def _pdf_blocks(path: Path) -> list[Block]: | |
| """Heuristic PDF extraction: short, title-cased lines are treated as headings.""" | |
| text = _pdf_text(path) | |
| blocks: list[Block] = [] | |
| for raw in text.splitlines(): | |
| line = raw.strip() | |
| if not line: | |
| continue | |
| words = line.split() | |
| looks_heading = ( | |
| len(words) <= 9 | |
| and not line.endswith((".", ",", ";")) | |
| and (line.isupper() or line[:1].isupper()) | |
| and len(line) <= 80 | |
| ) | |
| blocks.append(Block(text=line, is_heading=looks_heading, level=1 if looks_heading else 0)) | |
| return blocks | |
| def _pdf_text(path: Path) -> str: | |
| try: | |
| from pypdf import PdfReader | |
| reader = PdfReader(str(path)) | |
| return "\n".join((page.extract_text() or "") for page in reader.pages) | |
| except Exception as exc: # noqa: BLE001 | |
| logger.warning("pypdf extraction failed for %s (%s); trying PyMuPDF.", path, exc) | |
| try: | |
| import fitz # type: ignore[import-untyped] | |
| with fitz.open(str(path)) as d: | |
| return "\n".join(page.get_text() for page in d) | |
| except Exception as exc: # noqa: BLE001 | |
| logger.warning("PyMuPDF extraction failed for %s (%s).", path, exc) | |
| return "" | |
| def extract_blocks(path: Path) -> list[Block]: | |
| """Return ordered blocks for a DOCX / PDF / legacy DOC file.""" | |
| suffix = path.suffix.lower() | |
| if suffix == ".doc": | |
| converted, tmp = convert_doc_to_docx(path) | |
| if converted is None: | |
| return [] | |
| try: | |
| return _docx_blocks(converted) | |
| finally: | |
| shutil.rmtree(tmp, ignore_errors=True) | |
| if suffix in (".docx", ".docm"): | |
| return _docx_blocks(path) | |
| if suffix == ".pdf": | |
| return _pdf_blocks(path) | |
| raise ValueError(f"Unsupported document type: {path.suffix}") | |
| def extract_text(path: Path) -> str: | |
| """Return plain text for a document (used for reference ingest).""" | |
| return "\n".join(b.text for b in extract_blocks(path)) | |