epub2pdf/converters/common.py

"""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)