any-file to PDF (Phase 1): converters/ dispatcher, images/text/md/csv/json/xml/svg/pdf-passthrough, 112 tests green

This commit is contained in:
sirdave73
2026-09-02 01:53:24 -04:00
parent 0b575a3e1f
commit 345c39417b
17 changed files with 1306 additions and 17 deletions
+1
View File
@@ -26,6 +26,7 @@ RUN pip install --no-cache-dir -r requirements.txt
COPY app.py . COPY app.py .
COPY templates/ ./templates/ COPY templates/ ./templates/
COPY converters/ ./converters/
# Persistent data (uploads/output/work) — mount a volume at /app/data # Persistent data (uploads/output/work) — mount a volume at /app/data
VOLUME ["/app/data"] VOLUME ["/app/data"]
+39 -10
View File
@@ -34,6 +34,8 @@ from werkzeug.exceptions import HTTPException, RequestEntityTooLarge
from lxml import etree from lxml import etree
from pypdf import PdfWriter from pypdf import PdfWriter
import converters # PHASE 1: multi-format dispatch (any common file -> PDF)
BASE_DIR = Path(__file__).resolve().parent BASE_DIR = Path(__file__).resolve().parent
# DATA_DIR is overridable via env so container deployments (e.g. Coolify) # DATA_DIR is overridable via env so container deployments (e.g. Coolify)
# can point uploads/output/work at a persistent volume. # can point uploads/output/work at a persistent volume.
@@ -46,6 +48,12 @@ for _d in (UPLOAD_DIR, OUTPUT_DIR, WORK_DIR):
MAX_UPLOAD_MB = 400 MAX_UPLOAD_MB = 400
# req 3 (health): supported formats advertised to clients. Additive key — the
# 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"]
# ZIP-bomb guard: total uncompressed size must stay within 8x the upload # 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, # size, with a 2 GiB floor so normal books always fit (finding: zip bomb,
# app.py extractall). # app.py extractall).
@@ -127,6 +135,10 @@ body {
font-size: 10.5pt; font-size: 10.5pt;
line-height: 1.65; line-height: 1.65;
color: #1c1b1a; color: #1c1b1a;
/* WeasyPrint emits CSS kerning as positive TJ adjustments that
pypdf's extract_text misreads as inter-word spaces ("BRA VO");
disable it to keep text extraction faithful. */
font-kerning: none;
} }
h1, h2, h3, h4, h5, h6 { h1, h2, h3, h4, h5, h6 {
font-weight: 700; line-height: 1.25; color: #111; font-weight: 700; line-height: 1.25; color: #111;
@@ -526,8 +538,9 @@ def _run_job(job_id: str, epub_path: Path) -> None:
_set_job(job_id, progress=pct, message=msg) _set_job(job_id, progress=pct, message=msg)
try: try:
filename = convert_epub(epub_path, job_id, progress_cb=_cb, filename = converters.convert_file(
source_name=JOBS.get(job_id, {}).get("epub")) epub_path, job_id, progress_cb=_cb,
source_name=JOBS.get(job_id, {}).get("epub"))
_set_job(job_id, status="done", progress=100.0, message="Done", _set_job(job_id, status="done", progress=100.0, message="Done",
filename=filename, error=None) filename=filename, error=None)
except Exception as exc: # noqa: BLE001 - any failure is a job error except Exception as exc: # noqa: BLE001 - any failure is a job error
@@ -539,7 +552,10 @@ def _run_job(job_id: str, epub_path: Path) -> None:
except ValueError: except ValueError:
pass pass
try: try:
(UPLOAD_DIR / f"{job_id}.epub").unlink(missing_ok=True) # PHASE 1: uploads keep their real suffix ({job_id}.html, ...);
# job_id is 32-hex so the glob cannot match anything else.
for _up in UPLOAD_DIR.glob(f"{job_id}.*"):
_up.unlink(missing_ok=True)
except OSError: except OSError:
pass pass
_sweep_stale_work_dirs() _sweep_stale_work_dirs()
@@ -571,9 +587,11 @@ def health():
problems.append("weasyprint not importable") problems.append("weasyprint not importable")
if not _cjk_fonts_present(): if not _cjk_fonts_present():
problems.append("CJK fonts missing (fc-list found none)") problems.append("CJK fonts missing (fc-list found none)")
body = {"status": "ok", "max_upload_mb": MAX_UPLOAD_MB} body = {"status": "ok", "max_upload_mb": MAX_UPLOAD_MB,
"supported_formats": SUPPORTED_FORMATS}
if problems: if problems:
body = {"status": "degraded", "max_upload_mb": MAX_UPLOAD_MB, "problems": problems} body = {"status": "degraded", "max_upload_mb": MAX_UPLOAD_MB,
"supported_formats": SUPPORTED_FORMATS, "problems": problems}
return jsonify(body), 503 return jsonify(body), 503
return jsonify(body), 200 return jsonify(body), 200
@@ -583,15 +601,21 @@ def convert_upload():
up = request.files.get("file") up = request.files.get("file")
if up is None or not up.filename: if up is None or not up.filename:
return jsonify(error="No file uploaded"), 400 return jsonify(error="No file uploaded"), 400
if not up.filename.lower().endswith(".epub"): if not converters.is_supported(up.filename):
return jsonify(error="Only .epub files are accepted"), 400 # PHASE 1: gate on the full extension matrix; the error lists every
# accepted extension (incl. .epub) so the legacy 400 contract holds.
ext = Path(up.filename).suffix or "<none>"
return jsonify(error=(
f"Unsupported file type: '{ext}'. "
f"Supported: {', '.join(sorted(converters.SUPPORTED_EXTENSIONS))}"
)), 400
data = up.read() data = up.read()
if len(data) > MAX_UPLOAD_MB * 1024 * 1024: if len(data) > MAX_UPLOAD_MB * 1024 * 1024:
return jsonify(error=f"File too large (max {MAX_UPLOAD_MB} MB)"), 400 return jsonify(error=f"File too large (max {MAX_UPLOAD_MB} MB)"), 400
if not WORKER_SLOTS.acquire(blocking=False): if not WORKER_SLOTS.acquire(blocking=False):
return jsonify(error="Server busy: all conversion slots in use, retry shortly"), 503 return jsonify(error="Server busy: all conversion slots in use, retry shortly"), 503
job_id = _register_job(Path(up.filename).name) job_id = _register_job(Path(up.filename).name)
dest = UPLOAD_DIR / f"{job_id}.epub" dest = UPLOAD_DIR / f"{job_id}{Path(up.filename).suffix.lower()}"
try: try:
with open(dest, "wb") as out: with open(dest, "wb") as out:
out.write(data) out.write(data)
@@ -621,9 +645,12 @@ def download(name):
return jsonify(error="unknown file"), 404 return jsonify(error="unknown file"), 404
# req 4: explicit PDF mimetype + defensive .pdf on download_name # req 4: explicit PDF mimetype + defensive .pdf on download_name
download_name = safe if safe.lower().endswith(".pdf") else safe + ".pdf" download_name = safe if safe.lower().endswith(".pdf") else safe + ".pdf"
return send_file( resp = send_file(
target, as_attachment=True, mimetype="application/pdf", download_name=download_name target, as_attachment=True, mimetype="application/pdf", download_name=download_name
) )
# contract: quoted filename in Content-Disposition
resp.headers["Content-Disposition"] = f'attachment; filename="{download_name}"'
return resp
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
@@ -655,7 +682,9 @@ def _run_cli(argv: list[str]) -> int:
print(f"[{pct:5.1f}%] {message}") print(f"[{pct:5.1f}%] {message}")
try: try:
filename = convert_epub(src, job_id, progress_cb=_cb, output_dir=out_dir) filename = converters.convert_file(src, job_id, progress_cb=_cb,
output_dir=out_dir,
source_name=src.name)
except Exception as exc: # noqa: BLE001 - CLI reports any failure as error: except Exception as exc: # noqa: BLE001 - CLI reports any failure as error:
print(f"error: {exc}") print(f"error: {exc}")
return 1 return 1
+85
View File
@@ -0,0 +1,85 @@
"""converters — PHASE 1 multi-format dispatch (any common file -> PDF).
Public surface:
is_supported(name) -> bool (extension matrix, case-insensitive)
convert_file(path, job_id, progress_cb=None, output_dir=None,
source_name=None) -> output filename (str)
Routing is lazy per extension so an unsupported upload never pulls in an
optional dependency (e.g. PIL) it would not need. .epub routes to the
existing EPUB pipeline (app.convert_epub) — resolved lazily to keep the
import graph acyclic (app imports this package at module level).
Every converter shares one error contract: problems are ValueError with a
human message (the worker/CLI surface it verbatim as a job error / CLI
'error:' line); naming follows the pinned contract in converters.common.
"""
from __future__ import annotations
from pathlib import Path
__all__ = ["SUPPORTED_EXTENSIONS", "is_supported", "convert_file"]
# Extension -> converter module (resolved lazily inside convert_file).
_ROUTING: dict[str, str] = {
".html": "html_converter",
".htm": "html_converter",
".txt": "text_converter",
".log": "text_converter",
".md": "md_converter",
".markdown": "md_converter",
".csv": "csv_converter",
".tsv": "csv_converter",
".json": "json_converter",
".xml": "xml_converter",
".svg": "svg_converter",
".png": "image_converter",
".jpg": "image_converter",
".jpeg": "image_converter",
".webp": "image_converter",
".tiff": "image_converter",
".bmp": "image_converter",
".pdf": "pdf_passthrough",
}
# .epub is supported (existing pipeline) but routes differently.
SUPPORTED_EXTENSIONS = frozenset(_ROUTING) | {".epub"}
def is_supported(filename: str) -> bool:
"""True if `filename`'s extension is in the supported matrix.
Case-insensitive; extensionless names are unsupported.
"""
if not filename:
return False
ext = Path(filename).suffix.lower()
return ext in SUPPORTED_EXTENSIONS
def convert_file(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Convert `path` to a PDF; return the output file name (no dir).
`source_name` is the original upload/CLI file name — the on-disk name
for web uploads is `{job_id}<ext>`, so the export name must be derived
from the original name, not the temp path.
Raises ValueError for unsupported extensions and for any format-level
failure (empty/invalid content) — see each converter's docstring.
"""
ext = Path(path).suffix.lower()
if ext not in SUPPORTED_EXTENSIONS:
raise ValueError(
f"Unsupported file type: '{ext or '<none>'}'. "
f"Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
)
if ext == ".epub":
import app # lazy: the EPUB pipeline lives in app.py (avoid cycle)
return app.convert_epub(path, job_id, progress_cb=progress_cb,
output_dir=output_dir, source_name=source_name)
module_name = _ROUTING[ext]
module = __import__(f"converters.{module_name}", fromlist=["convert"])
return module.convert(Path(path), job_id, progress_cb=progress_cb,
output_dir=output_dir, source_name=source_name)
+188
View File
@@ -0,0 +1,188 @@
"""Shared building blocks for the PHASE 1 multi-format converters.
The EPUB pipeline (app.convert_epub) keeps its own inline naming code; the
logic here is an intentional mirror of that pinned contract (sanitize
[^A-Za-z0-9._ -]+ -> '_', 60-char cap, no doubled .pdf, first-free
'-2'/'-3' collision suffix) so every format exports names identically.
No module-level import of app: app.py imports converters at top level, so
app references are resolved lazily inside functions.
"""
from __future__ import annotations
import html as _html
import io
import os
import re
from pathlib import Path
__all__ = [
"PRE_CSS",
"convert_single",
"decode_declared",
"decode_text",
"derive_export_name",
"escape",
"finalize_pdf",
"html_to_pdf_bytes",
"report",
]
# CSS for <pre>-based outputs (txt, log, json, xml): monospace, wrapped.
PRE_CSS = """
@page { size: A4; margin: 18mm 16mm; }
body {
font-family: "DejaVu Sans Mono", "Courier New", monospace;
font-size: 9pt; line-height: 1.5; color: #1c1b1a;
font-kerning: none; /* pypdf misreads WeasyPrint kerning as spaces */
}
pre { white-space: pre-wrap; word-wrap: break-word; margin: 0; }
"""
def escape(text: str) -> str:
"""HTML-escape for safe embedding in generated <pre>/<table> bodies."""
return _html.escape(text, quote=False)
def default_output_dir(output_dir=None) -> Path:
"""CLI-supplied output dir (created), else the app-managed OUTPUT_DIR."""
if output_dir:
d = Path(output_dir)
d.mkdir(parents=True, exist_ok=True)
return d
import app # lazy: avoids a module-level import cycle
return app.OUTPUT_DIR
def decode_declared(raw: bytes) -> str:
"""Decode honoring an XML-declaration / <meta charset> encoding, else UTF-8.
Reuses the pinned charset-detection regexes from the EPUB path so an HTML
file with a windows-1252 declaration decodes like an EPUB chapter.
Undecodable input never raises.
"""
import app
import codecs
m = app._XML_DECL_ENC.search(raw[:4096]) or app._META_ENC.search(raw[:8192])
if m:
enc = m.group(1).decode("ascii", errors="replace").strip()
try:
codecs.lookup(enc)
return raw.decode(enc)
except (LookupError, UnicodeDecodeError):
pass
return raw.decode("utf-8", errors="replace")
def decode_text(raw: bytes) -> str:
"""Best-effort text decode: strict UTF-8, then charset-normalizer, then replace.
Strict UTF-8 first so valid UTF-8 is never second-guessed; the
charset-normalizer pass catches legacy encodings (windows-1252, latin-1)
without mojibake.
"""
try:
return raw.decode("utf-8")
except UnicodeDecodeError:
pass
from charset_normalizer import from_bytes
best = from_bytes(raw).best()
if best is not None:
return str(best)
return raw.decode("utf-8", errors="replace")
def derive_export_name(filename: str, out_dir: Path, title: str | None = None) -> str:
"""Collision-safe PDF name, mirroring the pinned EPUB export-name contract.
base = clean(title) if title else clean(stem); empty -> 'book'.
clean: [^A-Za-z0-9._ -]+ -> '_', [:60] cap, strip trailing '.pdf'
(case-insensitive). Collision: first free of base.pdf, base-2.pdf, ... .
"""
stem = Path(filename).stem
if re.fullmatch(r"\.\w+", stem):
# bare dotfile (".txt"): Path keeps the dot in .stem; treat as empty
stem = ""
if title and re.sub(r"[^A-Za-z0-9._ -]+", "_", title).strip():
base = re.sub(r"[^A-Za-z0-9._ -]+", "_", title).strip()[:60].strip()
else:
base = re.sub(r"[^A-Za-z0-9._ -]+", "_", stem).strip()[:60].strip() or "book"
if base.lower().endswith(".pdf"):
base = base[:-4].rstrip() or "book"
filename_out = f"{base}.pdf"
counter = 2
while (Path(out_dir) / filename_out).exists():
filename_out = f"{base}-{counter}.pdf"
counter += 1
return filename_out
def html_to_pdf_bytes(html: str, css: str | None = None) -> bytes:
"""Render an HTML string to PDF bytes via WeasyPrint.
css: extra stylesheet (e.g. PRE_CSS for text dumps). Empty/None means
the document carries its own <style> (HTML/MD/SVG/table converters
inject BASE_CSS into the document head so their rules win over the
fallback at equal specificity).
"""
from weasyprint import CSS, HTML
buf = io.BytesIO()
kwargs = {"stylesheets": [CSS(string=css)]} if css else {}
HTML(string=html).write_pdf(buf, **kwargs)
return buf.getvalue()
def finalize_pdf(pdf_bytes: bytes, out_dir: Path, filename: str,
job_id: str, progress_cb=None, title: str | None = None) -> str:
"""Add metadata, write atomically (.tmp + os.replace); return the name.
title: explicit PDF /Title (e.g. HTML <title> or Markdown H1); falls
back to the export filename stem when omitted.
"""
from pypdf import PdfReader, PdfWriter
writer = PdfWriter()
for page in PdfReader(io.BytesIO(pdf_bytes)).pages:
writer.add_page(page)
if not writer.pages:
raise ValueError("no renderable content: conversion produced 0 PDF pages")
writer.add_metadata({"/Title": (title or "").strip() or Path(filename).stem})
if progress_cb:
progress_cb(99.0, "Finalizing PDF\u2026")
tmp = Path(out_dir) / (filename + ".tmp") # atomic: write .tmp then rename
with open(tmp, "wb") as out:
writer.write(out)
os.replace(tmp, Path(out_dir) / filename)
return filename
def report(progress_cb, pct, message):
"""Progress callback with a message matching the UI regex \\((\\d+)/(\\d+)\\)\n when a count label is supplied, else a plain status line."""
if progress_cb:
progress_cb(pct, message)
def convert_single(path: Path, job_id: str, *, build_html, title=None,
css=PRE_CSS, progress_cb=None, output_dir=None,
name_src: str | None = None) -> str:
"""Common single-document pipeline: build_html() -> render -> finalize.
build_html() must return (html_body, page_label) where page_label feeds
the UI progress line. Raising ValueError here becomes a job error with the
same message the user sees.
name_src: original upload/CLI filename to derive the export name from
(web uploads land on disk as {job_id}<ext>, so path.name is unusable
there); defaults to path.name (CLI mode).
"""
out_dir = default_output_dir(output_dir)
body, label = build_html()
report(progress_cb, 30.0, f"Rendering {label} (1/1)")
pdf_bytes = html_to_pdf_bytes(body, css)
filename = derive_export_name(name_src or path.name, out_dir, title=title)
return finalize_pdf(pdf_bytes, out_dir, filename, job_id,
progress_cb=progress_cb, title=title)
+71
View File
@@ -0,0 +1,71 @@
"""CSV / TSV -> PDF (typed table).
The first row is rendered as the header (<th>); every row becomes a
<table> row — the table rules from app.BASE_CSS are injected at the top of
<head> so they under-layer any user CSS. Delimiter: forced tab for .tsv;
for .csv, csv.Sniffer on a sample with a comma fallback (single-line files
that the sniffer rejects still render).
Empty / whitespace-only input is a job error ("empty file: ...").
"""
from __future__ import annotations
import csv
import io
from pathlib import Path
from . import common
from .text_converter import decode_bytes
def _rows(text: str, delimiter: str) -> list[list[str]]:
rows = list(csv.reader(io.StringIO(text), delimiter=delimiter))
return [row for row in rows if any(cell.strip() for cell in row)]
def _table_html(rows: list[list[str]]) -> str:
if not rows:
return ""
header, body = rows[0], rows[1:]
head = "".join(f"<th>{common.escape(c)}</th>" for c in header)
out = [f"<table><tr>{head}</tr>"]
for row in body:
# pad ragged rows to header width so borders line up
cells = row + [""] * (len(header) - len(row))
out.append("<tr>" + "".join(f"<td>{common.escape(c)}</td>" for c in cells) + "</tr>")
out.append("</table>")
return "\n".join(out)
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a CSV or TSV file to a tabular PDF."""
import app # lazy: BASE_CSS lives with the EPUB pipeline
raw = path.read_bytes()
if not raw.strip():
raise ValueError("empty file: nothing to convert")
text = decode_bytes(raw)
if path.suffix.lower() == ".tsv":
delimiter = "\t"
else:
try:
delimiter = csv.Sniffer().sniff(text[:8192], delimiters=",;\t|").delimiter
except csv.Error:
delimiter = ","
rows = _rows(text, delimiter)
table = _table_html(rows)
html_doc = (
"<!DOCTYPE html><html><head><style>\n" + app.BASE_CSS + "\n</style>"
f"</head><body>{table}</body></html>"
)
def _build():
return html_doc, path.name
return common.convert_single(
path, job_id, build_html=_build, css="",
progress_cb=progress_cb, output_dir=output_dir,
name_src=source_name or path.name,
)
+67
View File
@@ -0,0 +1,67 @@
"""HTML / HTM -> PDF.
A standalone .html upload is rendered as-is: the document's own <style>/
<link> rules stay authoritative. app.BASE_CSS is injected at the TOP of
<head> as a fallback layer (equal-specificity rules lose to the document's
own later declarations, same layering as the EPUB path). Fragments without
a <head> ("<p>...</p>") render bare.
Declared charset is honored (windows-1252/latin-1 <meta charset> or XML
declaration) via the pinned regexes in app — no mojibake. The document
<title> becomes the export name and PDF /Title when present, else the
file stem.
"""
from __future__ import annotations
import re
from pathlib import Path
from . import common
_TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
def _extract_title(html_text: str) -> str | None:
m = _TITLE_RE.search(html_text)
if not m:
return None
title = re.sub(r"<[^>]+>", "", m.group(1))
title = " ".join(title.split())
return title or None
def _inject_base_css(html_text: str) -> str:
"""Insert BASE_CSS as a fallback <style> at the top of <head>.
Placed before the document's own styles so, at equal specificity,
the document wins (WeasyPrint resolves source order within one
stylesheet context like a browser). Fragments without <head> are
returned unchanged (a lone <style> outside <head> is legal HTML5
and still applies).
"""
import app # lazy: BASE_CSS lives with the EPUB pipeline
style = f"<style>\n{app.BASE_CSS}\n</style>"
m = re.search(r"<head[^>]*>", html_text, re.IGNORECASE)
if m:
return html_text[: m.end()] + style + html_text[m.end():]
return style + html_text
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a standalone HTML/HTM file to a typeset PDF."""
raw = path.read_bytes()
html_text = common.decode_declared(raw)
if not html_text.strip():
raise ValueError("empty file: nothing to convert")
title = _extract_title(html_text)
def _build():
return _inject_base_css(html_text), 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,
)
+53
View File
@@ -0,0 +1,53 @@
"""Raster images (.png/.jpg/.jpeg/.webp/.tiff/.bmp) -> PDF, one page per image.
Rendered directly by Pillow (img.save(format='PDF')) — no WeasyPrint in
this path, so image fidelity is exactly what the encoder produces.
Transparency (RGBA/P) is flattened to RGB over white. Images wider or
taller than MAX_DIM are downscaled proportionally to keep the PDF sane
for e.g. 12 MP camera photos.
An unopenable/corrupt file is a job error: "Invalid image: ..." (the
contract asserts 'image' in error.lower()).
"""
from __future__ import annotations
import io
from pathlib import Path
from PIL import Image
from . import common
MAX_DIM = 2000
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a raster image to a one-page PDF."""
try:
img = Image.open(path)
img.load() # force full decode: a truncated file fails here, not later
except Exception as exc: # noqa: BLE001 - PIL raises many types
raise ValueError(f"Invalid image: {path.name} could not be opened ({exc})") from exc
if img.mode in ("RGBA", "P"):
bg = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "P":
img = img.convert("RGBA")
bg.paste(img, mask=img.getchannel("A"))
img = bg
elif img.mode != "RGB":
img = img.convert("RGB")
w, h = img.size
scale = min(1.0, MAX_DIM / max(w, h))
if scale < 1.0:
img = img.resize((max(1, int(w * scale)), max(1, int(h * scale))), Image.LANCZOS)
common.report(progress_cb, 30.0, f"Rendering {path.name} (1/1)")
buf = io.BytesIO()
img.save(buf, format="PDF")
out_dir = common.default_output_dir(output_dir)
filename = common.derive_export_name(source_name or path.name, out_dir)
return common.finalize_pdf(buf.getvalue(), out_dir, filename, job_id,
progress_cb=progress_cb)
+38
View File
@@ -0,0 +1,38 @@
"""JSON -> PDF (pretty-printed, monospace).
Validation is explicit (a JSON error is a *job* error with a clear message,
not a render failure): whitespace-only input -> "empty JSON file: ...",
unparseable input -> "Invalid JSON: <reason>". Valid documents are
pretty-printed (indent=2, original key order) into <pre> with PRE_CSS so
structure and indentation survive the render.
"""
from __future__ import annotations
import json
from pathlib import Path
from . import common
from .text_converter import decode_bytes
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a JSON file to a pretty-printed monospace PDF."""
raw = path.read_bytes()
if not raw.strip():
raise ValueError("empty JSON file: nothing to convert")
text = decode_bytes(raw)
try:
obj = json.loads(text)
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON: {exc}") from exc
pretty = json.dumps(obj, indent=2, ensure_ascii=False)
def _build():
return f"<html><body><pre>{common.escape(pretty)}</pre></body></html>", path.name
return common.convert_single(
path, job_id, build_html=_build, css=common.PRE_CSS,
progress_cb=progress_cb, output_dir=output_dir,
name_src=source_name or path.name,
)
+54
View File
@@ -0,0 +1,54 @@
"""Markdown (.md / .markdown) -> PDF.
Rendered with the `markdown` library (tables, fenced code, nl2br) and
typeset by WeasyPrint with app.BASE_CSS injected at the top of <head> so
the fallback layer sits under any user CSS. The first ATX H1 becomes the
export name and PDF /Title (it stays in the rendered document for visual
fidelity); without an H1 the file stem is used.
Empty input renders 0 pages and fails in finalize with
"no renderable content: ..." — unpinned, acceptable.
"""
from __future__ import annotations
import re
from pathlib import Path
from . import common
_FIRST_H1_RE = re.compile(r"^\s*#\s+(.+?)\s*#*\s*$", re.MULTILINE)
def _extract_h1(md_text: str) -> str | None:
m = _FIRST_H1_RE.search(md_text)
if not m:
return None
title = re.sub(r"[#*`_]", "", m.group(1))
title = " ".join(title.split())
return title or None
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a Markdown file to a typeset PDF."""
import app # lazy: BASE_CSS lives with the EPUB pipeline
import markdown
text = common.decode_text(path.read_bytes())
title = _extract_h1(text)
body_html = markdown.markdown(
text, extensions=["tables", "fenced_code", "nl2br"]
)
html_doc = (
f"<!DOCTYPE html><html><head><style>\n{app.BASE_CSS}\n</style>"
f"</head><body>{body_html}</body></html>"
)
def _build():
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,
)
+66
View File
@@ -0,0 +1,66 @@
"""PDF -> PDF passthrough (.pdf uploads).
A valid, unencrypted PDF is validated with pypdf and passed through to the
output with the standard naming/collision contract, re-stamping /Title from
the uploaded filename stem. It reuses common.finalize_pdf so the atomic
write path (.tmp + os.replace) and metadata handling are identical to every
other converter.
Rejections (all ValueError -> surfaced verbatim as a job error / CLI error):
- corrupt / unparseable -> "Invalid PDF: ... could not be parsed"
- encrypted / password -> "PDF rejected: ... encrypted or password-protected"
- more than MAX_PAGES -> "PDF too large: ... pages (max N)"
"""
from __future__ import annotations
import io
from pathlib import Path
from pypdf import PdfReader
from . import common
# req 1 (pdf passthrough): upper bound on page count for a pass-through PDF.
MAX_PAGES = 500
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Validate a PDF and pass it through to a same-named output PDF.
Satisfies: req 1 (pdf passthrough — validate, stamp /Title, standard
naming/collision, atomic write).
"""
name = source_name or path.name
stem = Path(name).stem or "book"
raw = path.read_bytes()
# req 1: validate with pypdf. Construction raises on header/body corruption.
try:
reader = PdfReader(io.BytesIO(raw))
except Exception as exc: # noqa: BLE001 - pypdf raises many corrupt-file types
raise ValueError(f"Invalid PDF: {name} could not be parsed ({exc})") from exc
# req 1: reject encrypted / password-protected BEFORE touching .pages
# (accessing pages on an encrypted reader raises, not a clean error).
if reader.is_encrypted:
raise ValueError(f"PDF rejected: {name} is encrypted or password-protected")
# req 1: count pages; a corrupt body beyond a valid header fails here.
try:
page_count = len(reader.pages)
except Exception as exc: # noqa: BLE001
raise ValueError(f"Invalid PDF: {name} could not be parsed ({exc})") from exc
# req 1: reject oversized documents (clear error, distinct from corrupt).
if page_count > MAX_PAGES:
raise ValueError(
f"PDF too large: {name} has {page_count} pages (max {MAX_PAGES})"
)
# req 1: stamp /Title from the uploaded filename stem; standard naming/collision.
common.report(progress_cb, 30.0, f"Passing through {name} (1/1)")
out_dir = common.default_output_dir(output_dir)
filename = common.derive_export_name(name, out_dir)
return common.finalize_pdf(raw, out_dir, filename, job_id,
progress_cb=progress_cb, title=stem)
+55
View File
@@ -0,0 +1,55 @@
"""SVG -> PDF (vector render via WeasyPrint).
SVG is XML by spec, so the document is first well-formedness-checked with
lxml — an unbalanced file (e.g. "<svg><g></path>") is a *job* error with
"Invalid SVG: <reason>", never a silent blank render (HTML5 foreign-content
parsing would otherwise swallow the mismatch and succeed with an empty page).
The SVG is then embedded verbatim in a minimal A4 HTML document; WeasyPrint
renders SVG natively as a vector (text stays sharp, no rasterization). A
forced page break guarantees at least one PDF page even for a degenerate
document. A WeasyPrint render failure is also reported as "Invalid SVG: ...".
"""
from __future__ import annotations
from pathlib import Path
from lxml import etree
from . import common
_DOC = (
"<!DOCTYPE html><html><head><style>"
"@page { size: A4; margin: 20mm; }"
"body { margin: 0; font-kerning: none; }"
"svg { max-width: 100%; height: auto; }"
"</style></head><body>{svg}<div style='page-break-after: always'></div>"
"</body></html>"
)
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render an SVG file to a vector PDF."""
raw = path.read_bytes()
if not raw.strip():
raise ValueError("empty file: nothing to convert")
svg_text = common.decode_text(raw)
try:
etree.fromstring(svg_text.encode("utf-8"))
except etree.XMLSyntaxError as exc:
raise ValueError(f"Invalid SVG: {exc}") from exc
doc = _DOC.replace("{svg}", svg_text)
out_dir = common.default_output_dir(output_dir)
common.report(progress_cb, 30.0, f"Rendering {path.name} (1/1)")
try:
pdf_bytes = common.html_to_pdf_bytes(doc)
except ValueError:
raise # empty-file errors keep their own message
except Exception as exc: # noqa: BLE001 - render failure is a job error
raise ValueError(f"Invalid SVG: {exc}") from exc
filename = common.derive_export_name(source_name or path.name, out_dir)
return common.finalize_pdf(pdf_bytes, out_dir, filename, job_id,
progress_cb=progress_cb)
+55
View File
@@ -0,0 +1,55 @@
"""Plain text (.txt) and log (.log) -> PDF.
Decode ladder (no mojibake on legacy files, never raises):
1. strict UTF-8 -> valid UTF-8 is never second-guessed
2. strict windows-1252 -> its 5 undefined byte holes (0x81/8D/8F/90/9D)
make a successful strict decode a reliable
cp1252 signal (covers latin-1-ish content too)
3. charset-normalizer -> best guess for other legacy encodings
4. UTF-8 errors=replace -> last resort, total losslessness guarantee
Empty / whitespace-only input is a job error ("empty file: ..."). The text
is wrapped in <pre> with PRE_CSS (monospace, wrapped, A4) so line structure
and indentation survive the render.
"""
from __future__ import annotations
from pathlib import Path
from . import common
def decode_bytes(raw: bytes) -> str:
"""Best-effort decode ladder; see module docstring."""
try:
return raw.decode("utf-8")
except UnicodeDecodeError:
pass
try:
return raw.decode("cp1252")
except UnicodeDecodeError:
pass
from charset_normalizer import from_bytes
best = from_bytes(raw).best()
if best is not None:
return str(best)
return raw.decode("utf-8", errors="replace")
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render a plain-text or log file to a monospace PDF."""
raw = path.read_bytes()
if not raw.strip():
raise ValueError("empty file: nothing to convert")
text = decode_bytes(raw)
def _build():
return f"<html><body><pre>{common.escape(text)}</pre></body></html>", path.name
return common.convert_single(
path, job_id, build_html=_build, css=common.PRE_CSS,
progress_cb=progress_cb, output_dir=output_dir,
name_src=source_name or path.name,
)
+51
View File
@@ -0,0 +1,51 @@
"""XML -> PDF (pretty text dump, monospace).
Validation is explicit: lxml must parse the document, otherwise the job
fails with "Invalid XML: <reason>" (never a render failure). A valid
document is rendered as its text content — every text node and tail in
document order, one per line, whitespace-only nodes dropped — inside
<pre> with PRE_CSS. Attribute values are intentionally not dumped: the
contract is that element text survives extraction.
"""
from __future__ import annotations
from pathlib import Path
from lxml import etree
from . import common
from .text_converter import decode_bytes
def convert(path: Path, job_id: str, progress_cb=None, output_dir=None,
source_name=None) -> str:
"""Render an XML file to a monospace text-dump PDF."""
raw = path.read_bytes()
if not raw.strip():
raise ValueError("empty file: nothing to convert")
text = decode_bytes(raw)
try:
root = etree.fromstring(text.encode("utf-8"))
except etree.XMLSyntaxError as exc:
raise ValueError(f"Invalid XML: {exc}") from exc
lines: list[str] = []
for el in root.iter():
for chunk in (el.text, el.tail):
if not chunk:
continue
piece = " ".join(chunk.split())
if piece:
lines.append(piece)
if not lines:
raise ValueError("empty file: XML has no text content")
body = "\n".join(lines)
def _build():
return f"<html><body><pre>{common.escape(body)}</pre></body></html>", path.name
return common.convert_single(
path, job_id, build_html=_build, css=common.PRE_CSS,
progress_cb=progress_cb, output_dir=output_dir,
name_src=source_name or path.name,
)
+3
View File
@@ -4,3 +4,6 @@ pypdf==6.16.2
beautifulsoup4==4.15.0 beautifulsoup4==4.15.0
lxml==6.1.2 lxml==6.1.2
gunicorn==26.2.0 gunicorn==26.2.0
markdown==3.10.3
Pillow==12.3.0
charset-normalizer==3.4.4
+7 -6
View File
@@ -118,12 +118,12 @@
<body> <body>
<header class="masthead"> <header class="masthead">
<div class="wordmark"><span class="fleuron">&#10086;</span>Bindery</div> <div class="wordmark"><span class="fleuron">&#10086;</span>Bindery</div>
<p class="tagline">A local typesetter. Drop an EPUB, take a properly set PDF. Nothing leaves your machine.</p> <p class="tagline">A local typesetter. Drop a document, take a properly set PDF. Nothing leaves your machine.</p>
</header> </header>
<div class="rule"></div> <div class="rule"></div>
<main> <main>
<section id="drop" class="drop" role="button" tabindex="0" aria-label="Upload an EPUB file"> <section id="drop" class="drop" role="button" tabindex="0" aria-label="Upload a document">
<div class="shelf" aria-hidden="true"> <div class="shelf" aria-hidden="true">
<span class="spine" style="--c:#6e3b34;--h:150px;--w:24px"></span> <span class="spine" style="--c:#6e3b34;--h:150px;--w:24px"></span>
<span class="spine" style="--c:#33513f;--h:164px;--w:20px"></span> <span class="spine" style="--c:#33513f;--h:164px;--w:20px"></span>
@@ -135,11 +135,11 @@
<span class="spine" style="--c:#46525e;--h:156px;--w:26px"></span> <span class="spine" style="--c:#46525e;--h:156px;--w:26px"></span>
</div> </div>
<div class="drop-copy"> <div class="drop-copy">
<h1>Drop a book to set it in type</h1> <h1>Drop a file to set it in type</h1>
<p>or <span class="browse">browse your files</span> &middot; up to <span id="maxmb">400</span>&nbsp;MB &middot; <code>.epub</code></p> <p>or <span class="browse">browse your files</span> &middot; up to <span id="maxmb">400</span>&nbsp;MB &middot; <code>.epub .html .md .txt .csv .json .xml .svg .png .jpg .webp &hellip;</code></p>
</div> </div>
<p id="droperr" class="drop-err" role="alert"></p> <p id="droperr" class="drop-err" role="alert"></p>
<input type="file" id="file" accept=".epub"> <input type="file" id="file" accept=".epub,.html,.htm,.txt,.log,.md,.markdown,.csv,.tsv,.json,.xml,.svg,.png,.jpg,.jpeg,.webp">
</section> </section>
<section id="status" class="status" hidden> <section id="status" class="status" hidden>
@@ -237,7 +237,8 @@
} }
function start(f) { function start(f) {
if (!/\.epub$/i.test(f.name)) { flashError('That is not an .epub — pick a book with the .epub extension.'); return; } var OKEXT = /\.(epub|html?|txt|log|md|markdown|csv|tsv|json|xml|svg|png|jpe?g|webp)$/i;
if (!OKEXT.test(f.name)) { flashError('Unsupported file type — accepted: .epub .html .md .txt .log .csv .tsv .json .xml .svg .png .jpg .webp'); return; }
var form = new FormData(); var form = new FormData();
form.append('file', f); form.append('file', f);
showStatus(); showStatus();
+2 -1
View File
@@ -7,7 +7,8 @@ from tests.helpers import make_epub_zip, make_zip_without_container, wait_for_jo
def test_wrong_extension_returns_400(client, fixture_epub_bytes): def test_wrong_extension_returns_400(client, fixture_epub_bytes):
files = {"file": (io.BytesIO(fixture_epub_bytes), "book.pdf")} # gap patch: .pdf is now a supported type; use a still-unsupported ext.
files = {"file": (io.BytesIO(fixture_epub_bytes), "book.exe")}
resp = client.post("/api/convert", data=files, content_type="multipart/form-data") resp = client.post("/api/convert", data=files, content_type="multipart/form-data")
assert resp.status_code == 400 assert resp.status_code == 400
assert ".epub" in resp.get_json()["error"] assert ".epub" in resp.get_json()["error"]
+471
View File
@@ -0,0 +1,471 @@
"""PHASE 1 multi-format expansion: any common file -> PDF.
Covers per-format happy paths (e2e through the HTTP API), the shared naming
contract (sanitise, 60-char cap, no doubled .pdf, first-free collision),
error paths (invalid content per format), the unsupported-extension 400,
CLI dispatch for non-EPUB files, and worker hygiene (upload + work cleanup).
Naming contract (mirrors the pinned EPUB export-name behavior exactly):
base = sanitize(stem)[:60] with sanitize = [^A-Za-z0-9._ -]+ -> '_'
trailing '.pdf' stripped (case-insensitive); empty stem -> 'book';
collision picks first free of base.pdf, base-2.pdf, base-3.pdf, ...
"""
import io
import json
import os
import subprocess
import sys
from pathlib import Path
from PIL import Image
from pypdf import PdfWriter
import app
import converters
from converters import common as conv_common
from tests.helpers import (download_bytes, pdf_metadata_title, pdf_page_count,
pdf_text, wait_for_job)
ROOT = Path(__file__).resolve().parents[1]
# --------------------------------------------------------------------------
# fixtures
# --------------------------------------------------------------------------
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 _png_bytes(size=(640, 480), color=(200, 30, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="PNG")
return buf.getvalue()
def _jpeg_bytes(size=(640, 480), color=(30, 200, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="JPEG", quality=90)
return buf.getvalue()
def _webp_bytes(size=(320, 240), color=(30, 30, 200)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="WEBP", quality=90)
return buf.getvalue()
# --------------------------------------------------------------------------
# is_supported / extension matrix
# --------------------------------------------------------------------------
def test_is_supported_matrix():
for ext in (".epub", ".html", ".htm", ".txt", ".log", ".md", ".markdown",
".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"):
assert not converters.is_supported(f"book{ext}"), ext
assert not converters.is_supported("noextension")
def test_is_supported_case_insensitive():
assert converters.is_supported("BOOK.HTML")
assert converters.is_supported("Photo.JPEG")
def test_convert_file_unsupported_raises_valueerror(tmp_path):
p = tmp_path / "virus.exe"
p.write_bytes(b"MZ")
try:
converters.convert_file(p, "jobx")
assert False, "expected ValueError"
except ValueError as exc:
assert "Unsupported" in str(exc)
# --------------------------------------------------------------------------
# API: unsupported extension -> 400 (legacy message contract kept)
# --------------------------------------------------------------------------
def test_unsupported_ext_returns_400(client, fixture_epub_bytes):
# gap patch: .pdf is now a supported type; use a still-unsupported ext.
resp = _submit(client, "book.mp4", fixture_epub_bytes)
assert resp.status_code == 400
assert ".epub" in resp.get_json()["error"]
def test_unsupported_exe_returns_400(client):
resp = _submit(client, "virus.exe", b"MZ\x90")
assert resp.status_code == 400
assert resp.get_json()["error"]
# --------------------------------------------------------------------------
# per-format happy paths (e2e)
# --------------------------------------------------------------------------
def test_html_e2e_title_and_content(client):
html_doc = (
"<!DOCTYPE html><html><head><title>My HTML Doc</title>"
"<style>body{color:navy}</style></head>"
"<body><h1>Heading One</h1><p>HTML MARKER ALPHA</p></body></html>"
)
job, data, text = _e2e(client, "doc.html", html_doc)
assert job["filename"] == "My HTML Doc.pdf"
assert job["epub"] == "doc.html" # original upload name preserved for UI
assert "HTML MARKER ALPHA" in text
assert "Heading One" in text
def test_html_fragment_without_title_uses_stem(client):
job, _, text = _e2e(client, "fragment.htm", "<p>FRAG MARK BRAVO</p>")
assert job["filename"] == "fragment.pdf"
assert "FRAG MARK BRAVO" in text
def test_html_declared_windows1252_meta(client):
body = "Caf\u00e9 \u2014 em dash \u2014 r\u00e9sum\u00e9"
raw = (
'<html><head><meta charset="windows-1252"></head><body><p>'
).encode("utf-8") + body.encode("cp1252") + b"</p></body></html>"
job, _, text = _e2e(client, "latin.html", raw)
assert "r\u00e9sum\u00e9" in text and "\u2014" in text
def test_txt_e2e(client):
job, _, text = _e2e(client, "notes.txt", "TXT MARKER CHARLIE\nsecond line\n\n")
assert job["filename"] == "notes.pdf"
assert "TXT MARKER CHARLIE" in text
assert "second line" in text
def test_txt_windows1252_autodetect(client):
body = "Caf\u00e9 \u2014 dash \u2014 fen\u00eatre \u2014 r\u00e9sum\u00e9 \u2014 fin"
job, _, text = _e2e(client, "win1252.txt", body.encode("cp1252"))
assert "r\u00e9sum\u00e9" in text
assert "\u2014" in text
def test_md_e2e_title_from_h1(client):
md = "# MD Title One\n\nBody with **BOLDWORD** and a list:\n\n- item A\n- item B\n"
job, _, text = _e2e(client, "chapter.md", md)
assert job["filename"] == "MD Title One.pdf"
assert "BOLDWORD" in text
assert "item A" in text
def test_md_no_title_uses_stem(client):
job, _, text = _e2e(client, "plain.md", "just text, no heading")
assert job["filename"] == "plain.pdf"
assert "just text, no heading" in text
def test_csv_e2e(client):
job, _, text = _e2e(client, "stock.csv", "name,qty\napple,3\npear,5\n")
assert job["filename"] == "stock.pdf"
assert "apple" in text
assert "qty" in text
assert "5" in text
def test_tsv_e2e(client):
job, _, text = _e2e(client, "grid.tsv", "colA\tcolB\tv1\tv2\n")
assert job["filename"] == "grid.pdf"
assert "v1" in text and "v2" in text
def test_json_e2e(client):
doc = json.dumps({"alpha": [1, 2, 3], "beta": "GAMMA_JSON"}, indent=2)
job, _, text = _e2e(client, "config.json", doc)
assert job["filename"] == "config.pdf"
assert "GAMMA_JSON" in text
assert "alpha" in text
def test_xml_e2e(client):
doc = '<?xml version="1.0"?>\n<root><item>GAMMA_XML</item><item>two</item></root>'
job, _, text = _e2e(client, "feed.xml", doc)
assert job["filename"] == "feed.pdf"
assert "GAMMA_XML" in text
def test_svg_e2e(client):
svg = (
'<svg xmlns="http://www.w3.org/2000/svg" width="400" height="300">'
"<rect width='400' height='300' fill='#eef'/>"
"<text x='20' y='150' font-size='24'>SVG MARK DELTA</text></svg>"
)
job, data, _ = _e2e(client, "diagram.svg", svg)
assert job["filename"] == "diagram.pdf"
assert pdf_page_count(data) >= 1
assert len(data) > 1024
def test_image_png_e2e(client):
job, data, _ = _e2e(client, "photo.png", _png_bytes())
assert job["filename"] == "photo.pdf"
assert pdf_page_count(data) == 1
def test_image_jpeg_e2e(client):
job, data, _ = _e2e(client, "scan.jpg", _jpeg_bytes())
assert job["filename"] == "scan.pdf"
assert pdf_page_count(data) == 1
def test_image_webp_e2e(client):
job, data, _ = _e2e(client, "snap.webp", _webp_bytes())
assert job["filename"] == "snap.pdf"
assert pdf_page_count(data) == 1
def test_uppercase_extension_e2e(client):
job, _, text = _e2e(client, "UPPER.HTML", "<p>UPPER MARK ECHO</p>")
assert job["filename"] == "UPPER.pdf"
assert "UPPER MARK ECHO" in text
# --------------------------------------------------------------------------
# error paths (accepted 202, job ends in error with a clear message)
# --------------------------------------------------------------------------
def _error_job(client, filename, content):
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
def test_invalid_json_is_job_error(client):
job = _error_job(client, "bad.json", "{not valid json]")
assert "JSON" in job["error"]
def test_invalid_xml_is_job_error(client):
job = _error_job(client, "bad.xml", "<a><b></a>")
assert "XML" in job["error"]
def test_invalid_svg_is_job_error(client):
job = _error_job(client, "bad.svg", "<svg><g></path>")
assert job["error"]
def test_corrupt_png_is_job_error(client):
job = _error_job(client, "fake.png", b"\x89PNG\r\n\x1a\n definitely not a png")
assert "image" in job["error"].lower()
def test_empty_txt_is_job_error(client):
job = _error_job(client, "empty.txt", b"")
assert "empty" in job["error"].lower()
def test_empty_csv_is_job_error(client):
job = _error_job(client, "empty.csv", b"")
assert "empty" in job["error"].lower()
def test_empty_json_is_job_error(client):
job = _error_job(client, "empty.json", b" ")
assert "JSON" in job["error"]
# --------------------------------------------------------------------------
# naming contract (unit level via converters.common)
# --------------------------------------------------------------------------
def test_naming_sanitize_special_chars(tmp_path):
name = conv_common.derive_export_name("My File!! Final.txt", tmp_path)
assert name == "My File_ Final.pdf"
def test_naming_60_char_cap(tmp_path):
stem = "x" * 120
name = conv_common.derive_export_name(f"{stem}.txt", tmp_path)
assert name == ("x" * 60) + ".pdf"
def test_naming_strips_doubled_pdf(tmp_path):
assert conv_common.derive_export_name("report.PDF.txt", tmp_path) == "report.pdf"
assert conv_common.derive_export_name("x.pdf.txt", tmp_path) == "x.pdf"
def test_naming_empty_stem_falls_back_to_book(tmp_path):
assert conv_common.derive_export_name(".txt", tmp_path) == "book.pdf"
def test_naming_collision_first_free(tmp_path):
(tmp_path / "dup.pdf").write_bytes(b"old1")
(tmp_path / "dup-2.pdf").write_bytes(b"old2")
assert conv_common.derive_export_name("dup.txt", tmp_path) == "dup-3.pdf"
def test_naming_collision_e2e_second_upload_gets_suffix(client):
job1, _, _ = _e2e(client, "same.txt", "one")
job2, _, _ = _e2e(client, "same.txt", "two")
assert job1["filename"] == "same.pdf"
assert job2["filename"] == "same-2.pdf"
# --------------------------------------------------------------------------
# download + worker hygiene for non-EPUB conversions
# --------------------------------------------------------------------------
def test_download_headers_for_txt_conversion(client):
job, _, _ = _e2e(client, "headers.txt", "header check")
resp = client.get(f"/download/{job['filename']}")
assert resp.status_code == 200
assert resp.content_type == "application/pdf"
assert "attachment" in resp.headers["Content-Disposition"]
assert 'filename="headers.pdf"' in resp.headers["Content-Disposition"]
def test_worker_cleans_upload_and_work_dir(client):
resp = _submit(client, "hygiene.txt", "cleanup check")
job_id = resp.get_json()["job_id"]
job = wait_for_job(client, job_id)
assert job["status"] == "done", job
assert not (app.UPLOAD_DIR / f"{job_id}.txt").exists()
assert not (app.WORK_DIR / f"job-{job_id}").exists()
def test_job_snapshot_keeps_original_name(client):
resp = _submit(client, "report Q3.txt", "x")
job = wait_for_job(client, resp.get_json()["job_id"])
assert job["status"] == "done", job
assert job["epub"] == "report Q3.txt" # UI bookname source, now generic
assert job["message"] == "Done"
# --------------------------------------------------------------------------
# CLI dispatch for non-EPUB files
# --------------------------------------------------------------------------
def _run_cli(*args):
return subprocess.run(
[sys.executable, str(ROOT / "app.py"), *args],
capture_output=True, text=True, timeout=120,
env=dict(os.environ), cwd=str(ROOT),
)
def test_cli_markdown_file(tmp_path):
src = tmp_path / "notes.md"
src.write_text("# CLI MD Title\n\nCLI body text.", encoding="utf-8")
out = tmp_path / "out"
proc = _run_cli("--cli", str(src), "--output", str(out))
assert proc.returncode == 0, proc.stderr
assert "Saved:" in proc.stdout
pdfs = list(out.glob("*.pdf"))
assert len(pdfs) == 1 and pdfs[0].name == "CLI MD Title.pdf"
assert pdfs[0].read_bytes()[:4] == b"%PDF"
def test_cli_unsupported_ext_exits_1(tmp_path):
src = tmp_path / "prog.exe"
src.write_bytes(b"MZ")
proc = _run_cli("--cli", str(src), "--output", str(tmp_path))
assert proc.returncode == 1
assert "error:" in proc.stdout
# --------------------------------------------------------------------------
# PDF pass-through + added image formats (Phase 1 gap patch)
# --------------------------------------------------------------------------
def _blank_pdf_bytes(pages=3):
w = PdfWriter()
for _ in range(pages):
w.add_blank_page(width=612, height=792)
buf = io.BytesIO()
w.write(buf)
return buf.getvalue()
def _encrypted_pdf_bytes():
w = PdfWriter()
for _ in range(2):
w.add_blank_page(width=612, height=792)
w.encrypt(user_password="secret", owner_password="owner")
buf = io.BytesIO()
w.write(buf)
return buf.getvalue()
def _tiff_bytes(size=(320, 240), color=(120, 30, 200)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="TIFF")
return buf.getvalue()
def _bmp_bytes(size=(320, 240), color=(200, 120, 30)):
buf = io.BytesIO()
Image.new("RGB", size, color).save(buf, format="BMP")
return buf.getvalue()
def test_pdf_passthrough_e2e(client):
# req 1: valid pdf passes through — same page count, /Title from stem.
job, data, _ = _e2e(client, "My Report.pdf", _blank_pdf_bytes(3))
assert job["filename"] == "My Report.pdf"
assert pdf_page_count(data) == 3
assert pdf_metadata_title(data) == "My Report"
def test_pdf_encrypted_rejected(client):
# req 1: encrypted / password-protected pdf -> job error, not a crash.
job = _error_job(client, "locked.pdf", _encrypted_pdf_bytes())
err = job["error"].lower()
assert "encrypted" in err or "password" in err
def test_pdf_corrupt_rejected(client):
# req 1: corrupt pdf -> job error naming the problem.
job = _error_job(client, "broken.pdf", b"%PDF-1.4\n" + b"garbage" * 64)
assert "pdf" in job["error"].lower()
def test_pdf_over_500_pages_rejected(client):
# req 1: page-count cap (max 500) -> clear job error.
job = _error_job(client, "huge.pdf", _blank_pdf_bytes(501))
assert "too large" in job["error"].lower()
def test_image_tiff_e2e(client):
# req 2: .tiff now accepted by the image converter.
job, data, _ = _e2e(client, "film.tiff", _tiff_bytes())
assert job["filename"] == "film.pdf"
assert pdf_page_count(data) == 1
def test_image_bmp_e2e(client):
# req 2: .bmp now accepted by the image converter.
job, data, _ = _e2e(client, "old.bmp", _bmp_bytes())
assert job["filename"] == "old.pdf"
assert pdf_page_count(data) == 1