270 lines
10 KiB
Python
270 lines
10 KiB
Python
"""Office / LaTeX / structured-text formats -> PDF via Pandoc (Phase 2, Chunk A).
|
|
|
|
Formats: .docx .odt .rtf .tex .latex .org .adoc .rst
|
|
|
|
Pipeline:
|
|
1. Gate: can_handle() — docx/odt/rtf match on magic MIME (extension when no
|
|
MIME is supplied); tex/latex/org/adoc/rst match on extension.
|
|
.tex/.latex additionally require the text to contain \\documentclass or
|
|
open with a % comment; anything else falls through to the generic
|
|
plain-text converter (raw source rendered as monospace <pre>).
|
|
2. Guard: docx/odt are zip containers — the same zip-slip + zip-bomb checks
|
|
as the EPUB path (app._safe_extract_zip) run against the member listing
|
|
BEFORE pandoc ever touches the file.
|
|
3. Title: docx -> docProps/core.xml <dc:title>; odt -> meta.xml
|
|
<dc:title>; else the first <h1> in the generated HTML; else the file
|
|
stem (the naming fallback in common.derive_export_name).
|
|
4. Pandoc: `pandoc <src> -f <fmt> -t html5 --standalone --embed-resources
|
|
[--metadata title=<t>] -o <out.html>` with capture_output and a 600 s
|
|
timeout. Non-zero exit -> ValueError with a concise user-safe message;
|
|
stderr is logged at DEBUG only.
|
|
5. Sanitize: <script> blocks/tags are stripped from the generated HTML
|
|
before the WeasyPrint render (defence in depth — pandoc already drops
|
|
them, but a crafted source must not reach the renderer).
|
|
6. Render + finalize: the shared WeasyPrint pipeline (common.convert_single
|
|
-> html_to_pdf_bytes) with app.BASE_CSS injected as a fallback layer
|
|
(html_converter._inject_base_css), then the shared atomic finalize +
|
|
collision naming (forced .pdf) — the same tail as every other converter.
|
|
|
|
Pandoc is a system dependency (3.x) — no new Python dependencies. A missing
|
|
binary, a timeout, or an unreadable document all surface as ValueError with
|
|
a concise human message (the shared converter error contract).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import subprocess
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from . import common
|
|
|
|
log = logging.getLogger("epub2pdf")
|
|
|
|
# extension -> pandoc reader format
|
|
_FMT: dict[str, str] = {
|
|
".docx": "docx",
|
|
".odt": "odt",
|
|
".rtf": "rtf",
|
|
".tex": "latex",
|
|
".latex": "latex",
|
|
".org": "org",
|
|
".adoc": "asciidoc",
|
|
".rst": "rst",
|
|
}
|
|
|
|
# Gate (DoD: docx/odt/rtf via magic MIME, rest via extension).
|
|
_OFFICE_MIME: dict[str, str] = {
|
|
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
".odt": "application/vnd.oasis.opendocument.text",
|
|
".rtf": "application/rtf",
|
|
}
|
|
|
|
# Title member inside each zip container (DoD: dc:title from container meta).
|
|
_TITLE_MEMBER: dict[str, str] = {
|
|
".docx": "docProps/core.xml",
|
|
".odt": "meta.xml",
|
|
}
|
|
|
|
_DC_TITLE_RE = re.compile(r"<dc:title[^>]*>(.*?)</dc:title>",
|
|
re.IGNORECASE | re.DOTALL)
|
|
_FIRST_H1_RE = re.compile(r"<h1[^>]*>(.*?)</h1>", re.IGNORECASE | re.DOTALL)
|
|
_SCRIPT_BLOCK_RE = re.compile(r"<script\b.*?</script\s*>",
|
|
re.IGNORECASE | re.DOTALL)
|
|
_SCRIPT_TAG_RE = re.compile(r"<script\b[^>]*>", re.IGNORECASE)
|
|
_TEX_DOC_RE = re.compile(r"\\documentclass")
|
|
|
|
PANDOC_TIMEOUT = 600 # s; large office documents with many embedded images
|
|
|
|
|
|
def can_handle(mime: str | None, filename: str) -> bool:
|
|
"""True if this converter claims (mime, filename).
|
|
|
|
docx/odt/rtf: magic-MIME match when a MIME type is supplied (a docx/odt
|
|
is a zip, so sniffing outranks the name); extension fallback otherwise.
|
|
tex/latex/org/adoc/rst: extension-only — the tex/latex \\documentclass
|
|
content gate runs in convert() where the bytes are available, and a
|
|
miss falls through to the plain-text converter.
|
|
"""
|
|
if not filename:
|
|
return False
|
|
ext = Path(filename).suffix.lower()
|
|
if ext in _OFFICE_MIME:
|
|
if mime:
|
|
return mime.lower() == _OFFICE_MIME[ext]
|
|
return True
|
|
return ext in _FMT
|
|
|
|
|
|
def _validate_office_zip(path: Path) -> None:
|
|
"""Zip-slip + zip-bomb pre-pass for docx/odt (DoD: EPUB guards, pre-pandoc).
|
|
|
|
Mirrors the guard class in app._safe_extract_zip (finding: zip bomb /
|
|
path escape via zip members). Pandoc reads the container itself, so we
|
|
validate the member listing without extracting: same caps, same
|
|
unsafe-member rejections, no filesystem side effects.
|
|
"""
|
|
import app # lazy: ZIP_BOMB_MIN_CAP lives with the EPUB pipeline
|
|
|
|
upload_size = path.stat().st_size
|
|
cap = max(8 * upload_size, app.ZIP_BOMB_MIN_CAP)
|
|
try:
|
|
with zipfile.ZipFile(path) as zf:
|
|
infos = zf.infolist()
|
|
except zipfile.BadZipFile:
|
|
raise ValueError(
|
|
"Not a valid document: file is not a readable zip container"
|
|
) from None
|
|
total = sum(i.file_size for i in infos)
|
|
if total > cap:
|
|
raise ValueError(
|
|
f"zip-bomb guard: uncompressed size {total} exceeds cap {cap}"
|
|
)
|
|
root = path.parent.resolve()
|
|
for info in infos:
|
|
name = info.filename
|
|
if info.file_size > app.ZIP_BOMB_MIN_CAP:
|
|
raise ValueError(
|
|
f"zip-bomb guard: member {name} is {info.file_size} bytes "
|
|
f"(max member size {app.ZIP_BOMB_MIN_CAP})"
|
|
)
|
|
if ((info.external_attr >> 16) & 0xF000) == 0xA000: # S_IFLNK
|
|
raise ValueError(f"unsafe zip member (symlink): {name}")
|
|
if name.startswith("/") or name.startswith("\\"):
|
|
raise ValueError(f"unsafe zip member (absolute): {name}")
|
|
parts = name.replace("\\", "/").split("/")
|
|
if ".." in parts:
|
|
raise ValueError(f"unsafe zip member (dotdot): {name}")
|
|
target = (root / name).resolve()
|
|
if not target.is_relative_to(root):
|
|
raise ValueError(f"unsafe zip member (escapes): {info.filename}")
|
|
|
|
|
|
def _zip_title(path: Path, member: str) -> str | None:
|
|
"""Read <dc:title> from a zip-container member (core.xml / meta.xml)."""
|
|
try:
|
|
with zipfile.ZipFile(path) as zf:
|
|
raw = zf.read(member)
|
|
except (KeyError, zipfile.BadZipFile, OSError):
|
|
return None
|
|
m = _DC_TITLE_RE.search(raw.decode("utf-8", errors="replace"))
|
|
if not m:
|
|
return None
|
|
title = re.sub(r"<[^>]+>", "", m.group(1))
|
|
return " ".join(title.split()) or None
|
|
|
|
|
|
def _strip_scripts(html_text: str) -> str:
|
|
"""Remove <script> blocks and orphan tags before the WeasyPrint render.
|
|
|
|
Defence in depth (DoD: strip any <script>): pandoc already drops
|
|
scripts, but a crafted source document must not reach the renderer.
|
|
"""
|
|
return _SCRIPT_TAG_RE.sub("", _SCRIPT_BLOCK_RE.sub("", html_text))
|
|
|
|
|
|
def _first_h1(html_text: str) -> str | None:
|
|
"""First <h1> in generated HTML -> title (DoD title chain, third step)."""
|
|
m = _FIRST_H1_RE.search(html_text)
|
|
if not m:
|
|
return None
|
|
title = re.sub(r"<[^>]+>", "", m.group(1))
|
|
return " ".join(title.split()) or None
|
|
|
|
|
|
def _run_pandoc(src: Path, fmt: str, title: str | None, out_html: Path) -> None:
|
|
"""Run pandoc to standalone HTML5; raise user-safe ValueError on failure."""
|
|
cmd = ["pandoc", str(src), "-f", fmt, "-t", "html5",
|
|
"--standalone", "--embed-resources"]
|
|
if title:
|
|
cmd += ["--metadata", f"title={title}"]
|
|
cmd += ["-o", str(out_html)]
|
|
try:
|
|
proc = subprocess.run(cmd, capture_output=True, timeout=PANDOC_TIMEOUT)
|
|
except FileNotFoundError:
|
|
raise ValueError(
|
|
"Pandoc is not installed on the server; "
|
|
"cannot convert this format"
|
|
) from None
|
|
except subprocess.TimeoutExpired:
|
|
raise ValueError(
|
|
f"Conversion timed out after {PANDOC_TIMEOUT}s "
|
|
"(pandoc did not finish)"
|
|
) from None
|
|
if proc.returncode != 0:
|
|
# Concise user-safe message; stderr at DEBUG only (no leak of paths,
|
|
# template internals, or document content into job errors).
|
|
log.debug("pandoc failed rc=%s stderr=%s",
|
|
proc.returncode, proc.stderr[:2000])
|
|
raise ValueError(
|
|
"Pandoc could not read this document "
|
|
"(it may be corrupted or password-protected)"
|
|
)
|
|
|
|
|
|
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
|
|
source_name=None) -> str:
|
|
"""Convert a pandoc-supported document to a typeset PDF.
|
|
|
|
Signature mirrors the dispatcher contract in converters/__init__.py.
|
|
Raises ValueError with a human message on any failure (shared contract).
|
|
"""
|
|
ext = path.suffix.lower()
|
|
if not can_handle(None, path.name):
|
|
raise ValueError(f"Unsupported file type: '{ext or '<none>'}'")
|
|
|
|
# .tex/.latex content gate: a real TeX document carries \\documentclass or
|
|
# opens with a % comment; anything else falls through to the generic
|
|
# plain-text converter (DoD: fall-through, no shadowing of text handling).
|
|
if ext in (".tex", ".latex"):
|
|
tex_text = common.decode_text(path.read_bytes())
|
|
if tex_text.startswith("\ufeff"):
|
|
tex_text = tex_text[1:]
|
|
if (not _TEX_DOC_RE.search(tex_text)
|
|
and not tex_text.lstrip().startswith("%")):
|
|
from . import text_converter # generic text converter fallback
|
|
return text_converter.convert(
|
|
path, job_id, progress_cb=progress_cb, output_dir=output_dir,
|
|
source_name=source_name,
|
|
)
|
|
|
|
# docx/odt are zip containers: EPUB-class guards run BEFORE pandoc.
|
|
if ext in _TITLE_MEMBER:
|
|
_validate_office_zip(path)
|
|
|
|
import app # lazy: WORK_DIR/BASE_CSS live with the EPUB pipeline
|
|
from .html_converter import _inject_base_css # shared render fallback CSS
|
|
|
|
common.report(progress_cb, 10.0, "Converting with Pandoc\u2026")
|
|
|
|
fmt = _FMT[ext]
|
|
title = _zip_title(path, _TITLE_MEMBER[ext]) if ext in _TITLE_MEMBER else None
|
|
|
|
work = app.WORK_DIR / f"job-{job_id}"
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
out_html = work / f"{path.name}.html"
|
|
try:
|
|
_run_pandoc(path, fmt, title, out_html)
|
|
html_text = common.decode_declared(out_html.read_bytes())
|
|
finally:
|
|
out_html.unlink(missing_ok=True)
|
|
|
|
if not title:
|
|
title = _first_h1(html_text)
|
|
|
|
html_text = _strip_scripts(html_text)
|
|
html_doc = _inject_base_css(html_text)
|
|
|
|
def _build():
|
|
# Reuses the shared WeasyPrint render (common.convert_single ->
|
|
# html_to_pdf_bytes) and its progress stages, same tail as the
|
|
# HTML converter: atomic finalize + collision naming, forced .pdf.
|
|
return html_doc, path.name
|
|
|
|
return common.convert_single(
|
|
path, job_id, build_html=_build, title=title, css="",
|
|
progress_cb=progress_cb, output_dir=output_dir,
|
|
name_src=source_name or path.name,
|
|
)
|