Extract Office document text during live export

Author: AlexAgo83

logics/backlog/item_089_worker_text_extract_artifacts_for_sharepoint_documents.md

@@ -3,8 +3,8 @@
 > From version: 1.5.1
 > Schema version: 1.0
 > Status: Done
-> Understanding: 94%
-> Confidence: 91%
+> Understanding: 98%
+> Confidence: 94%
 > Progress: 100%
 > Complexity: High
 > Theme: Data / Worker / Corpus
@@ -57,4 +57,5 @@ flowchart LR
 
 - Implemented `RuntimeStore.extract_artifact_relative_path(...)` and worker live export extract artifact writing under the configured runtime store.
 - Corpus documents now carry `extractionStatus`, `extractionReason`, and `extractPath` when exported from the Graph worker path.
+- Extended worker extraction beyond text-like files by downloading bounded binary content and extracting OOXML body text for DOCX/PPTX/XLSX, with optional PDF text extraction through `pypdf`.
 - `rtk python3 -m pytest worker/tests/test_live_export_service.py` passed with coverage for successful text extract, metadata-only unsupported source, and empty text download classified as unreadable.

worker/app/services/live_export_service.py

@@ -2,10 +2,12 @@
 
 import hashlib
 import html
+import io
 import json
 import os
 import re
 import time
+import zipfile
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -23,6 +25,9 @@
 TEXTUAL_MIME_PREFIXES = ("text/", "application/json", "application/xml", "application/xhtml+xml")
 TEXTUAL_EXTENSIONS = {"txt", "md", "markdown", "csv", "tsv", "json", "xml", "html", "htm", "aspx"}
 MAX_TEXT_DOWNLOAD_BYTES = 256 * 1024
+MAX_BINARY_EXTRACT_BYTES = 15 * 1024 * 1024
+OOXML_EXTENSIONS = {"docx", "pptx", "xlsx"}
+PDF_EXTENSIONS = {"pdf"}
 
 
 @dataclass
@@ -80,6 +85,13 @@ def get_text(self, path_or_url: str) -> Tuple[str, str]:
         content_type = response.headers.get("content-type", "")
         return response.text, content_type
 
+    def get_bytes(self, path_or_url: str) -> Tuple[bytes, str]:
+        response = self._request("GET", path_or_url)
+        if response.status_code >= 400:
+            raise RuntimeError(f"Graph content request failed ({response.status_code})")
+        content_type = response.headers.get("content-type", "")
+        return response.content, content_type
+
     def list_all(
         self,
         path_or_url: str,
@@ -568,11 +580,19 @@ def _crawl_drive_items(
             raw_text = ""
             extraction_reason = ""
             is_textual_item = self._is_textual_item(item_name, mime_type)
+            is_binary_extractable = self._is_binary_extractable_item(item_name, mime_type)
             if is_textual_item and (not isinstance(item_size, int) or item_size <= MAX_TEXT_DOWNLOAD_BYTES):
                 raw_text = self._try_download_text(client, f"/drives/{drive['id']}/items/{item['id']}")
                 if not raw_text:
                     extraction_reason = "text_download_empty"
-            elif isinstance(item_size, int) and item_size > MAX_TEXT_DOWNLOAD_BYTES:
+            elif is_binary_extractable and (not isinstance(item_size, int) or item_size <= MAX_BINARY_EXTRACT_BYTES):
+                raw_text, extraction_reason = self._try_download_binary_extract(
+                    client,
+                    f"/drives/{drive['id']}/items/{item['id']}",
+                    item_name,
+                    mime_type,
+                )
+            elif isinstance(item_size, int) and item_size > (MAX_BINARY_EXTRACT_BYTES if is_binary_extractable else MAX_TEXT_DOWNLOAD_BYTES):
                 extraction_reason = "file_too_large"
                 report_text(f"[{site_name}] {drive_name}{current_path} over size limit ({item_size} bytes), keeping metadata only")
             else:
@@ -585,7 +605,7 @@ def _crawl_drive_items(
             extracted_text = raw_text if raw_text and not self._is_metadata_only_text(raw_text) else ""
             extraction_status = self._classify_extraction_status(
                 text=extracted_text,
-                attempted_text_download=is_textual_item and extraction_reason != "file_too_large",
+                attempted_text_download=(is_textual_item or is_binary_extractable) and extraction_reason != "file_too_large",
                 reason=extraction_reason,
             )
             if extracted_text:
@@ -650,6 +670,75 @@ def _try_download_text(self, client: GraphClient, item_path: str) -> str:
             return text.replace("\u0000", "").strip()
         return ""
 
+    def _try_download_binary_extract(self, client: GraphClient, item_path: str, name: str, mime_type: str) -> Tuple[str, str]:
+        try:
+            content, downloaded_mime_type = client.get_bytes(f"{item_path}/content")
+        except Exception:
+            return "", "binary_download_failed"
+
+        if not content:
+            return "", "binary_download_empty"
+        if len(content) > MAX_BINARY_EXTRACT_BYTES:
+            return "", "file_too_large"
+
+        effective_mime_type = downloaded_mime_type or mime_type
+        text = self._extract_plain_text_from_binary(name, effective_mime_type, content)
+        if text:
+            return text, ""
+        return "", "binary_text_extraction_failed"
+
+    def _extract_plain_text_from_binary(self, name: str, mime_type: str, content: bytes) -> str:
+        extension = name.rsplit(".", 1)[-1].lower() if "." in name else ""
+        if extension in OOXML_EXTENSIONS or "officedocument" in mime_type:
+            return self._extract_ooxml_text(content)
+        if extension in PDF_EXTENSIONS or mime_type == "application/pdf":
+            return self._extract_pdf_text(content)
+        return ""
+
+    def _extract_ooxml_text(self, content: bytes) -> str:
+        try:
+            with zipfile.ZipFile(io.BytesIO(content)) as archive:
+                text_parts: List[str] = []
+                for name in archive.namelist():
+                    if not self._is_ooxml_text_part(name):
+                        continue
+                    try:
+                        raw_xml = archive.read(name).decode("utf-8", errors="ignore")
+                    except Exception:
+                        continue
+                    text = self._xml_to_text(raw_xml)
+                    if text:
+                        text_parts.append(text)
+                return self._normalize_extracted_text(" ".join(text_parts))
+        except Exception:
+            return ""
+
+    def _is_ooxml_text_part(self, name: str) -> bool:
+        return (
+            (name.startswith("word/") and name.endswith(".xml")) or
+            (name.startswith("ppt/slides/") and name.endswith(".xml")) or
+            name == "xl/sharedStrings.xml"
+        )
+
+    def _xml_to_text(self, value: str) -> str:
+        text = re.sub(r"<[^>]+>", " ", value)
+        return html.unescape(text)
+
+    def _extract_pdf_text(self, content: bytes) -> str:
+        try:
+            from pypdf import PdfReader  # type: ignore[import-not-found]
+        except Exception:
+            return ""
+        try:
+            reader = PdfReader(io.BytesIO(content))
+            text_parts = [(page.extract_text() or "") for page in reader.pages[:80]]
+            return self._normalize_extracted_text(" ".join(text_parts))
+        except Exception:
+            return ""
+
+    def _normalize_extracted_text(self, value: str) -> str:
+        return re.sub(r"\s+", " ", value.replace("\u0000", " ")).strip()
+
     def _classify_extraction_status(self, *, text: str, attempted_text_download: bool, reason: str) -> str:
         if text.strip():
             return "full_text"
@@ -886,6 +975,10 @@ def _is_textual_item(self, name: str, mime_type: str) -> bool:
             return True
         return any(mime_type.startswith(prefix) for prefix in TEXTUAL_MIME_PREFIXES)
 
+    def _is_binary_extractable_item(self, name: str, mime_type: str) -> bool:
+        extension = name.rsplit(".", 1)[-1].lower() if "." in name else ""
+        return extension in OOXML_EXTENSIONS or extension in PDF_EXTENSIONS or "officedocument" in mime_type or mime_type == "application/pdf"
+
     def _infer_file_type(self, name: str, mime_type: str) -> str:
         extension = name.rsplit(".", 1)[-1].lower() if "." in name else ""
         file_type_map = {

worker/requirements.txt

@@ -1,6 +1,7 @@
 fastapi==0.118.0
 httpx==0.28.1
 pydantic-settings==2.11.0
+pypdf==6.2.0
 python-jose[cryptography]==3.5.0
 pytest==8.4.2
 pytest-asyncio==1.2.0

worker/tests/test_jobs.py

@@ -17,7 +17,7 @@
 def build_jobs_service(tmp_path) -> JobsService:
     settings = Settings(
         WORKER_MODE="local",
-        WORKER_RUNTIME_DATA_DIR=tmp_path,
+        WORKER_RUNTIME_DATA_DIR=tmp_path / "data" / "runtime",
     )
     runtime_store = RuntimeStore(settings.runtime_data_dir)
     corpus_service = CorpusService(settings=settings)
@@ -53,8 +53,8 @@ def test_jobs_service_runs_evaluate_and_persists_files(tmp_path) -> None:
     completed = wait_for_terminal_status(service, started["jobId"])
     assert completed["status"] == "succeeded"
     assert completed["result"]["passCount"] >= 1
-    assert (tmp_path / "jobs" / f"{started['jobId']}.json").exists()
-    assert (tmp_path / "jobs" / f"{started['jobId']}.events.jsonl").exists()
+    assert (service._runtime_store.jobs_dir() / f"{started['jobId']}.json").exists()
+    assert (service._runtime_store.jobs_dir() / f"{started['jobId']}.events.jsonl").exists()
 
 
 def test_jobs_service_runs_ingest_and_writes_sync_state(tmp_path) -> None:
@@ -65,7 +65,7 @@ def test_jobs_service_runs_ingest_and_writes_sync_state(tmp_path) -> None:
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["mode"] == "mock"
-    assert (tmp_path / "sync-state.json").exists()
+    assert (service._runtime_store.sync_state_path()).exists()
 
 
 def test_jobs_service_runs_ingest_in_live_mode_when_corpus_path_is_provided(tmp_path) -> None:
@@ -81,7 +81,7 @@ def test_jobs_service_runs_ingest_in_live_mode_when_corpus_path_is_provided(tmp_
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["mode"] == "live"
-    assert (tmp_path / "sync-state.live.json").exists()
+    assert (service._runtime_store.sync_state_path("live")).exists()
 
 
 def test_jobs_service_runs_analyze_and_writes_analysis_artifacts(tmp_path) -> None:
@@ -95,16 +95,16 @@ def test_jobs_service_runs_analyze_and_writes_analysis_artifacts(tmp_path) -> No
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["analyzed"] == 5
-    assert (tmp_path / "analyzed-corpus.json").exists()
-    assert (tmp_path / "analyze-report.json").exists()
+    assert (service._runtime_store.analyzed_corpus_path()).exists()
+    assert (service._runtime_store.analyze_report_path()).exists()
 
 
 def test_jobs_service_analyze_uses_extract_backed_text(tmp_path) -> None:
     service = build_jobs_service(tmp_path)
     corpus = json.loads(service._corpus_service.load_corpus_bytes().decode("utf-8"))
     extract_path = "extracts/site-test/doc-extract.json"
-    (tmp_path / extract_path).parent.mkdir(parents=True, exist_ok=True)
-    (tmp_path / extract_path).write_text(
+    (service._runtime_store.runtime_dir / extract_path).parent.mkdir(parents=True, exist_ok=True)
+    (service._runtime_store.runtime_dir / extract_path).write_text(
         json.dumps({"text": "Real extracted body text. It contains actual policy obligations and review owners."}),
         encoding="utf-8",
     )
@@ -135,11 +135,11 @@ def test_jobs_service_analyze_uses_extract_backed_text(tmp_path) -> None:
     completed = wait_for_terminal_status(service, started["jobId"])
 
     assert completed["status"] == "succeeded"
-    analyzed_corpus = json.loads((tmp_path / "analyzed-corpus.json").read_text(encoding="utf-8"))
+    analyzed_corpus = json.loads((service._runtime_store.analyzed_corpus_path()).read_text(encoding="utf-8"))
     section = analyzed_corpus["documents"][0]["analysis"]["sections"][0]["content"]
     assert "Real extracted body text" in section
     assert not section.startswith("Source:")
-    report = json.loads((tmp_path / "analyze-report.json").read_text(encoding="utf-8"))
+    report = json.loads((service._runtime_store.analyze_report_path()).read_text(encoding="utf-8"))
     assert report["extractionQuality"]["full_text"] == 1
 
 
@@ -174,10 +174,10 @@ def test_jobs_service_analyze_excludes_metadata_only_placeholder(tmp_path) -> No
     completed = wait_for_terminal_status(service, started["jobId"])
 
     assert completed["status"] == "succeeded"
-    analyzed_corpus = json.loads((tmp_path / "analyzed-corpus.json").read_text(encoding="utf-8"))
+    analyzed_corpus = json.loads((service._runtime_store.analyzed_corpus_path()).read_text(encoding="utf-8"))
     assert analyzed_corpus["documents"][0]["analysis"]["status"] == "excluded"
     assert analyzed_corpus["documents"][0]["analysis"]["excludedReason"] == "metadata_only_extract"
-    report = json.loads((tmp_path / "analyze-report.json").read_text(encoding="utf-8"))
+    report = json.loads((service._runtime_store.analyze_report_path()).read_text(encoding="utf-8"))
     assert report["extractionQuality"]["metadata_only"] == 1
 
 
@@ -193,7 +193,7 @@ def test_jobs_service_analyze_falls_back_to_local_when_provider_requested(tmp_pa
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["provider"] == "openai"
-    analyzed_corpus = (tmp_path / "analyzed-corpus.json").read_text(encoding="utf-8")
+    analyzed_corpus = (service._runtime_store.analyzed_corpus_path()).read_text(encoding="utf-8")
     assert '"providerStatus": "fallback"' in analyzed_corpus
 
 
@@ -253,13 +253,13 @@ def json(self) -> dict:
     assert completed["result"]["providerSuccesses"] == 1
     assert completed["result"]["providerFallbacks"] == 0
 
-    analyzed_corpus = json.loads((tmp_path / "analyzed-corpus.json").read_text(encoding="utf-8"))
+    analyzed_corpus = json.loads((service._runtime_store.analyzed_corpus_path()).read_text(encoding="utf-8"))
     first_analysis = analyzed_corpus["documents"][0]["analysis"]
     assert first_analysis["provider"] == "openai"
     assert first_analysis["providerStatus"] == "provider"
     assert first_analysis["summary"] == "Provider summary."
 
-    report = json.loads((tmp_path / "analyze-report.json").read_text(encoding="utf-8"))
+    report = json.loads((service._runtime_store.analyze_report_path()).read_text(encoding="utf-8"))
     assert report["actualInputTokens"] == 120
     assert report["actualOutputTokens"] == 45
     assert report["tokenCountMode"] == "actual"
@@ -282,9 +282,9 @@ def test_jobs_service_runs_export_live_and_writes_live_artifacts(tmp_path) -> No
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["sourceKind"] == "mock-baseline"
-    assert (tmp_path.parent.parent / "public" / "live-corpus.json").exists()
-    assert (tmp_path / "live-export-checkpoint.json").exists()
-    assert (tmp_path / "sync-state.live.json").exists()
+    assert (service._runtime_store.live_corpus_path()).exists()
+    assert (service._runtime_store.live_export_checkpoint_path()).exists()
+    assert (service._runtime_store.sync_state_path("live")).exists()
 
 
 def test_jobs_service_export_live_accepts_explicit_input_override(tmp_path) -> None:
@@ -303,8 +303,8 @@ def test_jobs_service_export_live_accepts_explicit_input_override(tmp_path) -> N
     )
     completed = wait_for_terminal_status(service, started["jobId"])
 
-    published_corpus = (tmp_path.parent.parent / "public" / "live-corpus.json").read_text(encoding="utf-8")
-    checkpoint = (tmp_path / "live-export-checkpoint.json").read_text(encoding="utf-8")
+    published_corpus = (service._runtime_store.live_corpus_path()).read_text(encoding="utf-8")
+    checkpoint = (service._runtime_store.live_export_checkpoint_path()).read_text(encoding="utf-8")
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["sourceKind"] == "explicit-input"
@@ -320,14 +320,14 @@ def test_jobs_service_runs_publish_analysis_after_analyze(tmp_path) -> None:
         options={"env": {"DEEPVAULT_DATA_MODE": "mock", "DEEPVAULT_ANALYZE_LIMIT": "3"}},
     )
     wait_for_terminal_status(service, analyze["jobId"])
-    assert (tmp_path / "analyzed-corpus.json").exists()
+    assert (service._runtime_store.analyzed_corpus_path()).exists()
 
     publish = service.start_job(job_type="publish-analysis", options={"env": {"DEEPVAULT_DATA_MODE": "mock"}})
     completed = wait_for_terminal_status(service, publish["jobId"])
 
     assert completed["status"] == "succeeded"
     assert completed["result"]["analyzedCount"] >= 1
-    published_path = tmp_path.parent.parent / "public" / "live-corpus.json"
+    published_path = service._runtime_store.live_corpus_path()
     assert published_path.exists()
     published = json.loads(published_path.read_text(encoding="utf-8"))
     analyzed_docs = [doc for doc in published["documents"] if isinstance(doc.get("analysis"), dict) and doc["analysis"].get("status") == "analyzed"]
@@ -352,7 +352,7 @@ def test_jobs_service_run_job_blocks_until_terminal_status(tmp_path) -> None:
     assert completed["status"] == "succeeded"
     assert completed["launchedBy"] == "worker-cli"
     assert completed["client"] == "worker-cli"
-    assert (tmp_path / "jobs" / f"{completed['jobId']}.json").exists()
+    assert (service._runtime_store.jobs_dir() / f"{completed['jobId']}.json").exists()
 
 
 def test_jobs_service_can_cancel_running_job(tmp_path) -> None: