any-file to PDF (Phase 2, Chunk A): pandoc converters for docx/odt/rtf/tex/latex/org/adoc/rst, 27 formats, 126 tests green
This commit is contained in:
sirdave73
@@ -17,6 +17,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
|
||||
fontconfig \
|
||||
fonts-dejavu-core \
|
||||
fonts-noto-cjk \
|
||||
pandoc \
|
||||
&& rm -rf /var/lib/apt/lists/* && fc-cache -f
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
+11
-7
@@ -1,7 +1,8 @@
|
||||
# Document → PDF Converter — User Guide
|
||||
|
||||
A small, self-hosted web app that converts common document formats —
|
||||
EPUB3 ebooks, Markdown, HTML, plain text, CSV/TSV, JSON, XML, SVG,
|
||||
EPUB3 ebooks, office documents (DOCX/ODT/RTF), TeX/LaTeX, Org,
|
||||
AsciiDoc, RST, Markdown, HTML, plain text, CSV/TSV, JSON, XML, SVG,
|
||||
images, and existing PDFs — into PDFs.
|
||||
Everything runs locally (or on your own server); **no file ever leaves your machine**.
|
||||
|
||||
@@ -14,15 +15,18 @@ Everything runs locally (or on your own server); **no file ever leaves your mach
|
||||
| Preserves the EPUB's own CSS, images, headings, tables, code blocks | Re-flow or re-typeset like a print publisher would |
|
||||
| Renders CJK (Chinese / Japanese / Korean) correctly via Noto fonts | Handle DRM-protected (e.g. protected) EPUBs |
|
||||
| Supports right-to-left books (Arabic, Hebrew) | Convert *to* other formats (only PDF out) |
|
||||
| Keeps book title/author as PDF metadata | Office formats (DOCX, PPTX, XLSX) — planned, not yet |
|
||||
| Keeps book title/author as PDF metadata | PPTX, XLSX and other Office formats — planned, not yet |
|
||||
|
||||
**Supported input formats (18):**
|
||||
**Supported input formats (27):**
|
||||
|
||||
| Family | Extensions | Notes |
|
||||
|---|---|---|
|
||||
| Ebooks | `.epub` | Full CSS / images / metadata fidelity (the original mode) |
|
||||
| Web | `.html` `.htm` | Native WeasyPrint rendering; `<script>` tags stripped |
|
||||
| Markup | `.md` `.markdown` | Headers, tables, fenced code, links |
|
||||
| Office | `.docx` `.odt` `.rtf` | Via pandoc; title from document metadata (DOCX/ODT) or the H1 |
|
||||
| TeX / LaTeX | `.tex` `.latex` | Via pandoc; needs `\documentclass` or a leading `%`, else treated as plain text |
|
||||
| Extended markup | `.org` `.adoc` `.rst` | Org mode, AsciiDoc, reStructuredText via pandoc |
|
||||
| Text | `.txt` `.log` | UTF-8 safe, paginated |
|
||||
| Data | `.csv` `.tsv` | Auto-detected delimiter → styled table; max 5000 rows × 50 cols |
|
||||
| Structured | `.json` `.xml` | Cleanly formatted listing |
|
||||
@@ -44,8 +48,8 @@ another in the same server process), max upload **400 MB**, max output **200 MB*
|
||||
### Option A — Run directly (no Docker)
|
||||
```bash
|
||||
# 1. install dependencies (Python 3.11+; WeasyPrint also needs the pango
|
||||
# system libraries, e.g. on Debian/Ubuntu: sudo apt install libpango-1.0-0
|
||||
# libpangoft2-1.0-0 libharfbuzz0b fontconfig fonts-noto-cjk shared-mime-info)
|
||||
# system libraries; pandoc for office/TeX/markup formats — e.g. on Debian/Ubuntu: sudo apt install libpango-1.0-0
|
||||
# libpangoft2-1.0-0 libharfbuzz0b fontconfig fonts-noto-cjk shared-mime-info pandoc)
|
||||
pip install -r requirements.txt
|
||||
|
||||
# 2. start the web server
|
||||
@@ -71,7 +75,7 @@ See `COOLIFY.md` in this directory (3 steps: Dockerfile service, port 8030, volu
|
||||
|
||||
1. Open `http://<host>:8030` in your browser.
|
||||
2. **Drop any supported file** onto the upload box (or click to browse).
|
||||
- The 18 formats in §1 are accepted; anything else is rejected with a message.
|
||||
- The 27 formats in §1 are accepted; anything else is rejected with a message.
|
||||
- Files over 400 MB are rejected before upload finishes.
|
||||
3. The job starts immediately — progress updates show the current stage
|
||||
(e.g. `Rendering chapter-04.xhtml (4/12)` for EPUBs), then `Done`.
|
||||
@@ -89,7 +93,7 @@ See `COOLIFY.md` in this directory (3 steps: Dockerfile service, port 8030, volu
|
||||
| `No readable content found in EPUB spine` | The EPUB has no HTML chapters (images-only books, broken exports) |
|
||||
| `DRM protected` style errors | Remove DRM first (Calibre / Calibre-DB tools) — this app can't decrypt |
|
||||
| Stuck at `Working…` forever | The server restarted mid-job; refresh the page and re-upload |
|
||||
| `Unsupported format` | File isn't one of the 18 supported types — see §1 |
|
||||
| `Unsupported format` | File isn't one of the 27 supported types — see §1 |
|
||||
| CSV/TSV rejected | Over 5000 rows or 50 columns — split the file |
|
||||
| PDF rejected | Encrypted/password-protected, or over 500 pages |
|
||||
|
||||
|
||||
@@ -52,7 +52,8 @@ MAX_UPLOAD_MB = 400
|
||||
# existing /health contract asserts only status + max_upload_mb.
|
||||
SUPPORTED_FORMATS = ["epub", "html", "htm", "md", "markdown", "txt", "log",
|
||||
"csv", "tsv", "json", "xml", "svg", "png", "jpg",
|
||||
"jpeg", "webp", "tiff", "bmp", "pdf"]
|
||||
"jpeg", "webp", "tiff", "bmp", "pdf",
|
||||
"docx", "odt", "rtf", "tex", "latex", "org", "adoc", "rst"]
|
||||
|
||||
# ZIP-bomb guard: total uncompressed size must stay within 8x the upload
|
||||
# size, with a 2 GiB floor so normal books always fit (finding: zip bomb,
|
||||
|
||||
@@ -40,6 +40,16 @@ _ROUTING: dict[str, str] = {
|
||||
".tiff": "image_converter",
|
||||
".bmp": "image_converter",
|
||||
".pdf": "pdf_passthrough",
|
||||
# PHASE 2 (Chunk A): office / LaTeX / structured text via Pandoc ->
|
||||
# standalone HTML5 -> the shared WeasyPrint render (pandoc_converter).
|
||||
".docx": "pandoc_converter",
|
||||
".odt": "pandoc_converter",
|
||||
".rtf": "pandoc_converter",
|
||||
".tex": "pandoc_converter",
|
||||
".latex": "pandoc_converter",
|
||||
".org": "pandoc_converter",
|
||||
".adoc": "pandoc_converter",
|
||||
".rst": "pandoc_converter",
|
||||
}
|
||||
|
||||
# .epub is supported (existing pipeline) but routes differently.
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
"""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,
|
||||
)
|
||||
@@ -136,10 +136,10 @@
|
||||
</div>
|
||||
<div class="drop-copy">
|
||||
<h1>Drop a file to set it in type</h1>
|
||||
<p>or <span class="browse">browse your files</span> · up to <span id="maxmb">400</span> MB · <code>.epub .html .md .txt .csv .json .xml .svg .png .jpg .webp …</code></p>
|
||||
<p>or <span class="browse">browse your files</span> · up to <span id="maxmb">400</span> MB · <code>.epub .docx .odt .rtf .tex .org .adoc .rst .html .md .txt .csv .json .xml .svg .png .jpg .webp …</code></p>
|
||||
</div>
|
||||
<p id="droperr" class="drop-err" role="alert"></p>
|
||||
<input type="file" id="file" accept=".epub,.html,.htm,.txt,.log,.md,.markdown,.csv,.tsv,.json,.xml,.svg,.png,.jpg,.jpeg,.webp">
|
||||
<input type="file" id="file" accept=".epub,.html,.htm,.txt,.log,.md,.markdown,.csv,.tsv,.json,.xml,.svg,.png,.jpg,.jpeg,.webp,.tiff,.bmp,.pdf,.docx,.odt,.rtf,.tex,.latex,.org,.adoc,.rst">
|
||||
</section>
|
||||
|
||||
<section id="status" class="status" hidden>
|
||||
|
||||
@@ -80,7 +80,9 @@ def test_is_supported_matrix():
|
||||
".csv", ".tsv", ".json", ".xml", ".svg", ".png", ".jpg",
|
||||
".jpeg", ".webp", ".tiff", ".bmp", ".pdf"):
|
||||
assert converters.is_supported(f"book{ext}"), ext
|
||||
for ext in (".exe", ".docx", ".rtf", ".mp4", ".zip"):
|
||||
# Phase 2 (Chunk A): .docx/.rtf are now supported (pandoc_converter) —
|
||||
# removed from the negative list; remaining entries stay rejected.
|
||||
for ext in (".exe", ".mp4", ".zip"):
|
||||
assert not converters.is_supported(f"book{ext}"), ext
|
||||
assert not converters.is_supported("noextension")
|
||||
|
||||
|
||||
@@ -0,0 +1,423 @@
|
||||
"""PHASE 2 (Chunk B): office / LaTeX / structured-text formats via Pandoc.
|
||||
|
||||
Pins the behavior of converters/pandoc_converter.py end-to-end through the
|
||||
HTTP API (Flask test client + worker threads, hermetic DATA_DIR via conftest):
|
||||
|
||||
* all 8 formats e2e -> job done, correct export name, real %PDF, content
|
||||
* title chain: docx from docProps/core.xml dc:title, odt from meta.xml
|
||||
dc:title, tex/org/adoc/rst from the <h1> pandoc promotes from the
|
||||
document's own title (\title / #+TITLE / '= Title' / overline)
|
||||
* embedded image (PNG) survives the pandoc -> html5 -> WeasyPrint pipeline
|
||||
* naming collision (-2 suffix) and the .tex not-latex fall-through to the
|
||||
plain-text converter
|
||||
* zip-slip / zip-bomb pre-pass + a non-zip .docx all surface as clean job
|
||||
errors (202 -> error), never an HTTP 500
|
||||
* /health advertises the 8 new formats; is_supported() accepts them
|
||||
|
||||
Pandoc-generated fixtures (docx/odt/rtf) are built ONCE per module run from a
|
||||
small markdown source. Pandoc is a hard prerequisite for this module (3.x is
|
||||
a project system dep); if the binary is absent the whole module skips.
|
||||
"""
|
||||
import io
|
||||
import re
|
||||
import shutil
|
||||
import struct
|
||||
import subprocess
|
||||
import tempfile
|
||||
import urllib.parse
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
import converters
|
||||
from tests.helpers import (download_bytes, pdf_page_count, pdf_text,
|
||||
wait_for_job)
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# skip guard: pandoc is a system dependency of this converter family
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
_pandoc_bin = shutil.which("pandoc")
|
||||
pytestmark = pytest.mark.skipif(_pandoc_bin is None, reason="pandoc not installed")
|
||||
|
||||
# Extension matrix for the pandoc converter family.
|
||||
PANDOC_EXTS = (".docx", ".odt", ".rtf", ".tex", ".latex", ".org", ".adoc", ".rst")
|
||||
PANDOC_BARE = ("docx", "odt", "rtf", "tex", "latex", "org", "adoc", "rst")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# pure builders (zip mutation + pandoc invocation) — no fixtures, testable
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run_pandoc(src: Path, out: Path) -> None:
|
||||
proc = subprocess.run(["pandoc", str(src), "-o", str(out)],
|
||||
capture_output=True, text=True)
|
||||
assert proc.returncode == 0, f"pandoc failed for {src.name}: {proc.stderr}"
|
||||
|
||||
|
||||
def _png_bytes(size=(32, 32), color=(255, 0, 0)) -> bytes:
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", size, color).save(buf, format="PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _rewrite_zip(raw: bytes, edits: dict) -> bytes:
|
||||
"""Rebuild a zip from `raw`, replacing the bytes of each `edits` member."""
|
||||
with zipfile.ZipFile(io.BytesIO(raw)) as src:
|
||||
items = {i.filename: src.read(i.filename) for i in src.infolist()}
|
||||
for name, new in edits.items():
|
||||
items[name] = new
|
||||
out = io.BytesIO()
|
||||
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as dst:
|
||||
for name, data in items.items():
|
||||
dst.writestr(name, data)
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
def _stamp_dc_title(raw: bytes, member: str, title: str) -> bytes:
|
||||
"""Set <dc:title> inside a zip member (core.xml / meta.xml) to `title`."""
|
||||
with zipfile.ZipFile(io.BytesIO(raw)) as z:
|
||||
cur = z.read(member).decode("utf-8", errors="replace")
|
||||
new = re.sub(r"<dc:title[^>]*>.*?</dc:title>",
|
||||
f"<dc:title>{title}</dc:title>", cur, flags=re.S)
|
||||
return _rewrite_zip(raw, {member: new.encode("utf-8")})
|
||||
|
||||
|
||||
def _forge_zip_bomb(raw: bytes) -> bytes:
|
||||
"""Inflate the first central-directory member's declared uncompressed
|
||||
size to 3 GiB so the zip-bomb guard total exceeds its 2 GiB cap.
|
||||
|
||||
zipfile.writestr recomputes sizes on write, so the central-directory
|
||||
field is patched in place (offset 24 of the CD file header).
|
||||
"""
|
||||
data = bytearray(raw)
|
||||
pos = data.find(b"PK\x01\x02")
|
||||
assert pos >= 0, "no central directory found in fixture zip"
|
||||
struct.pack_into("<I", data, pos + 24, 3 * 1024 ** 3)
|
||||
return bytes(data)
|
||||
|
||||
|
||||
def _forge_zip_slip(raw: bytes) -> bytes:
|
||||
"""Copy a valid office zip and add a dotdot member escaping the root."""
|
||||
out = io.BytesIO()
|
||||
with zipfile.ZipFile(io.BytesIO(raw)) as src, \
|
||||
zipfile.ZipFile(out, "w") as dst:
|
||||
for i in src.infolist():
|
||||
dst.writestr(i.filename, src.read(i.filename))
|
||||
dst.writestr("../../evil.txt", "pwn")
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
def _build_all(root: Path) -> dict:
|
||||
"""Build every pandoc fixture once; return {key: bytes}.
|
||||
|
||||
The PNG is written BEFORE pandoc runs: pandoc embeds by path, and a
|
||||
missing file is silently dropped (PandocResourceNotFound), which would
|
||||
make the embedded-image tests prove nothing.
|
||||
"""
|
||||
(root / "tiny.png").write_bytes(_png_bytes())
|
||||
png = str(root / "tiny.png") # absolute path — required for embedding
|
||||
|
||||
def md(name: str, text: str) -> None:
|
||||
(root / name).write_text(text, encoding="utf-8")
|
||||
|
||||
# Office trio: distinct H1 titles + a per-format ALL-CAPS marker.
|
||||
md("x.md", "# Pandoc DocX Alpha\n\nBody X with **BOLDX** and a code span.\n")
|
||||
md("b.md", "# Pandoc Odt Bravo\n\nBody B with BOLD_B text.\n")
|
||||
md("c.md", "# Pandoc Rtf Charlie\n\nBody C with BOLD_C text.\n")
|
||||
_run_pandoc(root / "x.md", root / "x.docx")
|
||||
_run_pandoc(root / "b.md", root / "b.odt")
|
||||
_run_pandoc(root / "c.md", root / "c.rtf")
|
||||
|
||||
# Embedded-image pair (PNG embedded by pandoc at build time).
|
||||
md("ix.md", f"# Embedded Image Doc\n\n\n\nIMGXMARK body.\n")
|
||||
md("io.md", f"# Embedded Image Odt\n\n\n\nIMGOMARK body.\n")
|
||||
_run_pandoc(root / "ix.md", root / "ix.docx")
|
||||
_run_pandoc(root / "io.md", root / "io.odt")
|
||||
|
||||
# Structured text formats (read directly by pandoc, no zip container).
|
||||
(root / "sample.tex").write_text(
|
||||
"\\documentclass{article}\n\\title{Pandoc TeX Delta}\n"
|
||||
"\\begin{document}\nTTEXMARK body.\n\\end{document}\n",
|
||||
encoding="utf-8")
|
||||
(root / "sample.latex").write_text(
|
||||
"\\documentclass{article}\n\\title{Pandoc LaTeX Echo}\n"
|
||||
"\\begin{document}\nLATEXMARK body.\n\\end{document}\n",
|
||||
encoding="utf-8")
|
||||
(root / "sample.org").write_text(
|
||||
"#+TITLE: Pandoc Org Foxtrot\n\n* A Heading\n\n"
|
||||
"Org body ORGMARK here.\n", encoding="utf-8")
|
||||
(root / "sample.adoc").write_text(
|
||||
"= Pandoc Adoc Golf\n\nAdoc body ADOCMARK here.\n", encoding="utf-8")
|
||||
(root / "sample.rst").write_text(
|
||||
"Pandoc Rst Hotel\n================\n\nRst body RSTMARK here.\n",
|
||||
encoding="utf-8")
|
||||
|
||||
# Fall-through: a .tex with no \\documentclass and no leading %.
|
||||
(root / "note.tex").write_text(
|
||||
"just a note\nnothing special here\n", encoding="utf-8")
|
||||
|
||||
# Collision: an org whose #+TITLE is 'Pandoc Collie'.
|
||||
(root / "collide.org").write_text(
|
||||
"#+TITLE: Pandoc Collie\n\nCollision body.\n", encoding="utf-8")
|
||||
|
||||
docx = (root / "x.docx").read_bytes()
|
||||
odt = (root / "b.odt").read_bytes()
|
||||
|
||||
# Title-chain fixtures: stamp a dc:title that DIFFERS from the body H1 so
|
||||
# the export name proves which source the converter actually read
|
||||
# (pandoc leaves core.xml / meta.xml dc:title empty; the stamp is non-empty).
|
||||
docx_core = _stamp_dc_title(docx, "docProps/core.xml", "CoreX Wins")
|
||||
odt_meta = _stamp_dc_title(odt, "meta.xml", "MetaX Wins")
|
||||
|
||||
# Security fixtures forged from a real, small docx.
|
||||
return {
|
||||
"docx": docx,
|
||||
"odt": odt,
|
||||
"rtf": (root / "c.rtf").read_bytes(),
|
||||
"tex": (root / "sample.tex").read_bytes(),
|
||||
"latex": (root / "sample.latex").read_bytes(),
|
||||
"org": (root / "sample.org").read_bytes(),
|
||||
"adoc": (root / "sample.adoc").read_bytes(),
|
||||
"rst": (root / "sample.rst").read_bytes(),
|
||||
"note_tex": (root / "note.tex").read_bytes(),
|
||||
"collide_org": (root / "collide.org").read_bytes(),
|
||||
"docx_core": docx_core,
|
||||
"odt_meta": odt_meta,
|
||||
"img_docx": (root / "ix.docx").read_bytes(),
|
||||
"img_odt": (root / "io.odt").read_bytes(),
|
||||
"bomb_docx": _forge_zip_bomb(docx),
|
||||
"slip_docx": _forge_zip_slip(docx),
|
||||
"fake_docx": b"\x89\x50\x4e\x47\x0d\x0a\x1a\x0a definitely not a docx",
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# fixtures
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def pandoc_fixtures():
|
||||
"""Build the pandoc fixture set once; return {key: bytes}."""
|
||||
root = Path(tempfile.mkdtemp(prefix="pandoc-fix-"))
|
||||
fx = _build_all(root)
|
||||
# Self-check: the image fixtures must really carry a PNG, else the
|
||||
# embedded-image tests would prove nothing (pandoc drops missing files).
|
||||
for key in ("img_docx", "img_odt"):
|
||||
with zipfile.ZipFile(io.BytesIO(fx[key])) as z:
|
||||
assert any(i.filename.lower().endswith(".png")
|
||||
for i in z.infolist()), f"{key}: PNG not embedded"
|
||||
return fx
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# e2e helpers (mirror tests/test_multi_format.py)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _submit(client, filename, content):
|
||||
if isinstance(content, str):
|
||||
content = content.encode("utf-8")
|
||||
files = {"file": (io.BytesIO(content), filename)}
|
||||
return client.post("/api/convert", data=files, content_type="multipart/form-data")
|
||||
|
||||
|
||||
def _e2e(client, filename, content):
|
||||
"""Submit, wait for terminal, assert done, return (job, pdf_bytes, pdf_text)."""
|
||||
resp = _submit(client, filename, content)
|
||||
assert resp.status_code == 202, resp.get_json()
|
||||
job = wait_for_job(client, resp.get_json()["job_id"])
|
||||
assert job["status"] == "done", job
|
||||
data = download_bytes(client, job["filename"])
|
||||
assert data[:4] == b"%PDF"
|
||||
return job, data, pdf_text(data)
|
||||
|
||||
|
||||
def _error_job(client, filename, content):
|
||||
"""Submit, expect 202 then a terminal error state; return the snapshot."""
|
||||
resp = _submit(client, filename, content)
|
||||
assert resp.status_code == 202, resp.get_json()
|
||||
job = wait_for_job(client, resp.get_json()["job_id"])
|
||||
assert job["status"] == "error", job
|
||||
return job
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# is_supported + /health advertisement
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_is_supported_pandoc_formats():
|
||||
for ext in (".docx", ".odt", ".rtf", ".tex", ".latex", ".org", ".adoc", ".rst"):
|
||||
assert converters.is_supported(f"book{ext}"), ext
|
||||
|
||||
|
||||
def test_health_lists_pandoc_formats(client):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
formats = resp.get_json()["supported_formats"]
|
||||
for bare in ("docx", "odt", "rtf", "tex", "latex", "org", "adoc", "rst"):
|
||||
assert bare in formats, f"{bare} missing from /health supported_formats"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# per-format e2e happy paths (all 8) — filename + content marker
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_docx_e2e_title_and_content(client, pandoc_fixtures):
|
||||
# Pandoc leaves core.xml dc:title empty, so the name comes from the first
|
||||
# <h1> pandoc promotes from the markdown source.
|
||||
job, data, text = _e2e(client, "alpha.docx", pandoc_fixtures["docx"])
|
||||
assert job["filename"] == "Pandoc DocX Alpha.pdf"
|
||||
assert "BOLDX" in text
|
||||
|
||||
|
||||
def test_odt_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "bravo.odt", pandoc_fixtures["odt"])
|
||||
assert job["filename"] == "Pandoc Odt Bravo.pdf"
|
||||
assert "BOLD_B" in text
|
||||
|
||||
|
||||
def test_rtf_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "charlie.rtf", pandoc_fixtures["rtf"])
|
||||
assert job["filename"] == "Pandoc Rtf Charlie.pdf"
|
||||
assert "BOLD_C" in text
|
||||
|
||||
|
||||
def test_tex_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "delta.tex", pandoc_fixtures["tex"])
|
||||
assert job["filename"] == "Pandoc TeX Delta.pdf"
|
||||
assert "TTEXMARK" in text
|
||||
|
||||
|
||||
def test_latex_e2e_title_and_content(client, pandoc_fixtures):
|
||||
# .latex is the same pandoc reader as .tex — extension-parity check.
|
||||
job, data, text = _e2e(client, "echo.latex", pandoc_fixtures["latex"])
|
||||
assert job["filename"] == "Pandoc LaTeX Echo.pdf"
|
||||
assert "LATEXMARK" in text
|
||||
|
||||
|
||||
def test_org_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "foxtrot.org", pandoc_fixtures["org"])
|
||||
assert job["filename"] == "Pandoc Org Foxtrot.pdf"
|
||||
assert "ORGMARK" in text
|
||||
|
||||
|
||||
def test_adoc_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "golf.adoc", pandoc_fixtures["adoc"])
|
||||
assert job["filename"] == "Pandoc Adoc Golf.pdf"
|
||||
assert "ADOCMARK" in text
|
||||
|
||||
|
||||
def test_rst_e2e_title_and_content(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "hotel.rst", pandoc_fixtures["rst"])
|
||||
assert job["filename"] == "Pandoc Rst Hotel.pdf"
|
||||
assert "RSTMARK" in text
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# title chain: zip dc:title beats the body <h1>
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_docx_title_from_core_xml(client, pandoc_fixtures):
|
||||
# A dc:title stamped into docProps/core.xml wins over the body H1
|
||||
# ('Pandoc DocX Alpha'), proving the core.xml step of the chain runs.
|
||||
job, data, _ = _e2e(client, "corex.docx", pandoc_fixtures["docx_core"])
|
||||
assert job["filename"] == "CoreX Wins.pdf"
|
||||
|
||||
|
||||
def test_odt_title_from_meta_xml(client, pandoc_fixtures):
|
||||
# A dc:title stamped into meta.xml wins over the body H1 ('Pandoc Odt Bravo').
|
||||
job, data, _ = _e2e(client, "metax.odt", pandoc_fixtures["odt_meta"])
|
||||
assert job["filename"] == "MetaX Wins.pdf"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# embedded image: PNG survives pandoc -> html5 -> WeasyPrint into the PDF
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_docx_embedded_image_reaches_pdf(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "imaged.docx", pandoc_fixtures["img_docx"])
|
||||
assert job["filename"] == "Embedded Image Doc.pdf"
|
||||
assert pdf_page_count(data) >= 1
|
||||
# The PNG is an actual XObject on a page (probe-verified: 1 image/page 0).
|
||||
from pypdf import PdfReader
|
||||
pages = PdfReader(io.BytesIO(data)).pages
|
||||
assert any(len(pg.images) >= 1 for pg in pages), "no image in PDF pages"
|
||||
assert "IMGXMARK" in text
|
||||
|
||||
|
||||
def test_odt_embedded_image_reaches_pdf(client, pandoc_fixtures):
|
||||
job, data, text = _e2e(client, "imageo.odt", pandoc_fixtures["img_odt"])
|
||||
assert job["filename"] == "Embedded Image Odt.pdf"
|
||||
assert pdf_page_count(data) >= 1
|
||||
from pypdf import PdfReader
|
||||
pages = PdfReader(io.BytesIO(data)).pages
|
||||
assert any(len(pg.images) >= 1 for pg in pages), "no image in PDF pages"
|
||||
assert "IMGOMARK" in text
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# naming: collision suffix + .tex not-latex fall-through
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_collision_second_upload_gets_suffix(client, pandoc_fixtures):
|
||||
job1, _, _ = _e2e(client, "collide.org", pandoc_fixtures["collide_org"])
|
||||
job2, _, _ = _e2e(client, "collide.org", pandoc_fixtures["collide_org"])
|
||||
assert job1["filename"] == "Pandoc Collie.pdf"
|
||||
assert job2["filename"] == "Pandoc Collie-2.pdf"
|
||||
|
||||
|
||||
def test_tex_not_latex_falls_through_to_text(client, pandoc_fixtures):
|
||||
# No \\documentclass and no leading % -> plain-text converter: named by
|
||||
# stem, raw source text rendered, no error.
|
||||
job, data, text = _e2e(client, "note.tex", pandoc_fixtures["note_tex"])
|
||||
assert job["filename"] == "note.pdf"
|
||||
assert "just a note" in text
|
||||
assert "nothing special here" in text
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# download headers (family representative)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_download_headers_for_pandoc_conversion(client, pandoc_fixtures):
|
||||
job, _, _ = _e2e(client, "alpha.docx", pandoc_fixtures["docx"])
|
||||
# URL-encode the filename (it contains spaces) for the direct GET.
|
||||
resp = client.get("/download/" + urllib.parse.quote(job["filename"]))
|
||||
assert resp.status_code == 200
|
||||
assert resp.content_type == "application/pdf"
|
||||
cd = resp.headers["Content-Disposition"]
|
||||
assert "attachment" in cd
|
||||
assert f'filename="{job["filename"]}"' in cd
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# security: zip-bomb / zip-slip / non-zip .docx -> clean job error, not 500
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_zip_bomb_docx_rejected(client, pandoc_fixtures):
|
||||
# Declared uncompressed size inflated past the 2 GiB cap -> job error.
|
||||
job = _error_job(client, "bomb.docx", pandoc_fixtures["bomb_docx"])
|
||||
assert "zip-bomb" in job["error"].lower()
|
||||
|
||||
|
||||
def test_zip_slip_docx_rejected(client, pandoc_fixtures):
|
||||
job = _error_job(client, "slip.docx", pandoc_fixtures["slip_docx"])
|
||||
assert "unsafe zip member" in job["error"]
|
||||
|
||||
|
||||
def test_non_zip_docx_rejected_not_500(client, pandoc_fixtures):
|
||||
# Random bytes with a .docx name: clean job error, never an HTTP 500.
|
||||
job = _error_job(client, "fake.docx", pandoc_fixtures["fake_docx"])
|
||||
assert job["error"], "expected a non-empty error message"
|
||||
Reference in New Issue
Block a user