Files
flowdeck/app/services/export.py
T
bruno fa97f07ec8
FlowDeck CI / test (push) Failing after 19s
FlowDeck CI / docker (push) Skipped
fix(export): v4.7.2 contenu des documents absent des exports MD/HTML/PDF
Le service d'export ne lisait que les pages content_format='blocks'. Les docs
stockees autrement sortaient avec le seul titre:
- content_format='file' (.md/code uploades): content=JSON meta, le vrai texte
  est sur disque (/data/uploads/workspace_*) -> n'etait jamais lu.
- content_format='markdown': HTML/PDF enveloppait chaque ligne en <p> (headings
  et listes aplatis).

Resolution de la vraie source pour les 3 formats (app/services/export.py):
- lit le fichier upload sur disque pour les pages file (repertoire via
  FLOWDECK_DATA_DIR, defaut /data),
- rend les pages markdown / fichiers .md en blocs (headings, listes, code,
  quote, todo) pour un HTML/PDF riche,
- fichiers texte non-markdown -> bloc de code,
- binaires (PDF/images) ignores.

195/195 tests (5 nouveaux v4.7.2). Verifie en reel via HTTP sur README.md et
l'arborescence Base de Connaissances (contenu complet dans les 3 formats).
2026-09-03 01:29:35 -04:00

603 lines
23 KiB
Python

"""FlowDeck — Export service (v4.7.2).
Four types of export, all generated server-side:
- Markdown (``page_to_markdown``): title + blocks + récursif sous-pages
- HTML (``page_to_standalone_html``): document autonome (styles inline)
- PDF (``page_to_pdf_bytes``): convertit un HTML print-friendly
- Site (``build_static_site``): site statique multi-pages (zip)
Supports the three ways a page's content can be stored:
- content_format == "blocks" -> JSON list of blocks in ``content``
- content_format == "markdown" -> raw Markdown in ``content``
- content_format == "file" -> ``content`` is JSON metadata; the real text
lives in an uploaded file on disk (uploads/workspace_*). We read it back so
an exported document carries its actual content, not just its title.
"""
from __future__ import annotations
import io
import json
import os
import re
import zipfile
from pathlib import Path
from urllib.parse import quote
from app.db import get_conn
# ═══════════════ Helpers ═══════════════
def _text(v: str, *, escape: bool = True) -> str:
"""Normalize a block's content string."""
s = (v or "").replace("\r\n", "\n").replace("\r", "\n")
if escape:
s = (s.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;"))
return s
def _sanitize_id(block_id) -> str:
if not block_id:
return ""
return "".join(ch for ch in str(block_id) if ch.isalnum())
def _page_title(page: dict) -> str:
return (page.get("title") or "Untitled").strip() or "Untitled"
def _blocks_of(page: dict) -> list:
content = page.get("content") or ""
fmt = page.get("content_format") or "blocks"
if fmt != "blocks" or not content:
return []
try:
data = json.loads(content)
except (json.JSONDecodeError, TypeError):
return []
return data if isinstance(data, list) else []
def _block_text(b: dict) -> str:
return _text(b.get("content"), escape=False)
# ── Source resolution: read real textual content for ANY page type ──
# Extensions whose content is plain text / code / markdown (textual exportable).
_TEXTUAL_EXTS = {
"md", "markdown", "txt", "log", "text",
"py", "js", "ts", "jsx", "tsx", "html", "htm", "css", "json", "xml",
"yaml", "yml", "toml", "ini", "cfg", "conf", "env", "sh", "bash", "zsh",
"ps1", "bat", "cmd", "rb", "go", "rs", "java", "c", "cpp", "h", "hpp",
"php", "swift", "kt", "scala", "sql", "r", "vue", "svelte", "astro",
"properties", "gitignore", "dockerfile", "makefile",
}
_CODE_LANG = {
"py": "python", "js": "javascript", "ts": "typescript", "jsx": "javascript",
"tsx": "typescript", "html": "html", "htm": "html", "css": "css",
"json": "json", "xml": "xml", "yaml": "yaml", "yml": "yaml",
"toml": "toml", "ini": "ini", "cfg": "ini", "conf": "ini", "env": "ini",
"sh": "bash", "bash": "bash", "zsh": "bash", "ps1": "powershell",
"bat": "batch", "cmd": "batch", "rb": "ruby", "go": "go", "rs": "rust",
"java": "java", "c": "c", "cpp": "cpp", "h": "c", "hpp": "cpp",
"php": "php", "swift": "swift", "kt": "kotlin", "scala": "scala",
"sql": "sql", "r": "r", "vue": "html", "svelte": "html",
"astro": "html", "properties": "ini", "md": "markdown",
"markdown": "markdown", "txt": "plaintext", "log": "plaintext",
"text": "plaintext",
}
_MARKDOWN_MIMES = {"text/markdown", "text/x-markdown", "application/octet-stream"}
def _data_root() -> Path:
"""Directory that contains ``uploads/`` (mirrors dashboard.py /data)."""
return Path(os.environ.get("FLOWDECK_DATA_DIR", "/data"))
def _file_meta(page: dict) -> dict:
try:
meta = json.loads(page.get("content") or "{}")
return meta if isinstance(meta, dict) else {}
except (json.JSONDecodeError, TypeError):
return {}
def _file_text(page: dict) -> str | None:
"""Return the textual content of an uploaded ``file`` page, or None.
Only reads plain-text / code / markdown files. Binary (PDF, images…)
returns None and is skipped by exporters (nothing meaningful to include).
"""
if (page.get("content_format") or "") != "file":
return None
meta = _file_meta(page)
rel = (meta.get("file_path") or "").replace("\\", "/").strip()
if not rel or ".." in rel.replace("\\", "/").split("/") or not rel.startswith("uploads/"):
return None
name = (rel.rsplit("/", 1)[-1] or "").lower()
ext = name.rsplit(".", 1)[-1] if "." in name else ""
mime = (meta.get("mime_type") or "").lower()
if not (ext in _TEXTUAL_EXTS or mime.startswith("text/")):
return None
try:
full = (_data_root() / rel).resolve()
root = _data_root().resolve()
if root not in full.parents:
return None
return full.read_text(encoding="utf-8", errors="replace")
except (OSError, ValueError):
return None
def _page_source(page: dict):
"""Return (kind, payload) describing where the page's real content lives.
kind ∈ {"blocks", "md", "code"}:
- "blocks": payload is the block list (block editor pages)
- "md" : payload is raw Markdown text
- "code" : payload is (text, language)
An empty/unsupported page yields ("blocks", []).
"""
fmt = (page.get("content_format") or "blocks")
content = page.get("content") or ""
if fmt == "blocks":
return "blocks", _blocks_of(page)
if fmt == "markdown":
if content.strip():
return "md", content
return "blocks", []
if fmt == "file":
text = _file_text(page)
if text is None:
return "blocks", []
meta = _file_meta(page)
name = (meta.get("file_path") or "").replace("\\", "/").rsplit("/", 1)[-1].lower()
ext = name.rsplit(".", 1)[-1] if "." in name else ""
mime = (meta.get("mime_type") or "").lower()
if ext in ("md", "markdown") or mime in _MARKDOWN_MIMES or mime.startswith("text/markdown"):
return "md", text
lang = _CODE_LANG.get(ext, "plaintext")
return "code", (text, lang)
# Unknown format (e.g. legacy) -> try to dump as raw text
if content.strip():
return "md", content
return "blocks", []
# ── Markdown renderer (raw markdown → exportable fragments) ──
def _md_to_blocks(md: str) -> list:
"""Convert raw Markdown text into the same lightweight block list the
editor produces (headings, lists, to-do, quote, code, divider, paragraph).
Kept intentionally simple: inline formatting (bold/links) is preserved as
literal text, matching how the block editor treats imported .md files.
"""
blocks: list = []
buf = md.replace("\r\n", "\n").replace("\r", "\n")
lines = buf.split("\n")
i = 0
n = len(lines)
para: list[str] = []
def flush_para():
nonlocal para
if para:
blocks.append({"type": "paragraph", "content": "\n".join(para).strip()})
para = []
while i < n:
line = lines[i].rstrip()
stripped = line.strip()
if not stripped:
flush_para()
i += 1
continue
if stripped.startswith("```") or stripped.startswith("~~~"):
flush_para()
fence = stripped[0:3]
lang = stripped[3:].strip()
i += 1
code: list[str] = []
while i < n and not lines[i].strip().startswith(fence):
code.append(lines[i])
i += 1
if i < n:
i += 1 # closing fence
blocks.append({"type": "code", "content": "\n".join(code), "language": lang})
continue
m = re.match(r"^(#{1,6})\s+(.*)$", stripped)
if m and line == stripped: # ATX heading must be whole line
level = len(m.group(1))
flush_para()
blocks.append({"type": f"heading_{min(level, 4)}", "content": m.group(2).strip()})
i += 1
continue
if stripped == "---" or stripped == "***" or stripped == "___":
flush_para()
blocks.append({"type": "divider", "content": ""})
i += 1
continue
if re.match(r"^\s*[-*+]\s+", line):
flush_para()
while i < n:
s = lines[i].strip()
m2 = re.match(r"^[-*+]\s+(.*)$", s)
if not m2:
break
blocks.append({"type": "bulleted_list", "content": m2.group(1).strip()})
i += 1
continue
if re.match(r"^\s*\d+[.)]\s+", line):
flush_para()
while i < n:
s = lines[i].strip()
m2 = re.match(r"^\d+[.)]\s+(.*)$", s)
if not m2:
break
blocks.append({"type": "numbered_list", "content": m2.group(1).strip()})
i += 1
continue
mtodo = re.match(r"^\s*[-*+]\s+\[([ xX])\]\s+(.*)$", stripped)
if mtodo:
flush_para()
while i < n:
s = lines[i].strip()
m2 = re.match(r"^[-*+]\s+\[([ xX])\]\s+(.*)$", s)
if not m2:
break
blocks.append({
"type": "to_do",
"content": m2.group(2).strip(),
"checked": m2.group(1).lower() == "x",
})
i += 1
continue
mq = re.match(r"^>\s?(.*)$", stripped)
if mq and line == stripped:
flush_para()
while i < n:
s = lines[i].strip()
m2 = re.match(r"^>\s?(.*)$", s)
if not m2:
break
para.append(m2.group(1))
i += 1
blocks.append({"type": "quote", "content": "\n".join(para)})
para = []
continue
para.append(stripped)
i += 1
flush_para()
return blocks
def _page_blocks(page: dict) -> list:
"""Blocks used for HTML/PDF rendering regardless of storage format."""
kind, payload = _page_source(page)
if kind == "blocks":
return payload
if kind == "code":
text, lang = payload
return [{"type": "code", "content": text, "language": lang}] if text else []
if kind == "md":
return _md_to_blocks(payload)
return []
def _page_markdown_source(page: dict) -> str:
"""Raw markdown when the page IS markdown-sourced, else empty string."""
kind, payload = _page_source(page)
if kind == "md":
return payload
return ""
# ═══════════════ Markdown ═══════════════
def blocks_to_markdown(blocks: list) -> str:
"""Convert a block array to Markdown (server-side, all block types)."""
out: list[str] = []
for b in blocks or []:
t = b.get("type", "paragraph")
c = _block_text(b)
if t == "heading_1":
out.append(f"# {c}")
elif t == "heading_2":
out.append(f"## {c}")
elif t == "heading_3":
out.append(f"### {c}")
elif t == "heading_4":
out.append(f"#### {c}")
elif t == "bulleted_list":
out.append(f"- {c}")
elif t == "numbered_list":
out.append(f"1. {c}")
elif t == "to_do":
out.append(f"{'- [x]' if b.get('checked') else '- [ ]'} {c}")
elif t == "quote":
out.append(f"> {c}")
elif t == "divider":
out.append("---")
elif t == "code":
lang = b.get("language") or ""
out.append(f"```{lang}\n{c}\n```")
elif t == "toggle":
out.append(f"### {c}")
if b.get("children"):
out.append(blocks_to_markdown(b["children"]))
elif t == "math":
out.append(f"$$\n{c}\n$$")
elif t == "table_of_contents":
out.append("[TOC]")
elif t == "columns":
for child in b.get("children") or []:
out.append(blocks_to_markdown([child]))
elif t == "image":
src = b.get("src") or ""
alt = (b.get("alt") or "").strip() or "image"
out.append(f"![{alt}]({src})")
else:
out.append(c)
return "\n\n".join(filter(None, out))
def _child_pages(page: dict) -> list:
"""Immediate non-deleted children of a page."""
with get_conn() as conn:
rows = conn.execute(
"SELECT * FROM pages WHERE parent_id=? AND deleted_at IS NULL "
"ORDER BY COALESCE(sort_order, created_at) ASC, id ASC",
(page["id"],),
).fetchall()
return [dict(r) for r in rows]
def page_to_markdown(page: dict, *, include_children: bool = True) -> str:
"""Markdown for a single page, with optional sub-pages appended."""
parts = [f"# {_page_title(page)}", ""]
md_source = _page_markdown_source(page)
if md_source:
parts.append(md_source.strip())
else:
md = blocks_to_markdown(_page_blocks(page))
if md:
parts.append(md)
md = "\n\n".join(filter(None, parts)).rstrip()
if include_children:
for sub in _child_pages(page):
sub_md = page_to_markdown(sub, include_children=True)
if sub_md:
md += f"\n\n---\n\n{sub_md}"
return md
# ═══════════════ HTML ═══════════════
def blocks_to_html(blocks: list) -> str:
"""Convert a block array to a self-contained HTML fragment."""
parts: list[str] = []
for b in blocks or []:
t = b.get("type", "paragraph")
c = _text(b.get("content"))
if t == "heading_1":
parts.append(f'<h1 id="h-{_sanitize_id(b.get("id"))}">{c}</h1>')
elif t == "heading_2":
parts.append(f'<h2 id="h-{_sanitize_id(b.get("id"))}">{c}</h2>')
elif t == "heading_3":
parts.append(f'<h3 id="h-{_sanitize_id(b.get("id"))}">{c}</h3>')
elif t == "heading_4":
parts.append(f'<h4 id="h-{_sanitize_id(b.get("id"))}">{c}</h4>')
elif t == "bulleted_list":
parts.append(f"<li>{c}</li>")
elif t == "numbered_list":
parts.append(f"<li>{c}</li>")
elif t == "to_do":
checked = "checked" if b.get("checked") else ""
style = "text-decoration:line-through;opacity:.55;" if b.get("checked") else ""
parts.append(
f'<div class="todo"><input type="checkbox" {checked} disabled>'
f'<span style="{style}">{c}</span></div>'
)
elif t == "toggle":
children = blocks_to_html(b.get("children") or [])
parts.append(f"<details open><summary>{c}</summary>{children}</details>")
elif t == "quote":
parts.append(f"<blockquote>{c}</blockquote>")
elif t == "divider":
parts.append("<hr>")
elif t == "code":
lang = b.get("language") or ""
label = f'<div class="code-lang">{_text(lang)}</div>' if lang else ""
parts.append(f"<pre>{label}<code>{c}</code></pre>")
elif t == "math":
parts.append(f'<div class="math">\\[{c}\\]</div>')
elif t == "table_of_contents":
toc = [x for x in (blocks or [])
if x.get("type", "").startswith("heading_") and (x.get("content") or "").strip()]
if toc:
items = "".join(
f'<div style="margin-left:{max(0, int(x["type"].split("_")[-1]) - 1) * 14}px;">'
f'<a href="#h-{_sanitize_id(x.get("id"))}">{_text(x.get("content"))}</a></div>'
for x in toc
)
parts.append(f'<nav class="toc"><div class="toc-title">On this page</div>{items}</nav>')
elif t == "columns":
cols = "".join(
f'<div class="column">{blocks_to_html([child])}</div>'
for child in (b.get("children") or [])
)
parts.append(f'<div class="columns">{cols}</div>')
elif t == "callout":
icon = b.get("icon") or "💡"
bg = (b.get("style") or {}).get("bgColor", "#eef2ff")
parts.append(f'<div class="callout" style="background:{bg}"><span>{_text(icon, escape=False)}</span><div>{c}</div></div>')
elif t == "image":
src = b.get("src") or ""
alt = _text(b.get("alt"))
parts.append(f'<figure><img src="{src}" alt="{alt}"><figcaption>{alt}</figcaption></figure>')
else:
parts.append(f"<p>{c}</p>")
return "\n".join(parts)
def _standalone_css() -> str:
return """
:root{color-scheme:light;}
*{box-sizing:border-box;}
body{margin:0;font-family:system-ui,-apple-system,'Segoe UI',Roboto,sans-serif;color:#1f2328;background:#fff;line-height:1.65;}
.wrap{max-width:780px;margin:0 auto;padding:48px 32px 96px;}
h1{font-size:2.4rem;line-height:1.2;margin:0 0 8px;}
h2{font-size:1.7rem;border-bottom:1px solid #ececec;padding-bottom:6px;margin:32px 0 12px;}
h3{font-size:1.35rem;margin:24px 0 8px;}
h4{font-size:1.1rem;margin:20px 0 6px;}
p{margin:8px 0;}
li{margin:4px 0;}
ol{list-style:decimal;padding-left:24px;}
ul{list-style:disc;padding-left:24px;}
blockquote{border-left:4px solid #d0d7de;margin:12px 0;padding:4px 16px;color:#57606a;}
hr{border:none;border-top:1px solid #eaeef2;margin:24px 0;}
pre{background:#f6f8fa;border-radius:8px;padding:16px 20px;overflow-x:auto;font-size:14px;}
code{font-family:'SFMono-Regular',Consolas,monospace;background:#f6f8fa;border-radius:4px;padding:2px 5px;font-size:.9em;}
pre code{background:none;padding:0;font-size:13px;}
.code-lang{font-size:11px;color:#8b949e;text-transform:uppercase;letter-spacing:.5px;margin-bottom:8px;}
details{background:#f6f8fa;border:1px solid #eaeef2;border-radius:8px;padding:10px 14px;margin:10px 0;}
details summary{cursor:pointer;font-weight:600;}
details[open] summary{margin-bottom:8px;}
.todo{display:flex;align-items:flex-start;gap:8px;margin:4px 0;}
.todo input{margin-top:5px;}
.toc{border:1px solid #eaeef2;border-radius:8px;padding:16px 20px;margin:12px 0;}
.toc-title{font-size:12px;font-weight:700;text-transform:uppercase;letter-spacing:.5px;color:#57606a;margin-bottom:10px;}
.toc a{color:#0969da;text-decoration:none;display:block;padding:4px 0;}
.columns{display:flex;gap:14px;margin:12px 0;align-items:stretch;}
.column{flex:1;min-width:0;background:#f9fafb;border:1px solid #eaeef2;border-radius:8px;padding:12px 14px;box-sizing:border-box;}
.callout{display:flex;gap:10px;align-items:flex-start;border:1px solid #e0e7ff;border-radius:8px;padding:14px 18px;margin:12px 0;font-size:15px;}
.callout>span{font-size:20px;flex-shrink:0;}
.math{margin:14px 0;overflow-x:auto;}
figure{margin:16px 0;text-align:center;}
figure img{max-width:100%;border-radius:8px;}
figcaption{font-size:13px;color:#8b949e;margin-top:6px;}
.footer{margin-top:56px;padding-top:16px;border-top:1px solid #eaeef2;color:#8b949e;font-size:12px;display:flex;justify-content:space-between;}
a{color:#0969da;}
@media print{body{background:#fff;}.wrap{padding:0;max-width:100%;}}
"""
def page_to_standalone_html(
page: dict,
*,
include_children: bool = True,
base_url: str = "",
) -> str:
"""Return a standalone, self-contained HTML document for a page."""
title = _page_title(page)
body = blocks_to_html(_page_blocks(page))
meta_updated = page.get("updated_at") or ""
footer = f"<div class='footer'><span>FlowDeck · {_page_title(page)}</span><span>{meta_updated}</span></div>"
sub_html = ""
if include_children:
for sub in _child_pages(page):
sub_html += '\n<hr style="border:none">\n<div class="subpage">'
sub_html += page_to_standalone_html(sub, include_children=True, base_url=base_url)
sub_html += "</div>"
return f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{_text(title)}</title>
<style>{_standalone_css()}</style>
</head>
<body>
<div class="wrap">
<h1>{_text(title)}</h1>
{body}
{sub_html}
{footer}
</div>
</body>
</html>"""
# ═══════════════ PDF ═══════════════
def _pdf_html(page: dict) -> str:
"""A print-friendly, minimal-CSS HTML for PDF conversion."""
title = _page_title(page)
body = blocks_to_html(_page_blocks(page))
return f"""<html><head><meta charset="utf-8"><title>{_text(title)}</title>
<style>
body{{font-family:Helvetica,Arial,sans-serif;color:#1f2328;font-size:12px;line-height:1.5;}}
h1{{font-size:26px;margin:0 0 10px;}}
h2{{font-size:19px;border-bottom:1px solid #ddd;padding-bottom:4px;margin:22px 0 8px;}}
h3{{font-size:16px;margin:18px 0 6px;}}
h4{{font-size:14px;margin:14px 0 4px;}}
p,li{{margin:4px 0;}}
pre{{background:#f4f4f4;padding:10px;font-size:10px;white-space:pre-wrap;}}
code{{font-family:monospace;font-size:10px;}}
blockquote{{border-left:3px solid #ccc;margin:8px 0;padding:2px 12px;font-style:italic;}}
table{{border-collapse:collapse;width:100%;}}
hr{{border:none;border-top:1px solid #ddd;margin:16px 0;}}
.todo{{margin:4px 0;}}
.math{{font-style:italic;margin:10px 0;}}
.callout{{background:#f0f4ff;border:1px solid #dbe4ff;border-radius:6px;padding:8px 12px;margin:8px 0;}}
.footer{{margin-top:30px;padding-top:8px;border-top:1px solid #ddd;font-size:9px;color:#888;}}
</style></head><body>
<h1>{_text(title)}</h1>
{body}
<div class="footer">FlowDeck · {_text(title)} · {page.get("updated_at") or ""}</div>
</body></html>"""
def page_to_pdf_bytes(page: dict) -> bytes:
"""Render a page to a PDF (via xhtml2pdf)."""
from xhtml2pdf import pisa
src = _pdf_html(page)
buf = io.BytesIO()
pdf = pisa.CreatePDF(src, dest=buf, encoding="utf-8")
if pdf.err:
raise RuntimeError(f"PDF generation failed: {pdf.err}")
return buf.getvalue()
# ═══════════════ Static site (zip) ═══════════════
def _site_index_html(pages: list[dict]) -> str:
"""Build the index.html of the static site (list of all pages)."""
def link(p: dict) -> str:
title = _page_title(p)
return f'<li><a href="{quote(title, safe="")}.html">{_text(title)}</a></li>'
items = "".join(link(p) for p in pages)
return f"""<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8">
<title>FlowDeck Site</title>
<style>body{{font-family:system-ui,sans-serif;max-width:720px;margin:40px auto;padding:0 20px;color:#1f2328;}}
a{{color:#0969da;text-decoration:none;}}li{{margin:8px 0;}}</style></head>
<body><h1>FlowDeck Site</h1><ul>{items}</ul></body></html>"""
def build_static_site_bytes(root_page: dict) -> bytes:
"""Build a full static site as a zip: index.html + one HTML file per page."""
pages = [root_page] + _child_pages(root_page)
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
z.writestr("index.html", _site_index_html(pages))
for p in pages:
title = _page_title(p)
name = f"{quote(title, safe='')}.html"
z.writestr(name, page_to_standalone_html(p, include_children=False))
return buf.getvalue()