"""Document final : un seul gabarit HTML pour l'aperçu et le PDF, ajustement sur une page mesuré par WeasyPrint."""
from __future__ import annotations
import hashlib
import json
import secrets
import subprocess
import tempfile
import threading
from collections import OrderedDict
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal
from jinja2 import Environment, FileSystemLoader, select_autoescape
from markupsafe import Markup, escape
from sqlalchemy.orm import Session
from ..core import isoweek
from ..models import STATUS_LABELS, BilanVersion
from ..services import bilans as bilan_service
from ..services import settings as settings_service
from . import branding
ASSETS = Path(__file__).resolve().parent.parent / "assets"
FONT_URL = "/api/assets/fonts"
MIN_FONT_PT = 9.5 # le texte des observations descend au plus à 9 pt (MIN_FONT_PT - 0,5)
PAGES = { # largeur × hauteur en mm (portrait)
"A4": (210.0, 297.0),
"A3": (297.0, 420.0),
}
@dataclass(frozen=True)
class Level:
name: str
description: str
font: float
pad: float
margin: float
obs_col: float
head: float
# Ordre imposé par le cahier des charges : espacements, marges, colonnes, taille du texte (plancher lisible).
LEVELS = [
Level("standard", "Mise en page standard", 11.0, 3.0, 15, 50, 1.0),
Level("espacements", "Espacements verticaux resserrés", 11.0, 1.6, 15, 50, 0.85),
Level("marges", "Marges réduites à 10 mm", 11.0, 1.6, 10, 50, 0.85),
Level("colonnes", "Colonne Observations élargie", 11.0, 1.5, 10, 58, 0.85),
Level("texte", "Texte légèrement réduit (10 pt)", 10.0, 1.3, 10, 58, 0.8),
Level("texte_min", f"Texte réduit au minimum lisible ({MIN_FONT_PT} pt)", MIN_FONT_PT, 1.1, 10, 60, 0.75),
]
@dataclass
class DocOptions:
paper: Literal["A4", "A3"] = "A4"
orientation: Literal["portrait", "paysage"] = "portrait"
fit: bool = True
allow_landscape: bool = False
include_logo: bool = True
include_observations: bool = True
include_details: bool = True
include_chart: bool = False
pagination: bool = True
show_generated_at: bool = True
title: str | None = None
indicator_ids: list[int] | None = None
version: int | None = None
def cache_key(self) -> str:
return json.dumps(asdict(self), sort_keys=True)
@dataclass
class Layout:
level: str
level_index: int
orientation: str
paper: str
font: float
pad: float
margin: float
label_col: float
num_col: float
head: float
pages: int = 1
steps: list[str] = field(default_factory=list)
fits: bool = True
culprits: list[dict[str, Any]] = field(default_factory=list)
suggestions: list[str] = field(default_factory=list)
measured: bool = False
def _layout_for(level_index: int, orientation: str, paper: str, observations: bool) -> Layout:
lv = LEVELS[level_index]
if observations:
num = 11.0 if orientation == "portrait" else 9.0
label = 100 - lv.obs_col - num
else:
num, label = 22.0, 78.0
return Layout(level=lv.name, level_index=level_index, orientation=orientation, paper=paper, font=lv.font,
pad=lv.pad, margin=lv.margin, label_col=label, num_col=num, head=lv.head)
# --- Données -------------------------------------------------------------------------------------
def _fr_number(value: str) -> str:
return value.replace(".", ",")
def details_text(kind: str | None, unit: str, details: list[dict[str, Any]]) -> str:
if not details:
return ""
unit = (details[0].get("unit") or unit or "").strip()
n = len(details)
if kind == "taux":
head = "Taux retenus" if n > 1 else "Taux retenu"
elif kind == "vitesse":
head = "Vitesses retenues" if n > 1 else "Vitesse retenue"
else:
head = "Détails"
values = " · ".join(_fr_number(str(d["value"])) for d in details)
return f"{head} : {values}{(' ' + unit) if unit else ''}"
def document_data(db: Session, iso_year: int, iso_week: int, opts: DocOptions) -> dict[str, Any]:
payload = bilan_service.bilan_payload(db, iso_year, iso_week)
bilan = bilan_service.find_bilan(db, iso_year, iso_week)
rows = payload["rows"]
official = False
status_line = "Document de travail — bilan non validé"
version_no = None
if bilan is not None and opts.version:
version = db.query(BilanVersion).filter_by(bilan_id=bilan.id, version_no=opts.version).one_or_none()
if version is None:
raise bilan_service.BilanError(404, "Version introuvable.", "not_found")
rows, official, version_no = version.snapshot["rows"], True, version.version_no
at = datetime.fromisoformat(version.snapshot["validated_at"])
status_line = (f"Version {version_no} validée le {isoweek.fr_datetime(at)} par "
f"{version.snapshot.get('validated_by', '—')}")
elif bilan is not None and bilan.status in {"valide", "archive"}:
version = bilan_service.latest_version(db, bilan.id)
if version is not None:
rows, official, version_no = version.snapshot["rows"], True, version.version_no
at = datetime.fromisoformat(version.snapshot["validated_at"])
status_line = (f"Bilan validé le {isoweek.fr_datetime(at)} par "
f"{version.snapshot.get('validated_by', '—')} · version {version_no}")
elif bilan is not None and bilan.status == "rouvert":
status_line = (f"Bilan rouvert le {isoweek.fr_datetime(bilan.reopened_at)} — modifications non validées"
if bilan.reopened_at else "Bilan rouvert — modifications non validées")
if opts.indicator_ids:
wanted = set(opts.indicator_ids)
rows = [r for r in rows if r["indicator_id"] in wanted]
out_rows = [{
"code": r["code"], "label": r["label"], "value": r["value"],
"observation": r["observation"] if opts.include_observations else "",
"details_text": details_text(r.get("detail_kind"), r.get("detail_unit") or "", r.get("details") or [])
if opts.include_details and opts.include_observations else "",
} for r in rows]
start, end = isoweek.week_bounds(iso_year, iso_week)
status = payload["status"]
return {
"iso_year": iso_year, "iso_week": iso_week, "start": isoweek.fr_date(start), "end": isoweek.fr_date(end),
"rows": out_rows, "official": official, "status": status,
"status_label": "version validée" if official else STATUS_LABELS.get(status, status),
"status_line": status_line, "version_no": version_no,
"settings": {k: settings_service.get(db, k) for k in
("document_title", "document_section", "document_footer", "diffusion_mention")},
}
def chart_svg(rows: list[dict[str, Any]]) -> str:
data = [(r["label"], r["value"]) for r in rows]
if not data:
return ""
vmax = max([v for _, v in data if v is not None] or [0]) or 1
row_h, label_w, width = 15, 230, 700
height = row_h * len(data) + 6
parts = [f'")
return "".join(parts)
# --- Rendu ---------------------------------------------------------------------------------------
def _nl2br(value: str) -> Markup:
return Markup("
").join(escape(line) for line in str(value).split("\n"))
_env = Environment(loader=FileSystemLoader(str(Path(__file__).parent / "templates")),
autoescape=select_autoescape(["html", "j2"]), trim_blocks=True, lstrip_blocks=True)
_env.filters["nl2br"] = _nl2br
def page_dims(paper: str, orientation: str) -> dict[str, Any]:
w, h = PAGES[paper]
if orientation == "paysage":
w, h = h, w
return {"width": w, "height": h,
"css_size": f"{paper} {'landscape' if orientation == 'paysage' else 'portrait'}"}
def render_html(data: dict[str, Any], opts: DocOptions, layout: Layout, *, screen: bool, nonce: str | None = None,
generated: datetime | None = None) -> str:
s = data["settings"]
title = (opts.title or s["document_title"]).strip() or s["document_title"]
gen = generated or datetime.now(timezone.utc)
return _env.get_template("document.html.j2").render(
title=title, section=s["document_section"], footer=s["document_footer"],
diffusion=s["diffusion_mention"].strip(), week=data["iso_week"], year=data["iso_year"],
start=data["start"], end=data["end"], rows=data["rows"], official=data["official"],
status_label=data["status_label"], status_line=data["status_line"],
has_empty=any(r["value"] is None for r in data["rows"]),
logo=branding.data_uri() if opts.include_logo else None,
chart_svg=Markup(chart_svg(data["rows"])) if opts.include_chart else None,
generated=isoweek.fr_datetime(gen), opts=opts, layout=layout,
page=page_dims(layout.paper, layout.orientation), screen=screen, nonce=nonce, font_base=FONT_URL,
)
def _url_fetcher(url: str) -> dict[str, Any]:
"""Aucune ressource réseau : seules les polices locales et les URI data: sont résolues."""
from weasyprint import default_url_fetcher
if url.startswith("data:"):
return default_url_fetcher(url)
for prefix in (FONT_URL + "/", "file://" + FONT_URL + "/"):
if url.startswith(prefix):
name = url[len(prefix):]
path = (ASSETS / "fonts" / name).resolve()
if path.parent == (ASSETS / "fonts").resolve() and path.is_file():
return {"string": path.read_bytes(), "mime_type": "font/ttf"}
raise ValueError(f"Ressource externe refusée : {url[:80]}")
_render_lock = threading.Lock() # WeasyPrint/Pango ne sont pas garantis thread-safe
def _weasy_document(html: str) -> Any:
from weasyprint import HTML
with _render_lock:
return HTML(string=html, url_fetcher=_url_fetcher, base_url="/").render()
def _row_heights_mm(document: Any) -> dict[str, float]:
heights: dict[str, float] = {}
def walk(box: Any) -> None:
if getattr(box, "element_tag", None) == "tr":
el = getattr(box, "element", None)
code = el.get("data-row") if el is not None else None
h = getattr(box, "height", None)
if code and isinstance(h, (int, float)):
heights[code] = heights.get(code, 0.0) + h * 25.4 / 96
for child in getattr(box, "children", None) or []:
walk(child)
for page in document.pages:
walk(page._page_box)
return heights
_cache: OrderedDict[str, Layout] = OrderedDict()
_CACHE_MAX = 64
def _measure(data: dict[str, Any], opts: DocOptions, layout: Layout) -> tuple[int, Any]:
html = render_html(data, opts, layout, screen=False, generated=datetime(2000, 1, 1, tzinfo=timezone.utc))
doc = _weasy_document(html)
return len(doc.pages), doc
def compute_layout(data: dict[str, Any], opts: DocOptions) -> Layout:
key = hashlib.sha256((json.dumps(data, sort_keys=True, default=str) + opts.cache_key()).encode()).hexdigest()
if key in _cache:
_cache.move_to_end(key)
return _cache[key]
obs = opts.include_observations
orientations = [opts.orientation]
if opts.fit and opts.allow_landscape and opts.orientation == "portrait":
orientations.append("paysage")
tried: list[tuple[Layout, Any]] = []
chosen: Layout | None = None
chosen_doc: Any = None
if not opts.fit:
lay = _layout_for(0, opts.orientation, opts.paper, obs)
lay.pages, chosen_doc = _measure(data, opts, lay)
chosen = lay
else:
for orient in orientations:
first = _layout_for(0, orient, opts.paper, obs)
first.pages, doc = _measure(data, opts, first)
tried.append((first, doc))
if first.pages == 1:
chosen, chosen_doc = first, doc
break
last = _layout_for(len(LEVELS) - 1, orient, opts.paper, obs)
last.pages, doc_last = _measure(data, opts, last)
tried.append((last, doc_last))
if last.pages > 1:
continue
for i in range(1, len(LEVELS) - 1):
lay = _layout_for(i, orient, opts.paper, obs)
lay.pages, doc = _measure(data, opts, lay)
tried.append((lay, doc))
if lay.pages == 1:
chosen, chosen_doc = lay, doc
break
if chosen is None:
chosen, chosen_doc = last, doc_last
break
if chosen is None:
# Rien ne tient sur une page : on garde la variante la plus compacte dans l'orientation demandée.
candidates = [t for t in tried if t[0].orientation == opts.orientation] or tried
chosen, chosen_doc = min(candidates, key=lambda t: (t[0].pages, -t[0].level_index))
assert chosen is not None
chosen.measured = True
chosen.fits = chosen.pages == 1
chosen.steps = [LEVELS[i].description for i in range(1, chosen.level_index + 1)]
if chosen.orientation != opts.orientation:
chosen.steps.append("Passage en paysage (autorisé)")
if not chosen.fits:
heights = _row_heights_mm(chosen_doc)
ordered = sorted(heights.items(), key=lambda kv: kv[1], reverse=True)
median = sorted(heights.values())[len(heights) // 2] if heights else 0
by_code = {r["code"]: r for r in data["rows"]}
chosen.culprits = [{"code": c, "label": by_code[c]["label"], "height_mm": round(h, 1),
"observation_length": len(by_code[c]["observation"] or "")}
for c, h in ordered if c in by_code and h > max(median * 1.6, 9)][:5]
sugg = []
if chosen.culprits:
sugg.append("Raccourcir les observations les plus longues : "
+ ", ".join(c["label"] for c in chosen.culprits[:3]) + ".")
if opts.orientation == "portrait" and not opts.allow_landscape:
sugg.append("Autoriser le passage en paysage.")
if opts.paper == "A4":
sugg.append("Choisir le format A3.")
sugg.append(f"Exporter exceptionnellement sur {chosen.pages} pages (aucune donnée n'est tronquée).")
chosen.suggestions = sugg
_cache[key] = chosen
while len(_cache) > _CACHE_MAX:
_cache.popitem(last=False)
return chosen
def preview_html(db: Session, iso_year: int, iso_week: int, opts: DocOptions) -> tuple[str, Layout, str]:
data = document_data(db, iso_year, iso_week, opts)
layout = compute_layout(data, opts)
nonce = secrets.token_urlsafe(16)
return render_html(data, opts, layout, screen=True, nonce=nonce), layout, nonce
def layout_info(layout: Layout) -> dict[str, Any]:
return {"pages": layout.pages, "fits": layout.fits, "level": layout.level, "orientation": layout.orientation,
"paper": layout.paper, "font_pt": layout.font, "margin_mm": layout.margin, "steps": layout.steps,
"culprits": layout.culprits, "suggestions": layout.suggestions, "min_font_pt": MIN_FONT_PT - 0.5,
"page": page_dims(layout.paper, layout.orientation)}
def count_pdf_pages(pdf: bytes) -> int:
"""Contrôle indépendant du nombre de pages réel, avec Poppler (pdfinfo)."""
with tempfile.NamedTemporaryFile(suffix=".pdf") as f:
f.write(pdf)
f.flush()
out = subprocess.run(["pdfinfo", f.name], capture_output=True, text=True, timeout=20, check=False)
for line in out.stdout.splitlines():
if line.startswith("Pages:"):
return int(line.split()[1])
raise RuntimeError("Impossible de compter les pages du PDF.")
def render_pdf(db: Session, iso_year: int, iso_week: int, opts: DocOptions) -> tuple[bytes, Layout, dict[str, Any]]:
data = document_data(db, iso_year, iso_week, opts)
layout = compute_layout(data, opts)
html = render_html(data, opts, layout, screen=False)
doc = _weasy_document(html)
title = f"{data['settings']['document_title']} — Semaine {iso_week} ({iso_year})"
with _render_lock:
doc.metadata.title = title
doc.metadata.authors = ["EDCF 52 — Bilan d'activité"]
pdf = doc.write_pdf()
pages = count_pdf_pages(pdf)
if pages != layout.pages:
layout.pages = pages
layout.fits = pages == 1
return pdf, layout, data
def pdf_page_png(pdf: bytes, page: int, dpi: int = 110) -> bytes:
with tempfile.TemporaryDirectory() as d:
src = Path(d) / "doc.pdf"
src.write_bytes(pdf)
out = subprocess.run(["pdftoppm", "-f", str(page), "-l", str(page), "-r", str(dpi), "-png", "-singlefile",
str(src), str(Path(d) / "page")], capture_output=True, timeout=30, check=False)
target = Path(d) / "page.png"
if out.returncode != 0 or not target.is_file():
raise RuntimeError("Rendu de l'aperçu PDF impossible.")
return target.read_bytes()