Files
3d.lehrstun.de/feedback_modul/fastapi_app.py
Hans Puettmann 66c52cc8b1 PummelHummel/Alph in 3D-Druck-AG integriert
Quelle: D:/Claude-Projekte/PummelHummel/feedback_modul/ (Standard-Pattern,
auch in arduino.lehrstun.de + meta.lehrstun.de erprobt). 1:1 nach
3d.lehrstun.de/feedback_modul/ kopiert.

app.py erweitert:
- init_feedback_modul-Aufruf mit unseren Adaptern
- _alph_user_kontext: Lehrer = admin, SuS = profil mit Vorname+Tier,
  Anonym = kein Widget
- _alph_widget_aktiv: liest einstellung.feedback_widget_aktiv
- _alph_admin_unauthorized: Redirect zu admin.admin_login

Hart-Aus via ENV ALPH_DEAKTIVIERT=1 weiterhin moeglich.
Soft-Aus via einstellung.feedback_widget_aktiv='0' (per Admin-Inline-Edit).

DB:
- feedback-Tabelle via feedback_modul.schema.DDL angelegt (idempotent)
- einstellung.feedback_widget_aktiv auf '1' gesetzt (Alpha-Test-Modus)

Test:
- Eingeloggter SuS sieht das Alph-Widget auf /lektion/1
- Anonym (kein Login) sieht es nicht (user_kontext gibt None)
- Admin sieht es mit Rolle 'admin'

Naechster Schritt: Deploy auf Hetzner (3d.lehrstun.de), Markus loggt
sich als Test-Schueler ein und gibt Feedback zu den 14 Lektionen.
2026-05-25 15:29:39 +02:00

451 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""FastAPI-Adapter fuer das Alph-PummelHummel-Feedback-Modul.
Bietet aequivalente Routen wie der Flask-Blueprint, fuer FastAPI-Projekte.
Einbindung:
from feedback_modul.fastapi_app import init_feedback_modul
def get_db():
return sqlite3.connect("/pfad/zu/feedback.db", check_same_thread=False)
init_feedback_modul(
app,
get_db=get_db,
app_name="meta-lehrstun",
ist_admin=lambda request: request.cookies.get("admin_token") == ADMIN_TOKEN,
# user_kontext optional — fuer reine Drop-Ins kann das weg bleiben.
user_kontext=lambda request: {"site": "meta", "rolle": "user"},
widget_aktiv=lambda: True,
)
In static/index.html vor </body>:
<script src="/static/feedback/alph.js"
data-endpoint="/api/feedback"
data-app="meta-lehrstun"
data-user='{"site":"meta","rolle":"user"}'
defer></script>
"""
import hashlib
import json
import sqlite3
import time
from collections import defaultdict, OrderedDict
from datetime import datetime
from pathlib import Path
from fastapi import APIRouter, HTTPException, Request, Response, Form
from fastapi.responses import (HTMLResponse, JSONResponse, PlainTextResponse,
RedirectResponse, FileResponse)
from fastapi.staticfiles import StaticFiles
from .schema import DDL, KATEGORIEN
_config = {
"get_db": None,
"app_name": "unknown",
"ist_admin": lambda request: False,
"user_kontext": lambda request: None,
"widget_aktiv": lambda: True,
"admin_unauthorized": None, # Callable(request) -> Response
}
_rate_buckets = {} # ip -> list[timestamp]
_RATE_LIMIT_PER_MIN = 10
_STATIC_DIR = Path(__file__).parent / "static"
def _rate_limited(ip: str) -> bool:
now = time.time()
bucket = _rate_buckets.setdefault(ip, [])
bucket[:] = [t for t in bucket if now - t < 60]
if len(bucket) >= _RATE_LIMIT_PER_MIN:
return True
bucket.append(now)
return False
def _cluster_hash(url_pfad: str, kategorie: str, text: str) -> str:
norm = (text or "").lower().strip()[:30]
raw = f"{url_pfad}|{kategorie}|{norm}"
return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
def _admin_check(request: Request):
"""Liefert None wenn Admin OK, sonst eine Response (Redirect/403)."""
if _config["ist_admin"](request):
return None
handler = _config.get("admin_unauthorized")
if callable(handler):
return handler(request)
raise HTTPException(status_code=403, detail="forbidden")
def _hole_offene_feedbacks() -> list:
conn = _config["get_db"]()
conn.row_factory = sqlite3.Row
rows = conn.execute(
"""SELECT * FROM feedback
WHERE app_name = ? AND status NOT IN ('erledigt', 'wontfix', 'spam')
ORDER BY url_pfad, kategorie, erstellt_am""",
(_config["app_name"],),
).fetchall()
conn.close()
return [dict(r) for r in rows]
def _build_md(eintraege: list) -> str:
cluster = defaultdict(list)
for e in eintraege:
cluster[(e["url_pfad"], e["kategorie"], e["cluster_hash"])].append(e)
kat_label = {k[0]: k[1] for k in KATEGORIEN}
kat_emoji = {"code": "🐛", "link": "🔗", "verstehen": "", "fehlt": "", "sonstiges": "💬"}
nach_kategorie = defaultdict(list)
for key, eintrags in cluster.items():
nach_kategorie[key[1]].append((key, eintrags))
out = []
out.append(f"# Alph-Feedback {_config['app_name']} — Stand {datetime.now().strftime('%Y-%m-%d %H:%M')}")
out.append("")
out.append(f"**{len(eintraege)} offene Feedbacks** in {len(cluster)} Clustern.")
out.append("")
for kat in ["code", "link", "verstehen", "fehlt", "sonstiges"]:
clusterliste = nach_kategorie.get(kat, [])
if not clusterliste:
continue
emoji = kat_emoji.get(kat, "")
out.append(f"## {emoji} {kat_label.get(kat, kat)} ({sum(len(c[1]) for c in clusterliste)})")
out.append("")
for key, eintrags in clusterliste:
url, _, _ = key
n = len(eintrags)
mehrfach = f" — **{n}× gemeldet**" if n > 1 else ""
out.append(f"### `{url}`{mehrfach}")
out.append("")
for e in eintrags:
user = {}
try: user = json.loads(e["user_kontext"] or "{}")
except Exception: pass
tech = {}
try: tech = json.loads(e["technik_kontext"] or "{}")
except Exception: pass
team = user.get("tier") or user.get("site") or "?"
rolle = user.get("rolle", "")
rolle_str = f" ({rolle})" if rolle else ""
viewport = tech.get("viewport", "")
out.append(f"- **{team}**{rolle_str} — *\"{e['text']}\"*")
meta = [str(e["erstellt_am"])]
if viewport:
meta.append(f"viewport {viewport}")
if e.get("element_selector"):
meta.append(f"@ `{e['element_selector']}`")
out.append(f" <sub>{' · '.join(meta)}</sub>")
out.append("")
return "\n".join(out)
def _build_router() -> APIRouter:
r = APIRouter()
@r.post("/api/feedback")
async def api_feedback_post(request: Request):
if _rate_limited(request.client.host if request.client else "unknown"):
return JSONResponse({"ok": False, "error": "rate_limit"}, status_code=429)
try:
data = await request.json()
except Exception:
data = {}
if not isinstance(data, dict):
data = {}
kategorie = (data.get("kategorie") or "").strip().lower()
text = (data.get("text") or "").strip()
url_pfad = (data.get("url_pfad") or "").strip()[:500]
valid_kats = {k[0] for k in KATEGORIEN}
if kategorie not in valid_kats:
return JSONResponse({"ok": False, "error": "kategorie_ungueltig"}, status_code=400)
if len(text) < 3:
return JSONResponse({"ok": False, "error": "text_zu_kurz"}, status_code=400)
if len(text) > 2000:
text = text[:2000]
if not url_pfad:
return JSONResponse({"ok": False, "error": "url_pfad_fehlt"}, status_code=400)
seitentitel = (data.get("seitentitel") or "")[:200]
element_selector = (data.get("element_selector") or "")[:300]
element_text = (data.get("element_text") or "")[:500]
# Server-Kontext hat Vorrang. Im Drop-In-Modus akzeptieren wir Client-Kontext.
user_kontext = _config["user_kontext"](request) or {}
if not user_kontext:
client_user = data.get("user_kontext")
if isinstance(client_user, dict):
user_kontext = {
str(k)[:40]: (str(v)[:120] if not isinstance(v, (int, float, bool)) else v)
for k, v in list(client_user.items())[:12]
}
technik_kontext = {
"viewport": (data.get("viewport") or "")[:20],
"sprache": (data.get("sprache") or "")[:10],
"ua": (request.headers.get("user-agent") or "")[:300],
"referer": (request.headers.get("referer") or "")[:500],
}
cluster = _cluster_hash(url_pfad, kategorie, text)
conn = _config["get_db"]()
conn.execute(
"""INSERT INTO feedback
(app_name, kategorie, text, url_pfad, seitentitel,
user_kontext, technik_kontext, element_selector, element_text,
cluster_hash)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(_config["app_name"], kategorie, text, url_pfad, seitentitel,
json.dumps(user_kontext, ensure_ascii=False),
json.dumps(technik_kontext, ensure_ascii=False),
element_selector, element_text, cluster),
)
conn.commit()
conn.close()
return JSONResponse({"ok": True})
@r.get("/admin/feedback", response_class=HTMLResponse)
async def admin_feedback_liste(request: Request, status: str = "offen", kategorie: str = ""):
unauth = _admin_check(request)
if unauth is not None:
return unauth
where = ["app_name = ?"]
params = [_config["app_name"]]
if status == "offen":
where.append("status NOT IN ('erledigt', 'wontfix', 'spam')")
elif status and status != "alle":
where.append("status = ?")
params.append(status)
if kategorie:
where.append("kategorie = ?")
params.append(kategorie)
sql = ("SELECT * FROM feedback WHERE " + " AND ".join(where)
+ " ORDER BY status='neu' DESC, erstellt_am DESC")
conn = _config["get_db"]()
conn.row_factory = sqlite3.Row
rows = conn.execute(sql, params).fetchall()
eintraege = [dict(r) for r in rows]
for e in eintraege:
try: e["user"] = json.loads(e["user_kontext"] or "{}")
except Exception: e["user"] = {}
try: e["technik"] = json.loads(e["technik_kontext"] or "{}")
except Exception: e["technik"] = {}
stats = OrderedDict()
for s in ["neu", "triagiert", "in_arbeit", "erledigt", "wontfix", "spam"]:
stats[s] = conn.execute(
"SELECT COUNT(*) FROM feedback WHERE app_name=? AND status=?",
(_config["app_name"], s),
).fetchone()[0]
conn.close()
return HTMLResponse(_render_admin_html(eintraege, stats, status, kategorie))
@r.post("/admin/feedback/{feedback_id}/status")
async def admin_feedback_status(feedback_id: int, request: Request,
status: str = Form(...), notiz: str = Form("")):
unauth = _admin_check(request)
if unauth is not None:
return unauth
erlaubt = {"neu", "triagiert", "in_arbeit", "erledigt", "wontfix", "spam"}
if status not in erlaubt:
raise HTTPException(status_code=400, detail="status_ungueltig")
conn = _config["get_db"]()
conn.execute(
"""UPDATE feedback SET status = ?, bearbeitet_notiz = ?,
aktualisiert_am = CURRENT_TIMESTAMP WHERE id = ?""",
(status, notiz or None, feedback_id),
)
conn.commit()
conn.close()
ref = request.headers.get("referer") or "/admin/feedback"
return RedirectResponse(ref, status_code=303)
@r.get("/admin/feedback/export.json")
async def admin_feedback_export_json(request: Request):
unauth = _admin_check(request)
if unauth is not None:
return unauth
eintraege = _hole_offene_feedbacks()
for e in eintraege:
try: e["user_kontext"] = json.loads(e["user_kontext"] or "{}")
except Exception: pass
try: e["technik_kontext"] = json.loads(e["technik_kontext"] or "{}")
except Exception: pass
return JSONResponse({
"app": _config["app_name"],
"exportiert_am": datetime.utcnow().isoformat() + "Z",
"anzahl": len(eintraege),
"feedbacks": eintraege,
})
@r.get("/admin/feedback/export.md")
async def admin_feedback_export_md(request: Request):
unauth = _admin_check(request)
if unauth is not None:
return unauth
return PlainTextResponse(_build_md(_hole_offene_feedbacks()),
media_type="text/markdown; charset=utf-8")
@r.get("/admin/feedback/claude-prompt.md")
async def admin_feedback_claude_prompt(request: Request):
unauth = _admin_check(request)
if unauth is not None:
return unauth
eintraege = _hole_offene_feedbacks()
md_body = _build_md(eintraege)
prompt = []
prompt.append("# Aufgabe fuer Claude Code: Alphatest-Feedback auswerten")
prompt.append("")
prompt.append(f"Du arbeitest im Repo der App **{_config['app_name']}**. Unten "
f"findest du {len(eintraege)} offene Feedbacks aus dem Alphatest, "
f"gesammelt vom Alph-PummelHummel-Widget.")
prompt.append("")
prompt.append("## Bitte arbeite das so ab:")
prompt.append("")
prompt.append("1. **Cluster lesen.** Mehrfach-Reports am gleichen Pfad sind Prioritaet.")
prompt.append("2. **Quelldateien anschauen.** URL `/foo` -> templates/foo.html etc.")
prompt.append("3. **Symptom vs. Ursache.** Die Reports beschreiben oft nur was sichtbar ist.")
prompt.append("4. **Pro Cluster ein Vorschlag:** Datei + Zeile + konkrete Aenderung.")
prompt.append("5. **Unklare Reports markieren.** Lieber zum Nachfragen flaggen als raten.")
prompt.append("6. **Spam erkennen.** Quatsch markiere mit STATUS:spam zum Wegklicken.")
prompt.append("")
prompt.append("## Antwortformat:")
prompt.append("")
prompt.append("```")
prompt.append("[ID|Cluster] Kategorie · Pfad")
prompt.append(" Diagnose: ...")
prompt.append(" Fix: Datei foo.html Zeile 42 — alt: '...' neu: '...'")
prompt.append(" Status-Vorschlag: in_arbeit | wontfix | spam | nachfragen")
prompt.append("```")
prompt.append("")
prompt.append("---")
prompt.append("")
prompt.append(md_body)
return PlainTextResponse("\n".join(prompt),
media_type="text/markdown; charset=utf-8")
return r
# Minimaler HTML-Renderer (kein Jinja-Zwang fuer FastAPI)
def _render_admin_html(eintraege, stats, status_filter, kategorie_filter):
from html import escape as h
kat_emoji = {"code": "🐛", "link": "🔗", "verstehen": "", "fehlt": "", "sonstiges": "💬"}
status_label = {"neu": "neu", "triagiert": "triagiert", "in_arbeit": "in Arbeit",
"erledigt": "erledigt", "wontfix": "wontfix", "spam": "spam"}
css = """body{font-family:-apple-system,sans-serif;max-width:1100px;margin:1rem auto;padding:0 1rem;color:#2C3E50}
.fb-stats{display:flex;flex-wrap:wrap;gap:.5rem;margin:.5rem 0 1rem}
.fb-stat{background:#F8F9FA;padding:.4rem .8rem;border-radius:6px;border:1px solid #DEE2E6;font-size:.88rem}
.fb-stat strong{color:#00979D}
.fb-card{border:1px solid #DEE2E6;border-radius:8px;padding:.8rem 1rem;margin-bottom:.8rem;background:white}
.fb-card.fb-neu{border-left:4px solid #E67E22}
.fb-head{display:flex;gap:.6rem;align-items:center;flex-wrap:wrap;margin-bottom:.4rem;font-size:.85rem;color:#7F8C8D}
.fb-kat{background:#EBF8F9;color:#00979D;padding:.15rem .5rem;border-radius:4px;font-weight:bold}
.fb-pfad{font-family:Consolas,monospace;background:#F4F6F7;padding:.1rem .4rem;border-radius:3px}
.fb-text{font-size:.95rem;line-height:1.4;margin:.4rem 0}
.fb-statusform{display:flex;gap:.4rem;flex-wrap:wrap;margin-top:.6rem;padding-top:.5rem;border-top:1px solid #ECEFF1}
.fb-statusform select,.fb-statusform input,.fb-statusform button{padding:.3rem .5rem;border:1px solid #DEE2E6;border-radius:4px;font-size:.85rem}
.fb-statusform input{flex:1;min-width:180px}
.fb-statusform button{background:#00979D;color:white;cursor:pointer}
.fb-exports a{margin-right:.5rem}"""
parts = [f"<!doctype html><html><head><meta charset='utf-8'><title>Alph-Feedback {h(_config['app_name'])}</title><style>{css}</style></head><body>"]
parts.append(f"<h1>🐝 Alph-Feedback — {h(_config['app_name'])}</h1>")
parts.append("<div class='fb-stats'>")
for s, anz in stats.items():
parts.append(f"<div class='fb-stat'><strong>{anz}</strong> {h(status_label.get(s,s))}</div>")
parts.append("</div>")
parts.append("<p class='fb-exports'>"
"<a href='./feedback/export.json' target='_blank'>export.json</a> · "
"<a href='./feedback/export.md' target='_blank'>export.md</a> · "
"<a href='./feedback/claude-prompt.md' target='_blank'>claude-prompt.md</a></p>")
if not eintraege:
parts.append("<p>Keine Feedbacks im aktuellen Filter.</p>")
for e in eintraege:
u = e.get("user", {})
team = u.get("tier") or u.get("site") or "?"
kasten = u.get("kasten")
team_str = f"{team}" + (f" / Kasten {kasten}" if kasten else "")
rolle = u.get("rolle", "")
parts.append(f"<div class='fb-card{(' fb-neu' if e['status']=='neu' else '')}'>")
parts.append("<div class='fb-head'>")
parts.append(f"<span class='fb-kat'>{kat_emoji.get(e['kategorie'],'')} {h(e['kategorie'])}</span>")
parts.append(f"<span><strong>{h(team_str)}</strong>{(' ('+h(rolle)+')' if rolle else '')}</span>")
parts.append(f"<span class='fb-pfad'>{h(e['url_pfad'])}</span>")
parts.append(f"<span style='margin-left:auto'>{h(str(e['erstellt_am']))}</span>")
parts.append(f"<span>· Status: {h(status_label.get(e['status'],e['status']))}</span>")
parts.append("</div>")
parts.append(f"<div class='fb-text'>{h(e['text'])}</div>")
parts.append(f"<form class='fb-statusform' method='post' action='./feedback/{e['id']}/status'>")
parts.append("<select name='status'>")
for s in ["neu", "triagiert", "in_arbeit", "erledigt", "wontfix", "spam"]:
sel = " selected" if e['status'] == s else ""
parts.append(f"<option value='{s}'{sel}>{h(status_label.get(s,s))}</option>")
parts.append("</select>")
parts.append(f"<input type='text' name='notiz' value='{h(e.get('bearbeitet_notiz') or '')}' placeholder='Notiz / Commit-SHA'>")
parts.append("<button type='submit'>Setzen</button>")
parts.append("</form>")
parts.append("</div>")
parts.append("</body></html>")
return "".join(parts)
def init_feedback_modul(app, *, get_db, app_name, ist_admin=None,
user_kontext=None, widget_aktiv=None,
admin_unauthorized=None,
static_mount: str = "/static/feedback"):
"""Registriert das Modul in einer FastAPI-App.
Args:
app: FastAPI-Instanz
get_db: Callable () -> sqlite3.Connection
app_name: Identifier (z.B. "meta-lehrstun")
ist_admin: Callable(request) -> bool — pro Request, mit Cookies/Header
user_kontext: Callable(request) -> dict | None
widget_aktiv: Callable() -> bool (default: True). Drop-In-typisch:
einfach True, weil der serverseitige Widget-Snippet hier
nicht gerendert wird (siehe README).
admin_unauthorized: Callable(request) -> Response (z.B. RedirectResponse)
static_mount: URL-Prefix fuer alph.js (default /static/feedback).
Wird automatisch gemounted; falls schon belegt, mit
einem anderen Prefix aufrufen.
"""
_config["get_db"] = get_db
_config["app_name"] = app_name
if ist_admin is not None:
_config["ist_admin"] = ist_admin
if user_kontext is not None:
_config["user_kontext"] = user_kontext
if widget_aktiv is not None:
_config["widget_aktiv"] = widget_aktiv
if admin_unauthorized is not None:
_config["admin_unauthorized"] = admin_unauthorized
# Schema sicherstellen
conn = get_db()
conn.executescript(DDL)
conn.commit()
conn.close()
# Statische alph.js mounten
app.mount(static_mount, StaticFiles(directory=str(_STATIC_DIR)), name="feedback_static")
# Routen registrieren
app.include_router(_build_router())