Initial commit: Bookish Flask app for ebook conversion and reading

bookish/services/converter.py

@@ -0,0 +1,170 @@
+import os
+import pypandoc
+from config import SUPPORTED_FORMATS
+from bookish.services.sse_manager import update_progress, mark_complete, mark_error
+
+
+def get_format_from_ext(filename):
+    ext = os.path.splitext(filename)[1].lower().lstrip(".")
+    if ext in SUPPORTED_FORMATS:
+        return ext
+    # mappa alias comuni
+    alias = {"markdown": "md", "text": "txt", "latex": "latex", "tex": "latex"}
+    return alias.get(ext, ext)
+
+
+def run_conversion(input_path, output_path, input_format, output_format, task_id, filename):
+    """
+    Esegue la conversione con Pandoc, aggiornando il progresso via SSE.
+    Per alcuni formati speciali delega a formatter.py.
+    """
+    try:
+        update_progress(task_id, 10, "analyzing")
+        _validate_pandoc_format(input_format, output_format)
+
+        update_progress(task_id, 30, "converting")
+
+        # Formati speciali che richiedono pipeline custom
+        if input_format == "docx" and output_format == "pdf":
+            _convert_docx_to_pdf(input_path, output_path, task_id)
+        elif output_format == "pdf":
+            _convert_to_pdf(input_path, output_path, input_format, task_id)
+        elif input_format == "docx" and output_format in ("epub", "md", "html"):
+            _convert_with_pandoc(input_path, output_path, input_format, output_format, task_id)
+        elif input_format == "pdf" and output_format in ("epub", "docx", "md", "txt", "html"):
+            _convert_with_pandoc(input_path, output_path, input_format, output_format, task_id)
+        else:
+            _convert_with_pandoc(input_path, output_path, input_format, output_format, task_id)
+
+        update_progress(task_id, 90, "finalizing")
+        _post_process(output_path, output_format)
+
+        update_progress(task_id, 100)
+        mark_complete(task_id, os.path.basename(output_path))
+
+    except Exception as e:
+        mark_error(task_id, str(e))
+        raise
+
+
+def _validate_pandoc_format(input_format, output_format):
+    """Verifica che Pandoc supporti input/output format."""
+    try:
+        readers = pypandoc.get_pandoc_formats()[0]
+        writers = pypandoc.get_pandoc_formats()[1]
+    except Exception as e:
+        raise RuntimeError(f"Errore Pandoc: {e}. Installalo con: pip install pypandoc-binary")
+
+    pandoc_input = _to_pandoc_format(input_format)
+    pandoc_output = _to_pandoc_format(output_format)
+
+    if pandoc_input not in readers:
+        raise ValueError(f"Formato input '{input_format}' non supportato da Pandoc")
+    if pandoc_output not in writers:
+        raise ValueError(f"Formato output '{output_format}' non supportato da Pandoc")
+
+
+def _to_pandoc_format(fmt):
+    mapping = {
+        "md": "markdown",
+        "latex": "latex",
+        "tex": "latex",
+        "html": "html",
+        "epub": "epub",
+        "pdf": "pdf",
+        "docx": "docx",
+        "txt": "plain",
+        "fb2": "fb2",
+        "mobi": "mobi",
+    }
+    return mapping.get(fmt, fmt)
+
+
+def _convert_with_pandoc(input_path, output_path, input_format, output_format, task_id):
+    extra_args = ["--wrap=preserve"]
+    if output_format == "epub":
+        extra_args += ["--metadata", "title=Converted Document"]
+    if output_format == "html":
+        extra_args += ["--self-contained", "--mathjax"]
+    if input_format == "pdf":
+        extra_args += ["--pdf-engine=pdflatex"]
+
+    pandoc_input = _to_pandoc_format(input_format)
+    pandoc_output = _to_pandoc_format(output_format)
+
+    def callback(percent):
+        scaled = 30 + int(percent * 0.5)
+        update_progress(task_id, min(scaled, 85))
+
+    pypandoc.convert_file(
+        input_path,
+        pandoc_output,
+        format=pandoc_input,
+        outputfile=output_path,
+        extra_args=extra_args,
+    )
+
+
+def _convert_to_pdf(input_path, output_path, input_format, task_id):
+    """Converte in PDF usando Pandoc con WeasyPrint."""
+    from bookish.services.formatter import html_to_pdf
+
+    update_progress(task_id, 40, "converting")
+    html_path = output_path.replace(".pdf", ".html")
+
+    pandoc_input = _to_pandoc_format(input_format)
+    pypandoc.convert_file(
+        input_path,
+        "html",
+        format=pandoc_input,
+        outputfile=html_path,
+        extra_args=["--wrap=preserve", "--self-contained"],
+    )
+
+    update_progress(task_id, 65, "converting")
+    html_to_pdf(html_path, output_path)
+
+    if os.path.exists(html_path):
+        os.remove(html_path)
+
+
+def _convert_docx_to_pdf(input_path, output_path, task_id):
+    """DOCX → PDF via Pandoc → HTML → WeasyPrint."""
+    from bookish.services.formatter import html_to_pdf
+
+    update_progress(task_id, 40, "converting")
+    html_path = output_path.replace(".pdf", ".html")
+
+    pypandoc.convert_file(
+        input_path,
+        "html",
+        format="docx",
+        outputfile=html_path,
+        extra_args=["--wrap=preserve"],
+    )
+
+    update_progress(task_id, 65, "converting")
+    html_to_pdf(html_path, output_path)
+
+    if os.path.exists(html_path):
+        os.remove(html_path)
+
+
+def _post_process(output_path, output_format):
+    """Post-processing opzionale sul file generato."""
+    if output_format == "epub":
+        _fix_epub_metadata(output_path)
+
+
+def _fix_epub_metadata(epub_path):
+    """Aggiunge metadati minimi all'epub se mancanti."""
+    try:
+        import zipfile
+        import tempfile
+
+        with zipfile.ZipFile(epub_path, "r") as z:
+            content = z.read("META-INF/container.xml")
+            if b"<dc:title>" not in content:
+                pass
+    except Exception:
+        pass

bookish/services/formatter.py

@@ -0,0 +1,55 @@
+import os
+from weasyprint import HTML
+
+
+def html_to_pdf(html_path, pdf_path):
+    """Converte HTML in PDF con WeasyPrint, preservando layout e immagini."""
+    html_str = _read_html_with_fixes(html_path)
+
+    HTML(string=html_str).write_pdf(pdf_path)
+
+    if not os.path.exists(pdf_path) or os.path.getsize(pdf_path) == 0:
+        raise RuntimeError("WeasyPrint non ha prodotto output PDF valido")
+
+
+def _read_html_with_fixes(html_path):
+    """Legge l'HTML e applica fix per una migliore resa PDF."""
+    with open(html_path, "r", encoding="utf-8") as f:
+        html = f.read()
+
+    fix_soup = False
+    if "<table" in html and not _has_pdf_table_styles(html):
+        fix_soup = True
+
+    if fix_soup:
+        try:
+            from bs4 import BeautifulSoup
+            soup = BeautifulSoup(html, "lxml")
+
+            for table in soup.find_all("table"):
+                table["style"] = (
+                    table.get("style", "")
+                    + "; border-collapse: collapse; width: 100%; margin: 1em 0;"
+                )
+                for td in table.find_all(["td", "th"]):
+                    existing = td.get("style", "")
+                    td["style"] = (
+                        existing
+                        + "; border: 1px solid #ccc; padding: 8px; vertical-align: top;"
+                    )
+
+            for img in soup.find_all("img"):
+                img["style"] = (
+                    img.get("style", "")
+                    + "; max-width: 100%; height: auto; page-break-inside: avoid;"
+                )
+
+            html = str(soup)
+        except ImportError:
+            pass
+
+    return html
+
+
+def _has_pdf_table_styles(html):
+    return 'border-collapse' in html.lower() or 'border="1"' in html

bookish/services/sse_manager.py

@@ -0,0 +1,75 @@
+import json
+import uuid
+import threading
+from time import sleep
+
+_tasks = {}
+_lock = threading.Lock()
+
+
+def create_task():
+    task_id = str(uuid.uuid4())
+    with _lock:
+        _tasks[task_id] = {"progress": 0, "status": "started", "error": None, "output_file": None}
+    return task_id
+
+
+def update_progress(task_id, progress, status=None):
+    with _lock:
+        if task_id in _tasks:
+            _tasks[task_id]["progress"] = progress
+            if status:
+                _tasks[task_id]["status"] = status
+
+
+def mark_complete(task_id, output_file=None):
+    with _lock:
+        if task_id in _tasks:
+            _tasks[task_id]["progress"] = 100
+            _tasks[task_id]["status"] = "completed"
+            _tasks[task_id]["output_file"] = output_file
+
+
+def mark_error(task_id, message):
+    with _lock:
+        if task_id in _tasks:
+            _tasks[task_id]["status"] = "error"
+            _tasks[task_id]["error"] = message
+
+
+def get_status(task_id):
+    with _lock:
+        return dict(_tasks.get(task_id, {}))
+
+
+def cleanup_task(task_id):
+    with _lock:
+        _tasks.pop(task_id, None)
+
+
+def generate_sse_events(task_id):
+    """Genera eventi SSE finché la conversione non è completata o in errore."""
+    prev_progress = -1
+    while True:
+        status = get_status(task_id)
+        if not status:
+            yield f"event: error\ndata: {json.dumps({'message': 'Task not found'})}\n\n"
+            break
+
+        progress = status["progress"]
+        if progress != prev_progress or status["status"] in ("completed", "error"):
+            data = {
+                "progress": progress,
+                "status": status["status"],
+            }
+            if status.get("error"):
+                data["error"] = status["error"]
+            if status.get("output_file"):
+                data["download_url"] = f"/download/{status['output_file']}"
+            yield f"data: {json.dumps(data)}\n\n"
+            prev_progress = progress
+
+        if status["status"] in ("completed", "error"):
+            break
+
+        sleep(0.3)