132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""Ablage im Dateisystem: Schemas, CSV-Versionen, veroeffentlichte Seiten.
|
|
|
|
Layout:
|
|
data/tables/<slug>/schema.json
|
|
data/tables/<slug>/versions/<zeitstempel>.csv
|
|
public/<slug>/index.html (statisch, liefert Nginx aus)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from . import engine
|
|
|
|
DATA_DIR = Path(os.environ.get("QT_DATA_DIR", "data"))
|
|
PUBLIC_DIR = Path(os.environ.get("QT_PUBLIC_DIR", "public"))
|
|
BASE_URL = os.environ.get("QT_BASE_URL", "http://localhost:5000").rstrip("/")
|
|
|
|
|
|
def _table_dir(slug: str) -> Path:
|
|
return DATA_DIR / "tables" / slug
|
|
|
|
|
|
def public_url(slug: str) -> str:
|
|
return f"{BASE_URL}/{slug}/"
|
|
|
|
|
|
def load_schema(slug: str) -> dict | None:
|
|
path = _table_dir(slug) / "schema.json"
|
|
if not path.exists():
|
|
return None
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def save_schema(schema: dict) -> None:
|
|
table_dir = _table_dir(schema["slug"])
|
|
(table_dir / "versions").mkdir(parents=True, exist_ok=True)
|
|
(table_dir / "schema.json").write_text(
|
|
json.dumps(schema, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
|
|
def list_tables() -> list[dict]:
|
|
"""Alle Tabellen mit Versions-Info, neueste zuerst."""
|
|
tables_dir = DATA_DIR / "tables"
|
|
if not tables_dir.exists():
|
|
return []
|
|
tables = []
|
|
for table_dir in sorted(tables_dir.iterdir()):
|
|
schema = load_schema(table_dir.name)
|
|
if schema is None:
|
|
continue
|
|
versions = list_versions(schema["slug"])
|
|
tables.append(
|
|
{
|
|
"schema": schema,
|
|
"slug": schema["slug"],
|
|
"title": schema["title"],
|
|
"url": public_url(schema["slug"]),
|
|
"versions": versions,
|
|
"last_upload": versions[0] if versions else None,
|
|
"published": (PUBLIC_DIR / schema["slug"] / "index.html").exists(),
|
|
}
|
|
)
|
|
tables.sort(key=lambda t: t["last_upload"] or "", reverse=True)
|
|
return tables
|
|
|
|
|
|
def list_versions(slug: str) -> list[str]:
|
|
versions_dir = _table_dir(slug) / "versions"
|
|
if not versions_dir.exists():
|
|
return []
|
|
return sorted((p.name for p in versions_dir.glob("*.csv")), reverse=True)
|
|
|
|
|
|
def read_version(slug: str, filename: str) -> bytes | None:
|
|
path = _table_dir(slug) / "versions" / Path(filename).name
|
|
return path.read_bytes() if path.exists() else None
|
|
|
|
|
|
def find_table(filename: str, header: list[str]) -> str | None:
|
|
"""Upload einer Tabelle zuordnen: erst Dateiname, dann Header-Signatur."""
|
|
tables = list_tables()
|
|
stem = engine.slugify(Path(filename).stem)
|
|
for t in tables:
|
|
if stem == t["slug"] or stem.startswith(t["slug"]):
|
|
return t["slug"]
|
|
norm_header = {" ".join(h.strip().lower().split()) for h in header if h.strip()}
|
|
matches = [
|
|
t["slug"]
|
|
for t in tables
|
|
if {" ".join(c["label"].strip().lower().split()) for c in t["schema"]["columns"]}
|
|
<= norm_header
|
|
]
|
|
return matches[0] if len(matches) == 1 else None
|
|
|
|
|
|
def publish_upload(slug: str, raw: bytes) -> tuple[bool, list[str], list[str], str]:
|
|
"""CSV validieren, als Version ablegen und die Seite neu veroeffentlichen.
|
|
|
|
Rueckgabe: (ok, errors, warnings, version_name). Bei Fehlern wird
|
|
nichts gespeichert und nichts veroeffentlicht.
|
|
"""
|
|
schema = load_schema(slug)
|
|
if schema is None:
|
|
return False, [f'Tabelle "{slug}" existiert nicht.'], [], ""
|
|
|
|
header, rows = engine.parse_csv(raw)
|
|
records, errors, warnings = engine.validate_and_map(schema, header, rows)
|
|
if errors:
|
|
return False, errors, warnings, ""
|
|
|
|
stamp = datetime.now()
|
|
version_name = stamp.strftime("%Y-%m-%d_%H-%M-%S") + ".csv"
|
|
(_table_dir(slug) / "versions" / version_name).write_bytes(raw)
|
|
|
|
html = engine.render_page(
|
|
schema, records, version_label=stamp.strftime("%d.%m.%Y, %H:%M Uhr")
|
|
)
|
|
page_dir = PUBLIC_DIR / slug
|
|
page_dir.mkdir(parents=True, exist_ok=True)
|
|
(page_dir / "index.html").write_text(html, encoding="utf-8")
|
|
return True, [], warnings, version_name
|
|
|
|
|
|
def delete_table(slug: str) -> None:
|
|
shutil.rmtree(_table_dir(slug), ignore_errors=True)
|
|
shutil.rmtree(PUBLIC_DIR / slug, ignore_errors=True)
|