epub2pdf/converters/pandoc_converter.py

"""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}"
        )
    # C1 (real enforcement): stream every member's DECOMPRESSED bytes through the
    # shared counter (dst=None -> count-only) so a container that inflates a small
    # header into a huge stream is stopped mid-read, before pandoc ever opens it.
    # The header check above is defense-in-depth; this counts actual bytes and
    # mirrors _safe_extract_zip's streaming guard.
    with zipfile.ZipFile(path) as zf:
        for info in zf.infolist():
            if info.is_dir():
                continue
            with zf.open(info) as member:
                common._copy_stream_with_cap(member, None, 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,
    )