529 lines
20 KiB
Python
529 lines
20 KiB
Python
"""File conversion utilities.
|
|
|
|
Converts document files (PDF, PPT, Excel, Word) to Markdown.
|
|
|
|
PDF conversion strategy (auto mode):
|
|
1. Try pymupdf4llm if installed — better heading detection, faster on most files.
|
|
2. If output is suspiciously short (< _MIN_CHARS_PER_PAGE chars/page, or < 200 chars
|
|
total when page count is unavailable), treat as image-based and fall back to MarkItDown.
|
|
3. If pymupdf4llm is not installed, use MarkItDown directly (existing behaviour).
|
|
|
|
Parsing runs in a bounded pool of spawned processes so CPU-heavy document
|
|
readers cannot hold the gateway's GIL. Async callers wait and write sidecars in
|
|
a dedicated thread pool, leaving the event loop and default executor available.
|
|
|
|
No FastAPI or HTTP dependencies — pure utility functions.
|
|
"""
|
|
|
|
import atexit
|
|
import logging
|
|
import multiprocessing
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from concurrent.futures.process import BrokenProcessPool
|
|
from pathlib import Path
|
|
|
|
from deerflow.config.app_config import get_app_config
|
|
from deerflow.utils.file_io import FileIOPool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# File extensions that should be converted to markdown
|
|
CONVERTIBLE_EXTENSIONS = {
|
|
".pdf",
|
|
".ppt",
|
|
".pptx",
|
|
".xls",
|
|
".xlsx",
|
|
".doc",
|
|
".docx",
|
|
".odt",
|
|
".ods",
|
|
".odp",
|
|
}
|
|
|
|
# MarkItDown reads modern OOXML files directly. LibreOffice is used as a
|
|
# compatibility bridge for legacy binary Office files and OpenDocument files,
|
|
# which are common in offline/intranet deployments but are not handled
|
|
# consistently by the pure-Python readers.
|
|
_LIBREOFFICE_TARGETS = {
|
|
".doc": ".docx",
|
|
".ppt": ".pptx",
|
|
".xls": ".xlsx",
|
|
".odt": ".docx",
|
|
".odp": ".pptx",
|
|
".ods": ".xlsx",
|
|
}
|
|
_LIBREOFFICE_TIMEOUT_SECONDS = 120
|
|
|
|
MAX_CONVERTED_MARKDOWN_CHARS = 2_000_000
|
|
_CONVERSION_CONCURRENCY = 2
|
|
_CONVERSION_SEMAPHORE = threading.BoundedSemaphore(_CONVERSION_CONCURRENCY)
|
|
_CONVERSION_IO = FileIOPool(_CONVERSION_CONCURRENCY, _CONVERSION_CONCURRENCY, "document-conversion")
|
|
_PROCESS_POOL_LOCK = threading.Lock()
|
|
_PROCESS_POOL: ProcessPoolExecutor | None = None
|
|
|
|
|
|
def _get_conversion_process_pool() -> ProcessPoolExecutor:
|
|
"""Lazily start isolated readers without forking a multithreaded gateway."""
|
|
global _PROCESS_POOL
|
|
with _PROCESS_POOL_LOCK:
|
|
if _PROCESS_POOL is None:
|
|
_PROCESS_POOL = ProcessPoolExecutor(
|
|
max_workers=_CONVERSION_CONCURRENCY,
|
|
mp_context=multiprocessing.get_context("spawn"),
|
|
)
|
|
return _PROCESS_POOL
|
|
|
|
|
|
def shutdown_conversion_workers(*, wait: bool = True) -> None:
|
|
"""Release document reader processes; a later request can start a new pool."""
|
|
global _PROCESS_POOL
|
|
with _PROCESS_POOL_LOCK:
|
|
pool, _PROCESS_POOL = _PROCESS_POOL, None
|
|
if pool is not None:
|
|
pool.shutdown(wait=wait, cancel_futures=True)
|
|
|
|
|
|
def _discard_broken_process_pool(pool: ProcessPoolExecutor) -> None:
|
|
global _PROCESS_POOL
|
|
with _PROCESS_POOL_LOCK:
|
|
if _PROCESS_POOL is not pool:
|
|
return
|
|
_PROCESS_POOL = None
|
|
pool.shutdown(wait=False, cancel_futures=True)
|
|
|
|
|
|
def _convert_document_worker(file_path: str, pdf_converter: str) -> str:
|
|
"""Child entry point: only the path and converter cross the process boundary."""
|
|
return _bound_converted_markdown(_do_convert_unbounded(Path(file_path), pdf_converter))
|
|
|
|
|
|
def _run_conversion_in_process(file_path: Path, pdf_converter: str) -> str:
|
|
pool = _get_conversion_process_pool()
|
|
try:
|
|
return pool.submit(_convert_document_worker, str(file_path), pdf_converter).result()
|
|
except BrokenProcessPool:
|
|
# Do not retry a document that may have crashed a native parser. The
|
|
# existing public API reports conversion failure; the next request gets
|
|
# fresh workers instead of inheriting a permanently unusable executor.
|
|
_discard_broken_process_pool(pool)
|
|
raise
|
|
|
|
|
|
atexit.register(shutdown_conversion_workers)
|
|
|
|
# If pymupdf4llm produces fewer characters *per page* than this threshold,
|
|
# the PDF is likely image-based or encrypted — fall back to MarkItDown.
|
|
# Rationale: normal text PDFs yield 200-2000 chars/page; image-based PDFs
|
|
# yield close to 0. 50 chars/page gives a wide safety margin.
|
|
# Falls back to absolute 200-char check when page count is unavailable.
|
|
_MIN_CHARS_PER_PAGE = 50
|
|
|
|
|
|
def markdown_sidecar_path(file_path: Path) -> Path:
|
|
"""Return the collision-free Markdown companion path for a document.
|
|
|
|
Keeping the original extension (``report.docx.md``) avoids the historical
|
|
collision where ``report.docx`` and ``report.pdf`` both overwrote
|
|
``report.md``.
|
|
"""
|
|
|
|
return file_path.with_name(f"{file_path.name}.md")
|
|
|
|
|
|
def find_markdown_sidecar(file_path: Path) -> Path | None:
|
|
"""Find a new-style sidecar, falling back to the legacy ``<stem>.md``."""
|
|
|
|
# Only formats handled by the conversion pipeline can own a generated
|
|
# Markdown companion. Without this guard, unrelated uploads such as
|
|
# ``report.txt`` or ``report.jpg`` incorrectly claimed a real user upload
|
|
# named ``report.md`` as their legacy sidecar.
|
|
if file_path.suffix.lower() not in CONVERTIBLE_EXTENSIONS:
|
|
return None
|
|
|
|
current = markdown_sidecar_path(file_path)
|
|
if current.is_file():
|
|
return current
|
|
legacy = file_path.with_suffix(".md")
|
|
if legacy != file_path and legacy.is_file():
|
|
return legacy
|
|
return None
|
|
|
|
|
|
def _pymupdf_output_too_sparse(text: str, file_path: Path) -> bool:
|
|
"""Return True if pymupdf4llm output is suspiciously short (image-based PDF).
|
|
|
|
Uses chars-per-page rather than an absolute threshold so that both short
|
|
documents (few pages, few chars) and long documents (many pages, many chars)
|
|
are handled correctly.
|
|
"""
|
|
chars = len(text.strip())
|
|
doc = None
|
|
pages: int | None = None
|
|
try:
|
|
import pymupdf
|
|
|
|
doc = pymupdf.open(str(file_path))
|
|
pages = len(doc)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
if doc is not None:
|
|
try:
|
|
doc.close()
|
|
except Exception:
|
|
pass
|
|
if pages is not None and pages > 0:
|
|
return (chars / pages) < _MIN_CHARS_PER_PAGE
|
|
# Fallback: absolute threshold when page count is unavailable
|
|
return chars < 200
|
|
|
|
|
|
def _convert_pdf_with_pymupdf4llm(file_path: Path) -> str | None:
|
|
"""Attempt PDF conversion with pymupdf4llm.
|
|
|
|
Returns the markdown text, or None if pymupdf4llm is not installed or
|
|
if conversion fails (e.g. encrypted/corrupt PDF).
|
|
"""
|
|
try:
|
|
import pymupdf4llm
|
|
except ImportError:
|
|
return None
|
|
|
|
try:
|
|
return pymupdf4llm.to_markdown(str(file_path))
|
|
except Exception:
|
|
logger.exception("pymupdf4llm failed to convert %s; falling back to MarkItDown", file_path.name)
|
|
return None
|
|
|
|
|
|
def _convert_with_markitdown(file_path: Path) -> str:
|
|
"""Convert any supported file to markdown text using MarkItDown."""
|
|
from markitdown import MarkItDown
|
|
|
|
md = MarkItDown()
|
|
return md.convert(str(file_path)).text_content
|
|
|
|
|
|
def _convert_with_libreoffice_then_markitdown(file_path: Path) -> str | None:
|
|
"""Normalize a legacy Office/OpenDocument file, then extract its text.
|
|
|
|
Each conversion gets an isolated LibreOffice user profile. This avoids
|
|
the profile-lock conflict that otherwise appears when several uploads are
|
|
converted concurrently in the same sandbox container.
|
|
"""
|
|
target_suffix = _LIBREOFFICE_TARGETS.get(file_path.suffix.lower())
|
|
if target_suffix is None:
|
|
return None
|
|
|
|
executable = shutil.which("soffice") or shutil.which("libreoffice")
|
|
if executable is None:
|
|
logger.warning(
|
|
"LibreOffice is unavailable; trying MarkItDown directly for %s",
|
|
file_path.name,
|
|
)
|
|
return None
|
|
|
|
with tempfile.TemporaryDirectory(prefix="deerflow-office-") as temporary:
|
|
temp_root = Path(temporary).resolve()
|
|
output_dir = temp_root / "output"
|
|
profile_dir = temp_root / "profile"
|
|
output_dir.mkdir()
|
|
profile_dir.mkdir()
|
|
command = [
|
|
executable,
|
|
"--headless",
|
|
"--nologo",
|
|
"--nodefault",
|
|
"--nolockcheck",
|
|
"--nofirststartwizard",
|
|
f"-env:UserInstallation={profile_dir.as_uri()}",
|
|
"--convert-to",
|
|
target_suffix.lstrip("."),
|
|
"--outdir",
|
|
str(output_dir),
|
|
str(file_path.resolve()),
|
|
]
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=_LIBREOFFICE_TIMEOUT_SECONDS,
|
|
env={**os.environ, "HOME": str(temp_root)},
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
logger.exception("LibreOffice failed to normalize %s", file_path.name)
|
|
return None
|
|
|
|
converted_path = output_dir / f"{file_path.stem}{target_suffix}"
|
|
if completed.returncode != 0 or not converted_path.is_file():
|
|
logger.error(
|
|
"LibreOffice could not normalize %s (exit=%s, stderr=%s)",
|
|
file_path.name,
|
|
completed.returncode,
|
|
completed.stderr.strip()[-1000:],
|
|
)
|
|
return None
|
|
return _convert_with_markitdown(converted_path)
|
|
|
|
|
|
def _bound_converted_markdown(text: str) -> str:
|
|
"""Apply a non-disableable cap before converted text is persisted or returned."""
|
|
if len(text) <= MAX_CONVERTED_MARKDOWN_CHARS:
|
|
return text
|
|
marker = (
|
|
"\n\n<!-- deerflow-conversion-truncated: extracted text exceeded "
|
|
f"{MAX_CONVERTED_MARKDOWN_CHARS} characters; use the original file for targeted inspection -->\n"
|
|
)
|
|
return text[: MAX_CONVERTED_MARKDOWN_CHARS - len(marker)] + marker
|
|
|
|
|
|
def _do_convert(file_path: Path, pdf_converter: str) -> str:
|
|
"""Synchronous conversion with a shared limit for sync and async callers.
|
|
|
|
Args:
|
|
file_path: Path to the file.
|
|
pdf_converter: "auto" | "pymupdf4llm" | "markitdown"
|
|
"""
|
|
with _CONVERSION_SEMAPHORE:
|
|
return _run_conversion_in_process(file_path, pdf_converter)
|
|
|
|
|
|
def _do_convert_unbounded(file_path: Path, pdf_converter: str) -> str:
|
|
"""Parser implementation; callers must use :func:`_do_convert`."""
|
|
is_pdf = file_path.suffix.lower() == ".pdf"
|
|
|
|
normalized_office_text = _convert_with_libreoffice_then_markitdown(file_path)
|
|
if normalized_office_text is not None:
|
|
return normalized_office_text
|
|
|
|
if is_pdf and pdf_converter != "markitdown":
|
|
# Try pymupdf4llm first (auto or explicit)
|
|
pymupdf_text = _convert_pdf_with_pymupdf4llm(file_path)
|
|
|
|
if pymupdf_text is not None:
|
|
# pymupdf4llm is installed
|
|
if pdf_converter == "pymupdf4llm":
|
|
# Explicit — use as-is regardless of output length
|
|
return pymupdf_text
|
|
# auto mode: fall back if output looks like a failed parse.
|
|
# Use chars-per-page to distinguish image-based PDFs (near 0) from
|
|
# legitimately short documents.
|
|
if not _pymupdf_output_too_sparse(pymupdf_text, file_path):
|
|
return pymupdf_text
|
|
logger.warning(
|
|
"pymupdf4llm produced only %d chars for %s (likely image-based PDF); falling back to MarkItDown",
|
|
len(pymupdf_text.strip()),
|
|
file_path.name,
|
|
)
|
|
# pymupdf4llm not installed or fallback triggered → use MarkItDown
|
|
|
|
return _convert_with_markitdown(file_path)
|
|
|
|
|
|
def convert_file_to_markdown_sync(file_path: Path) -> Path | None:
|
|
"""Convert and write a sidecar inside an existing file worker.
|
|
|
|
Callers must keep this synchronous parsing and disk work off an event loop.
|
|
The shared conversion gate also applies to upload workers calling directly.
|
|
"""
|
|
try:
|
|
pdf_converter = _get_pdf_converter()
|
|
text = _do_convert(file_path, pdf_converter)
|
|
|
|
md_path = markdown_sidecar_path(file_path)
|
|
md_path.write_text(text, encoding="utf-8")
|
|
|
|
logger.info("Converted %s to markdown: %s (%d chars)", file_path.name, md_path.name, len(text))
|
|
return md_path
|
|
except Exception as e:
|
|
logger.error("Failed to convert %s to markdown: %s", file_path.name, e)
|
|
return None
|
|
|
|
|
|
async def convert_file_to_markdown(file_path: Path) -> Path | None:
|
|
"""Convert and persist Markdown without blocking other requests.
|
|
|
|
Admission waits on the event loop, never inside the default executor.
|
|
Cancellation waits for the worker so callers can safely remove its input.
|
|
"""
|
|
return await _CONVERSION_IO.run_until_complete(convert_file_to_markdown_sync, file_path)
|
|
|
|
|
|
def _extract_file_to_markdown_sync(file_path: Path) -> str:
|
|
return _do_convert(file_path, _get_pdf_converter())
|
|
|
|
|
|
async def extract_file_to_markdown(file_path: Path) -> str | None:
|
|
"""Extract a supported document to Markdown without creating a sidecar file.
|
|
|
|
Knowledge-ingest scanners use this variant because running a read-only
|
|
distillation task must never add generated ``.md`` files to a live skill.
|
|
Conversion always runs off the event loop; Office/PDF parsers can be CPU
|
|
intensive even for deceptively small inputs.
|
|
"""
|
|
|
|
if file_path.suffix.lower() not in CONVERTIBLE_EXTENSIONS:
|
|
return None
|
|
try:
|
|
return await _CONVERSION_IO.run_until_complete(_extract_file_to_markdown_sync, file_path)
|
|
except Exception as exc:
|
|
logger.error("Failed to extract %s as markdown: %s", file_path.name, exc)
|
|
return None
|
|
|
|
|
|
# Regex for bold-only lines that look like section headings.
|
|
# Targets SEC filing structural headings that pymupdf4llm renders as **bold**
|
|
# rather than # Markdown headings (because they use same font size as body text,
|
|
# distinguished only by bold+caps formatting).
|
|
#
|
|
# Pattern requires ALL of:
|
|
# 1. Entire line is a single **...** block (no surrounding prose)
|
|
# 2. Starts with a recognised structural keyword:
|
|
# - ITEM / PART / SECTION (with optional number/letter after)
|
|
# - SCHEDULE, EXHIBIT, APPENDIX, ANNEX, CHAPTER
|
|
# All-caps addresses, boilerplate ("CURRENT REPORT", "SIGNATURES",
|
|
# "WASHINGTON, DC 20549") do NOT start with these keywords and are excluded.
|
|
#
|
|
# Chinese headings (第三节...) are already captured as standard # headings
|
|
# by pymupdf4llm, so they don't need this pattern.
|
|
_BOLD_HEADING_RE = re.compile(r"^\*\*((ITEM|PART|SECTION|SCHEDULE|EXHIBIT|APPENDIX|ANNEX|CHAPTER)\b[A-Z0-9 .,\-]*)\*\*\s*$")
|
|
|
|
# Regex for split-bold headings produced by pymupdf4llm when a heading spans
|
|
# multiple text spans in the PDF (e.g. section number and title are separate spans).
|
|
# Matches lines like: **1** **Introduction** or **3.2** **Multi-Head Attention**
|
|
# Requirements:
|
|
# 1. Entire line consists only of **...** blocks separated by whitespace (no prose)
|
|
# 2. First block is a section number (digits and dots, e.g. "1", "3.2", "A.1")
|
|
# 3. Second block must not be purely numeric/punctuation — excludes financial table
|
|
# headers like **2023** **2022** **2021** while allowing non-ASCII titles such as
|
|
# **1** **概述** or accented words (negative lookahead instead of [A-Za-z])
|
|
# 4. At most two additional blocks (four total) with [^*]+ (no * inside) to keep
|
|
# the regex linear and avoid ReDoS on attacker-controlled content
|
|
_SPLIT_BOLD_HEADING_RE = re.compile(r"^\*\*[\dA-Z][\d\.]*\*\*\s+\*\*(?!\d[\d\s.,\-–—/:()%]*\*\*)[^*]+\*\*(?:\s+\*\*[^*]+\*\*){0,2}\s*$")
|
|
|
|
# Maximum number of outline entries injected into the agent context.
|
|
# Keeps prompt size bounded even for very long documents.
|
|
MAX_OUTLINE_ENTRIES = 50
|
|
|
|
_ALLOWED_PDF_CONVERTERS = {"auto", "pymupdf4llm", "markitdown"}
|
|
|
|
|
|
def _clean_bold_title(raw: str) -> str:
|
|
"""Normalise a title string that may contain pymupdf4llm bold artefacts.
|
|
|
|
pymupdf4llm sometimes emits adjacent bold spans as ``**A** **B**`` instead
|
|
of a single ``**A B**`` block. This helper merges those fragments and then
|
|
strips the outermost ``**...**`` wrapper so the caller gets plain text.
|
|
|
|
Examples::
|
|
|
|
"**Overview**" → "Overview"
|
|
"**UNITED STATES** **SECURITIES**" → "UNITED STATES SECURITIES"
|
|
"plain text" → "plain text" (unchanged)
|
|
"""
|
|
# Merge adjacent bold spans: "** **" → " "
|
|
merged = re.sub(r"\*\*\s*\*\*", " ", raw).strip()
|
|
# Strip outermost **...** if the whole string is wrapped
|
|
if m := re.fullmatch(r"\*\*(.+?)\*\*", merged, re.DOTALL):
|
|
return m.group(1).strip()
|
|
return merged
|
|
|
|
|
|
def extract_outline(md_path: Path) -> list[dict]:
|
|
"""Extract document outline (headings) from a Markdown file.
|
|
|
|
Recognises three heading styles produced by pymupdf4llm:
|
|
|
|
1. Standard Markdown headings: lines starting with one or more '#'.
|
|
Inline ``**...**`` wrappers and adjacent bold spans (``** **``) are
|
|
cleaned so the title is plain text.
|
|
|
|
2. Bold-only structural headings: ``**ITEM 1. BUSINESS**``, ``**PART II**``,
|
|
etc. SEC filings use bold+caps for section headings with the same font
|
|
size as body text, so pymupdf4llm cannot promote them to # headings.
|
|
|
|
3. Split-bold headings: ``**1** **Introduction**``, ``**3.2** **Attention**``.
|
|
pymupdf4llm emits these when the section number and title text are
|
|
separate spans in the underlying PDF (common in academic papers).
|
|
|
|
Args:
|
|
md_path: Path to the .md file.
|
|
|
|
Returns:
|
|
List of dicts with keys: title (str), line (int, 1-based).
|
|
When the outline is truncated at MAX_OUTLINE_ENTRIES, a sentinel entry
|
|
``{"truncated": True}`` is appended as the last element so callers can
|
|
render a "showing first N headings" hint without re-scanning the file.
|
|
Returns an empty list if the file cannot be read or has no headings.
|
|
"""
|
|
outline: list[dict] = []
|
|
try:
|
|
with md_path.open(encoding="utf-8") as f:
|
|
for lineno, line in enumerate(f, 1):
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
|
|
# Style 1: standard Markdown heading
|
|
if stripped.startswith("#"):
|
|
title = _clean_bold_title(stripped.lstrip("#").strip())
|
|
if title:
|
|
outline.append({"title": title, "line": lineno})
|
|
|
|
# Style 2: single bold block with SEC structural keyword
|
|
elif m := _BOLD_HEADING_RE.match(stripped):
|
|
title = m.group(1).strip()
|
|
if title:
|
|
outline.append({"title": title, "line": lineno})
|
|
|
|
# Style 3: split-bold heading — **<num>** **<title>**
|
|
# Regex already enforces max 4 blocks and non-numeric second block.
|
|
elif _SPLIT_BOLD_HEADING_RE.match(stripped):
|
|
title = " ".join(re.findall(r"\*\*([^*]+)\*\*", stripped))
|
|
if title:
|
|
outline.append({"title": title, "line": lineno})
|
|
|
|
if len(outline) >= MAX_OUTLINE_ENTRIES:
|
|
outline.append({"truncated": True})
|
|
break
|
|
except Exception:
|
|
return []
|
|
|
|
return outline
|
|
|
|
|
|
def _get_uploads_config_value(key: str, default: object) -> object:
|
|
"""Read a value from the uploads config, supporting dict and attribute access."""
|
|
cfg = get_app_config()
|
|
uploads_cfg = getattr(cfg, "uploads", None)
|
|
if isinstance(uploads_cfg, dict):
|
|
return uploads_cfg.get(key, default)
|
|
return getattr(uploads_cfg, key, default)
|
|
|
|
|
|
def _get_pdf_converter() -> str:
|
|
"""Read pdf_converter setting from app config, defaulting to 'auto'.
|
|
|
|
Normalizes the value to lowercase and validates it against the allowed set
|
|
so that values like 'AUTO' or 'MarkItDown' from config.yaml don't silently
|
|
fall through to unexpected behaviour.
|
|
"""
|
|
try:
|
|
raw = str(_get_uploads_config_value("pdf_converter", "auto")).strip().lower()
|
|
if raw not in _ALLOWED_PDF_CONVERTERS:
|
|
logger.warning("Invalid pdf_converter value %r; falling back to 'auto'", raw)
|
|
return "auto"
|
|
return raw
|
|
except Exception:
|
|
pass
|
|
return "auto"
|