117 lines
4.9 KiB
Python
117 lines
4.9 KiB
Python
"""DOCX/ODT/RTF -> PDF via LibreOffice (Rich mode).
|
|
|
|
Word-faithful render (fonts, page colors, tables), unlike pandoc flatten.
|
|
Gate: pandoc_converter.can_handle. Guard (docx/odt): zip-slip+zip-bomb via
|
|
_validate_office_zip. Render: libreoffice --headless --convert-to pdf.
|
|
Title: docx core.xml, odt meta.xml, rtf docinfo. Finalize: shared
|
|
common.finalize_pdf tail. LibreOffice is a system binary, local (no egress).
|
|
Progress: 'Rendering with LibreOffice...'.
|
|
|
|
subprocess.run is referenced only via this module's own _run_office, so tests
|
|
may mock it hermetically (no live LibreOffice in the suite, avoiding the
|
|
soffice first-run lock hang).
|
|
"""
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from . import common
|
|
from .pandoc_converter import (_TITLE_MEMBER, _validate_office_zip, _zip_title)
|
|
|
|
log = logging.getLogger("epub2pdf")
|
|
LIBREOFFICE_TIMEOUT = 600
|
|
_RTF_TITLE_RE = re.compile(r"\\{\\info\\{\\title\\s+(.*?)\\}", re.IGNORECASE)
|
|
|
|
|
|
def _run_office(src: Path, work: Path) -> bytes:
|
|
"""Render `src` to PDF via LibreOffice headless; return the PDF bytes.
|
|
|
|
Runs `libreoffice --headless --convert-to pdf --outdir <work>`, then
|
|
locates the produced `<work>/<stem>.pdf` (named after the on-disk input
|
|
basename, e.g. web job job-abc123.docx -> job-abc123.pdf) and returns it.
|
|
Satisfies: spec render RICH_RENDER.
|
|
"""
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
binary = shutil.which("libreoffice") or "/usr/bin/libreoffice"
|
|
try:
|
|
subprocess.run(
|
|
[binary, "--headless", "--convert-to", "pdf",
|
|
"--outdir", str(work), str(src)],
|
|
check=True, timeout=LIBREOFFICE_TIMEOUT,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
)
|
|
except FileNotFoundError as exc:
|
|
raise ValueError("LibreOffice is not installed. Install it to enable Rich mode.") from exc
|
|
except subprocess.CalledProcessError as exc:
|
|
# Read the captured stderr (already on exc) into a clean, actionable message
|
|
detail = exc.stderr.decode(errors="replace").strip() or f"exit status {exc.returncode}"
|
|
raise ValueError(f"LibreOffice could not render the document ({detail})") from exc
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise ValueError("LibreOffice timed out rendering the document (600s)") from exc
|
|
# LibreOffice names its output after the ON-DISK input basename (e.g. the
|
|
# web job path job-abc123.docx -> job-abc123.pdf), NOT the original upload
|
|
# name, so locate the PDF there rather than by the source stem.
|
|
produced = work / f"{src.stem}.pdf"
|
|
if not produced.exists():
|
|
raise ValueError("LibreOffice produced no PDF output")
|
|
return produced.read_bytes()
|
|
|
|
|
|
def _office_title(src: Path) -> str:
|
|
"""Extract a document title for the PDF /Title metadata.
|
|
|
|
docx/odt: guard the zip then read the DC title member.
|
|
rtf: read the \\title field from the docinfo paragraph.
|
|
Returns '' when absent; finalize_pdf falls back to the stem.
|
|
Satisfies: spec metadata TITLE_FROM_DOCX/ODT/RTF.
|
|
"""
|
|
ext = src.suffix.lower()
|
|
if ext in _TITLE_MEMBER:
|
|
_validate_office_zip(src)
|
|
return _zip_title(src, _TITLE_MEMBER[ext]) or ""
|
|
if ext == ".rtf":
|
|
match = _RTF_TITLE_RE.search(common.decode_text(src.read_bytes()))
|
|
return match.group(1).strip() if match else ""
|
|
return ""
|
|
|
|
|
|
def convert(src: Path, job_id: str, progress_cb=None,
|
|
output_dir=None, source_name=None) -> str:
|
|
"""Render an office document to PDF via LibreOffice (Rich mode).
|
|
|
|
Gates on pandoc_converter.can_handle, extracts the title, renders with
|
|
LibreOffice, then finalizes with the shared common.finalize_pdf tail.
|
|
Satisfies: spec orchestrate RICH_ORCHESTRATION.
|
|
"""
|
|
if not can_handle(src.name):
|
|
raise ValueError(f"Unsupported file type: '{src.suffix or '<none>'}'")
|
|
|
|
import app # lazy: WORK_DIR lives with the EPUB pipeline (avoid import cycle)
|
|
# AC: render_mode defaults to "rich"; the dispatcher routes the office
|
|
# trio (.docx/.odt/.rtf) to this module when rich is active. An omitted
|
|
# source_name falls back to the on-disk name.
|
|
src_name = source_name or src.name
|
|
out_dir = common.default_output_dir(output_dir)
|
|
|
|
work = app.WORK_DIR / f"job-{job_id}"
|
|
filename = common.derive_export_name(src_name, out_dir)
|
|
title = _office_title(src)
|
|
|
|
common.report(progress_cb, 20.0, "Preparing document\u2026")
|
|
common.report(progress_cb, 50.0, "Rendering with LibreOffice\u2026")
|
|
try:
|
|
pdf_bytes = _run_office(src, work)
|
|
except ValueError as exc:
|
|
raise ValueError(f"LibreOffice conversion failed: {exc}") from exc
|
|
|
|
return common.finalize_pdf(pdf_bytes, out_dir, filename, job_id,
|
|
progress_cb=progress_cb, title=title)
|
|
|
|
|
|
def can_handle(ext: str) -> bool:
|
|
"""True if this converter claims `ext` (the office trio)."""
|
|
from .pandoc_converter import can_handle as _ph
|
|
return _ph(None, ext)
|