Admin-Panel: Schema-Designer, Dropzone mit Auto-Erkennung, Linkliste, Versionen

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

app/storage.py

@@ -0,0 +1,131 @@
+"""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)