188 lines
6.2 KiB
Python
188 lines
6.2 KiB
Python
"""Kern-Engine: Schema + CSV -> validierte Daten -> gerenderte HTML-Seite.
|
|
|
|
Bewusst framework-frei gehalten, damit sie ohne Flask testbar bleibt.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import segno
|
|
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
|
|
|
COLUMN_TYPES = ("text", "text_long", "number", "date", "url", "rating_5")
|
|
|
|
TYPE_LABELS = {
|
|
"text": "Text (kurz)",
|
|
"text_long": "Text (lang)",
|
|
"number": "Zahl",
|
|
"date": "Datum",
|
|
"url": "Link (mit QR-Code)",
|
|
"rating_5": "Bewertung (1-5 Sterne)",
|
|
}
|
|
|
|
TEMPLATE_DIR = Path(__file__).parent / "templates"
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
_env = Environment(
|
|
loader=FileSystemLoader(str(TEMPLATE_DIR)),
|
|
autoescape=select_autoescape(["html"]),
|
|
)
|
|
|
|
|
|
def slugify(text: str) -> str:
|
|
"""Titel -> URL-tauglicher Slug (Umlaute transliteriert)."""
|
|
text = text.strip().lower()
|
|
for src, dst in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")):
|
|
text = text.replace(src, dst)
|
|
text = re.sub(r"[^a-z0-9]+", "-", text)
|
|
return text.strip("-")
|
|
|
|
|
|
def column_key(label: str) -> str:
|
|
return slugify(label).replace("-", "_") or "spalte"
|
|
|
|
|
|
def _norm_label(label: str) -> str:
|
|
return " ".join(label.strip().lower().split())
|
|
|
|
|
|
def _decode(raw: bytes) -> str:
|
|
# Excel (Windows) liefert oft cp1252, "CSV UTF-8" liefert BOM.
|
|
for encoding in ("utf-8-sig", "cp1252"):
|
|
try:
|
|
return raw.decode(encoding)
|
|
except UnicodeDecodeError:
|
|
continue
|
|
return raw.decode("utf-8", errors="replace")
|
|
|
|
|
|
def _sniff_delimiter(text: str) -> str:
|
|
first_line = text.splitlines()[0] if text.splitlines() else ""
|
|
counts = {d: first_line.count(d) for d in (";", ",", "\t")}
|
|
best = max(counts, key=counts.get) # deutsches Excel: Semikolon
|
|
return best if counts[best] > 0 else ";"
|
|
|
|
|
|
def parse_csv(raw: bytes) -> tuple[list[str], list[list[str]]]:
|
|
"""Rohbytes -> (Kopfzeile, Datenzeilen). Leerzeilen werden verworfen."""
|
|
text = _decode(raw)
|
|
reader = csv.reader(io.StringIO(text), delimiter=_sniff_delimiter(text))
|
|
rows = [row for row in reader if any(cell.strip() for cell in row)]
|
|
if not rows:
|
|
return [], []
|
|
return [cell.strip() for cell in rows[0]], rows[1:]
|
|
|
|
|
|
def _check_cell(col: dict, value: str) -> str | None:
|
|
if not value:
|
|
return None # leere Zellen sind erlaubt
|
|
if col["type"] == "url" and not re.fullmatch(r"https?://\S+", value):
|
|
return "keine gueltige URL (muss mit http:// oder https:// beginnen)"
|
|
if col["type"] == "rating_5" and (not value.isdigit() or not 1 <= int(value) <= 5):
|
|
return "Bewertung muss eine ganze Zahl von 1 bis 5 sein"
|
|
if col["type"] == "number":
|
|
try:
|
|
float(value.replace(",", "."))
|
|
except ValueError:
|
|
return "keine Zahl"
|
|
return None
|
|
|
|
|
|
def validate_and_map(schema: dict, header: list[str], rows: list[list[str]]):
|
|
"""CSV-Inhalt gegen das Schema pruefen.
|
|
|
|
Rueckgabe: (records, errors, warnings) — records ist eine Liste
|
|
von Dicts key -> Zellwert, nur gefuellt wenn errors leer ist.
|
|
"""
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
norm_header = [_norm_label(h) for h in header]
|
|
|
|
col_index: dict[str, int] = {}
|
|
for col in schema["columns"]:
|
|
try:
|
|
col_index[col["key"]] = norm_header.index(_norm_label(col["label"]))
|
|
except ValueError:
|
|
errors.append(f'Spalte "{col["label"]}" fehlt in der CSV.')
|
|
|
|
known = {_norm_label(c["label"]) for c in schema["columns"]}
|
|
for h in header:
|
|
if _norm_label(h) not in known:
|
|
warnings.append(f'Unbekannte Spalte "{h}" wird ignoriert.')
|
|
|
|
if errors:
|
|
return [], errors, warnings
|
|
|
|
records: list[dict] = []
|
|
for lineno, row in enumerate(rows, start=2):
|
|
record = {}
|
|
for col in schema["columns"]:
|
|
idx = col_index[col["key"]]
|
|
value = row[idx].strip() if idx < len(row) else ""
|
|
problem = _check_cell(col, value)
|
|
if problem:
|
|
errors.append(f'Zeile {lineno}, Spalte "{col["label"]}": {problem}')
|
|
record[col["key"]] = value
|
|
records.append(record)
|
|
|
|
if errors:
|
|
return [], errors, warnings
|
|
return records, errors, warnings
|
|
|
|
|
|
def qr_svg(data: str) -> str:
|
|
"""QR-Code als Inline-SVG (skaliert per CSS, kein width/height-Attribut)."""
|
|
buf = io.BytesIO()
|
|
segno.make(data, error="m").save(
|
|
buf, kind="svg", xmldecl=False, omitsize=True, border=1, dark="#26221c"
|
|
)
|
|
return buf.getvalue().decode("utf-8")
|
|
|
|
|
|
def _shorten_url(url: str, max_len: int = 42) -> str:
|
|
display = re.sub(r"^https?://(www\.)?", "", url).rstrip("/")
|
|
return display if len(display) <= max_len else display[: max_len - 1] + "…"
|
|
|
|
|
|
def _build_cells(schema: dict, records: list[dict]) -> list[list[dict]]:
|
|
rows = []
|
|
for record in records:
|
|
cells = []
|
|
for col in schema["columns"]:
|
|
value = record[col["key"]]
|
|
cell = {"type": col["type"], "text": value}
|
|
if col["type"] == "url" and value:
|
|
cell["qr"] = qr_svg(value)
|
|
cell["short"] = _shorten_url(value)
|
|
if col["type"] == "rating_5" and value:
|
|
n = int(value)
|
|
cell["stars"] = "★" * n + "☆" * (5 - n)
|
|
cells.append(cell)
|
|
rows.append(cells)
|
|
return rows
|
|
|
|
|
|
def render_page(schema: dict, records: list[dict], version_label: str = "") -> str:
|
|
"""Fertige, in sich geschlossene HTML-Seite (CSS eingebettet)."""
|
|
css = (STATIC_DIR / "table.css").read_text(encoding="utf-8")
|
|
return _env.get_template("table.html").render(
|
|
schema=schema,
|
|
columns=schema["columns"],
|
|
rows=_build_cells(schema, records),
|
|
css=css,
|
|
version_label=version_label,
|
|
generated=datetime.now().strftime("%d.%m.%Y %H:%M"),
|
|
)
|
|
|
|
|
|
def template_csv(schema: dict) -> bytes:
|
|
"""Leere CSV-Vorlage: Semikolon + UTF-8-BOM, damit Excel sie direkt oeffnet."""
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf, delimiter=";", lineterminator="\r\n")
|
|
writer.writerow([col["label"] for col in schema["columns"]])
|
|
return buf.getvalue().encode("utf-8-sig")
|