StefanandClaude Fable 5
c0a18ac735
Drei P3-Funde aus der Durchsicht vom 14.09.: Reservieren: Statusprüfung und Schreiben waren getrennt - zwei gleichzeitige Anfragen konnten beide passieren, die zweite überschrieb Name und Token der ersten, ohne dass die es erfuhr. Jetzt entscheidet ein UPDATE mit Status-Bedingung; der Verlierer bekommt None und die Route meldet "schon weg" (Seite) bzw. 409 (API). /?limit=abc lieferte jedem anonymen Besucher einen internen Serverfehler, limit=-1 hiess in SQLite "alles". _limit_lesen() fällt bei Unsinn auf die Seitengrösse zurück und deckelt bei 500. Erfassen/Bearbeiten: int(category_id) und die Pydantic-Prüfung warfen im Handler - 500 statt Fehlermeldung. Jetzt Meldung; ausserdem werden die Angaben VOR den Bildern geprüft, damit bei abgelehnten Angaben keine verwaisten Bilddateien liegen bleiben. Vier neue Tests, jeder einmal gegen den alten Code gelaufen und dabei rot geworden. 84 lokal grün (HEIC-Test braucht pillow-heif, Docker). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
490 lines
17 KiB
Python
490 lines
17 KiB
Python
"""Die sichtbaren Seiten.
|
|
|
|
Gerendert mit Jinja2, HTMX tauscht beim Filtern nur die Liste aus. Alle
|
|
Formulare tragen ein CSRF-Token: SameSite=Lax schützt zwar schon gegen
|
|
seitenfremde POSTs, aber bei Formularen ist die zusätzliche Prüfung billig
|
|
und fängt auch ältere Browser ab.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import crud, images, security
|
|
from app.config import einstellungen
|
|
from app.database import get_db
|
|
from app.models import Item, Status
|
|
from app.schemas import ItemAendern, ItemAnlegen
|
|
from app.version import version_info
|
|
|
|
router = APIRouter(include_in_schema=False)
|
|
|
|
vorlagen = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templates"))
|
|
|
|
SEITE = 24 # Kacheln pro Nachladen
|
|
|
|
|
|
# --------------------------------------------------------------- Anzeige ---
|
|
|
|
GESCHLECHT_TEXT = {"girl": "Mädchen", "boy": "Buben", "unisex": "egal"}
|
|
ZUSTAND_TEXT = {"new": "neu", "very_good": "sehr gut", "good": "gut", "worn": "getragen"}
|
|
SAISON_TEXT = {
|
|
"spring_summer": "Frühling/Sommer",
|
|
"autumn_winter": "Herbst/Winter",
|
|
"all_year": "ganzjährig",
|
|
}
|
|
|
|
|
|
def _groesse_text(item: Item) -> str:
|
|
teile = [item.size] if item.size else []
|
|
if item.kategorie:
|
|
teile.append(item.kategorie.name)
|
|
return " · ".join(teile) or "ohne Angabe"
|
|
|
|
|
|
def _umgebung(request: Request, db: Session, **extra) -> dict:
|
|
"""Gemeinsame Werte für jede Seite."""
|
|
betreiber = security.ist_betreiber(request)
|
|
entwurf_anzahl = 0
|
|
if betreiber:
|
|
_, entwurf_anzahl = crud.items_suchen(db, status=Status.draft.value, limit=1)
|
|
|
|
return {
|
|
"request": request,
|
|
"betreiber": betreiber,
|
|
"entwurf_anzahl": entwurf_anzahl,
|
|
"csrf": security.csrf_token(request),
|
|
"geschlecht_text": lambda w: GESCHLECHT_TEXT.get(w, w),
|
|
"zustand_text": lambda w: ZUSTAND_TEXT.get(w, w),
|
|
"saison_text": lambda w: SAISON_TEXT.get(w, w),
|
|
"groesse_text": _groesse_text,
|
|
"version": version_info(),
|
|
"meldung": request.query_params.get("meldung"),
|
|
"meldung_art": request.query_params.get("art", "ok"),
|
|
**extra,
|
|
}
|
|
|
|
|
|
def _csrf_oder_fehler(request: Request, csrf: str) -> None:
|
|
if not security.csrf_pruefen(request, csrf):
|
|
raise HTTPException(
|
|
status_code=400, detail="Das Formular ist abgelaufen. Bitte neu laden."
|
|
)
|
|
|
|
|
|
def _weiter(pfad: str, meldung: str | None = None, art: str = "ok") -> RedirectResponse:
|
|
"""Nach jedem POST eine Weiterleitung (Post/Redirect/Get).
|
|
|
|
Ohne das würde ein Neuladen der Seite das Formular erneut abschicken -
|
|
beim Reservieren wäre das besonders unschön.
|
|
"""
|
|
from urllib.parse import quote
|
|
|
|
if meldung:
|
|
trenner = "&" if "?" in pfad else "?"
|
|
pfad = f"{pfad}{trenner}meldung={quote(meldung)}&art={art}"
|
|
return RedirectResponse(pfad, status_code=303)
|
|
|
|
|
|
# --------------------------------------------------------------- Galerie ---
|
|
|
|
def _liste_daten(db: Session, request: Request, limit: int) -> dict:
|
|
p = request.query_params
|
|
status = p.get("status", Status.available.value) or None
|
|
kat = p.get("category_id") or None
|
|
|
|
items, gesamt = crud.items_suchen(
|
|
db,
|
|
size=p.get("size") or None,
|
|
category_id=int(kat) if kat and kat.isdigit() else None,
|
|
status=status,
|
|
gender=p.get("gender") or None,
|
|
limit=limit,
|
|
offset=0,
|
|
)
|
|
|
|
# Für "Mehr anzeigen": dieselben Filter, nur mehr Zeilen.
|
|
weiter = dict(p)
|
|
weiter["limit"] = str(limit + SEITE)
|
|
mehr = "&".join(f"{k}={v}" for k, v in weiter.items())
|
|
|
|
return {
|
|
"items": items,
|
|
"gesamt": gesamt,
|
|
"mehr_url": f"/teil/liste?{mehr}",
|
|
"filter": {
|
|
"size": p.get("size", ""),
|
|
"category_id": p.get("category_id", ""),
|
|
"gender": p.get("gender", ""),
|
|
"status": p.get("status", Status.available.value),
|
|
},
|
|
}
|
|
|
|
|
|
def _limit_lesen(request: Request) -> int:
|
|
"""?limit= kommt aus der URL und ist damit Nutzereingabe.
|
|
|
|
Ohne die Prüfung liefert /?limit=abc jedem anonymen Besucher einen
|
|
internen Serverfehler, und limit=-1 heisst in SQLite "alles". Der
|
|
Deckel ist bewusst grosszügiger als die 200 der API: "Mehr anzeigen"
|
|
wächst in 24er-Schritten und soll den ganzen Bestand erreichen können.
|
|
"""
|
|
try:
|
|
wert = int(request.query_params.get("limit", ""))
|
|
except ValueError:
|
|
return SEITE
|
|
return max(1, min(wert, 500))
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
def galerie(request: Request, db: Session = Depends(get_db)):
|
|
limit = _limit_lesen(request)
|
|
return vorlagen.TemplateResponse(
|
|
request, "galerie.html", _umgebung(request, db, kategorien=crud.kategorien(db),
|
|
groessen=crud.groessen(db), **_liste_daten(db, request, limit)),
|
|
)
|
|
|
|
|
|
@router.get("/teil/liste", response_class=HTMLResponse)
|
|
def liste_ausschnitt(request: Request, db: Session = Depends(get_db)):
|
|
"""Nur die Liste - von HTMX beim Filtern und Nachladen geholt."""
|
|
limit = _limit_lesen(request)
|
|
return vorlagen.TemplateResponse(
|
|
request, "_liste.html", _umgebung(request, db, **_liste_daten(db, request, limit))
|
|
)
|
|
|
|
|
|
@router.get("/kleid/{item_id}", response_class=HTMLResponse)
|
|
def detail(item_id: int, request: Request, db: Session = Depends(get_db)):
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None or (item.status == Status.draft.value
|
|
and not security.ist_betreiber(request)):
|
|
raise HTTPException(status_code=404, detail="Nicht gefunden.")
|
|
return vorlagen.TemplateResponse(
|
|
request, "detail.html", _umgebung(request, db, item=item))
|
|
|
|
|
|
# ---------------------------------------------------------- Reservierung ---
|
|
|
|
@router.post("/kleid/{item_id}/reservieren")
|
|
@security.limiter.shared_limit(
|
|
lambda: einstellungen().limit_reservieren, scope="reservieren"
|
|
)
|
|
def reservieren(
|
|
request: Request,
|
|
item_id: int,
|
|
reserved_by: str = Form(...),
|
|
csrf: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
_csrf_oder_fehler(request, csrf)
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None or item.status != Status.available.value:
|
|
return _weiter(f"/kleid/{item_id}", "Das ist leider schon weg.", "fehler")
|
|
|
|
from app.schemas import ReservierenRein
|
|
|
|
try:
|
|
geprueft = ReservierenRein(reserved_by=reserved_by)
|
|
except ValueError:
|
|
return _weiter(f"/kleid/{item_id}", "Bitte nur einen Namen angeben.", "fehler")
|
|
|
|
token = crud.reservieren(db, item, geprueft.reserved_by)
|
|
if token is None:
|
|
# Zwischen Prüfung und Schreiben war jemand schneller.
|
|
return _weiter(f"/kleid/{item_id}", "Das ist leider schon weg.", "fehler")
|
|
freigabe = request.url_for("freigeben_per_link", item_id=item.id)
|
|
return vorlagen.TemplateResponse(
|
|
request, "reserviert.html", _umgebung(request, db, item=item, freigabe_url=f"{freigabe}?token={token}"),
|
|
)
|
|
|
|
|
|
@router.get("/kleid/{item_id}/freigeben", name="freigeben_per_link")
|
|
def freigeben_per_link(
|
|
item_id: int, request: Request, token: str = "", db: Session = Depends(get_db)
|
|
):
|
|
"""Selbst-Freigabe über den Link aus der Reservierung."""
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None or item.status != Status.reserved.value:
|
|
return _weiter(f"/kleid/{item_id}", "Da ist nichts reserviert.", "fehler")
|
|
if not crud.token_gueltig(item, token):
|
|
return _weiter(f"/kleid/{item_id}",
|
|
"Dieser Link passt nicht zur Reservierung.", "fehler")
|
|
crud.freigeben(db, item)
|
|
return _weiter(f"/kleid/{item_id}", "Reservierung aufgehoben.")
|
|
|
|
|
|
@router.post("/kleid/{item_id}/freigeben")
|
|
def freigeben_als_betreiber(
|
|
request: Request, item_id: int, csrf: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
item = crud.item_holen(db, item_id)
|
|
if item and item.status == Status.reserved.value:
|
|
crud.freigeben(db, item)
|
|
return _weiter(f"/kleid/{item_id}", "Reservierung aufgehoben.")
|
|
|
|
|
|
@router.post("/kleid/{item_id}/abgegeben")
|
|
def abgegeben(request: Request, item_id: int, csrf: str = Form(...),
|
|
db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
item = crud.item_holen(db, item_id)
|
|
if item:
|
|
crud.als_abgegeben_markieren(db, item)
|
|
return _weiter(f"/kleid/{item_id}", "Als abgegeben vermerkt.")
|
|
|
|
|
|
# -------------------------------------------------------------- Anmeldung ---
|
|
|
|
@router.get("/anmelden", response_class=HTMLResponse)
|
|
def anmelden_seite(request: Request, db: Session = Depends(get_db)):
|
|
return vorlagen.TemplateResponse(
|
|
request, "anmelden.html", _umgebung(request, db))
|
|
|
|
|
|
@router.post("/anmelden")
|
|
@security.limiter.limit(lambda: einstellungen().limit_anmelden)
|
|
def anmelden(request: Request, passwort: str = Form(...), csrf: str = Form(...),
|
|
db: Session = Depends(get_db)):
|
|
_csrf_oder_fehler(request, csrf)
|
|
hash_wert = security.hinterlegter_hash()
|
|
if hash_wert is None:
|
|
return _weiter("/anmelden", "Der Erfassungsbereich ist nicht eingerichtet.", "fehler")
|
|
if not security.passwort_stimmt(passwort, hash_wert):
|
|
return _weiter("/anmelden", "Passwort stimmt nicht.", "fehler")
|
|
security.anmelden(request)
|
|
return _weiter("/erfassen", "Angemeldet.")
|
|
|
|
|
|
@router.post("/abmelden")
|
|
def abmelden(request: Request, csrf: str = Form(...)):
|
|
_csrf_oder_fehler(request, csrf)
|
|
security.abmelden(request)
|
|
return _weiter("/", "Abgemeldet.")
|
|
|
|
|
|
# --------------------------------------------------------------- Erfassen ---
|
|
|
|
def _zuletzt(request: Request) -> dict:
|
|
"""Werte des zuletzt gespeicherten Teils.
|
|
|
|
Beim Abarbeiten einer Kiste sind Grösse, Kategorie und Saison meist
|
|
gleich - das spart pro Stück drei Felder.
|
|
"""
|
|
return request.session.get("zuletzt", {})
|
|
|
|
|
|
def _zuletzt_merken(request: Request, daten: dict) -> None:
|
|
request.session["zuletzt"] = {
|
|
k: daten.get(k) for k in ("size", "category_id", "gender", "season", "condition")
|
|
}
|
|
|
|
|
|
def _formular_umgebung(request: Request, db: Session, **extra) -> dict:
|
|
return _umgebung(
|
|
request, db,
|
|
kategorien=crud.kategorien(db),
|
|
groessen=crud.groessen(db),
|
|
zuletzt=_zuletzt(request),
|
|
platzhalter_titel="wird sonst automatisch gebildet",
|
|
**extra,
|
|
)
|
|
|
|
|
|
@router.get("/erfassen", response_class=HTMLResponse)
|
|
def erfassen_seite(request: Request, db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
return vorlagen.TemplateResponse(
|
|
request, "erfassen.html", _formular_umgebung(request, db, vorgabe=None)
|
|
)
|
|
|
|
|
|
def _bilder_speichern(dateien: list[UploadFile] | None) -> list[str]:
|
|
cfg = einstellungen()
|
|
namen = []
|
|
for datei in dateien or []:
|
|
if not datei or not datei.filename:
|
|
continue
|
|
rohdaten = datei.file.read(cfg.max_upload_bytes + 1)
|
|
namen.append(images.speichern(rohdaten, cfg.upload_pfad, cfg.max_upload_bytes))
|
|
return namen
|
|
|
|
|
|
@router.post("/erfassen")
|
|
def erfassen(
|
|
request: Request,
|
|
csrf: str = Form(...),
|
|
title: str = Form(""),
|
|
description: str = Form(""),
|
|
size: str = Form(...),
|
|
category_id: str = Form(...),
|
|
gender: str = Form("unisex"),
|
|
season: str = Form("all_year"),
|
|
condition: str = Form("good"),
|
|
dateien: list[UploadFile] = File(default=[]),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
|
|
# Erst die Angaben prüfen, dann die Bilder speichern: scheitert die
|
|
# Prüfung, lägen sonst schon Dateien ohne Eintrag auf der Platte.
|
|
# int() und die Pydantic-Prüfung werfen beide ValueError - ohne das
|
|
# Abfangen würde daraus ein 500er statt einer Fehlermeldung. Die
|
|
# HTML-Formulare verhindern das zwar auf dem normalen Weg, aber ein
|
|
# Handgriff an der Anfrage darf keinen Serverfehler auslösen.
|
|
try:
|
|
daten = ItemAnlegen(
|
|
title=title, description=description, size=size,
|
|
category_id=int(category_id), gender=gender, season=season,
|
|
condition=condition,
|
|
)
|
|
except ValueError:
|
|
return _weiter("/erfassen",
|
|
"Bitte die Angaben prüfen - Kategorie wählen, "
|
|
"Titel höchstens 100 Zeichen.", "fehler")
|
|
|
|
try:
|
|
namen = _bilder_speichern(dateien)
|
|
except images.BildFehler as e:
|
|
return _weiter("/erfassen", str(e), "fehler")
|
|
|
|
item = crud.item_anlegen(db, daten)
|
|
for name in namen:
|
|
crud.bild_anhaengen(db, item, name)
|
|
|
|
_zuletzt_merken(request, {
|
|
"size": size, "category_id": category_id, "gender": gender,
|
|
"season": season, "condition": condition,
|
|
})
|
|
# Zurück ins Formular statt auf die Detailseite: beim Erfassen einer
|
|
# Kiste will man sofort weitermachen.
|
|
return _weiter("/erfassen", f"„{item.anzeige_titel}“ gespeichert.")
|
|
|
|
|
|
@router.post("/stapel")
|
|
def stapel(request: Request, csrf: str = Form(...),
|
|
dateien: list[UploadFile] = File(...), db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
|
|
try:
|
|
namen = _bilder_speichern(dateien)
|
|
except images.BildFehler as e:
|
|
return _weiter("/erfassen", f"Abgebrochen: {e}", "fehler")
|
|
|
|
if not namen:
|
|
return _weiter("/erfassen", "Keine Fotos ausgewählt.", "fehler")
|
|
|
|
entwuerfe = crud.entwuerfe_anlegen(db, len(namen))
|
|
for entwurf, name in zip(entwuerfe, namen):
|
|
crud.bild_anhaengen(db, entwurf, name)
|
|
|
|
return _weiter("/entwuerfe", f"{len(namen)} Fotos hochgeladen.")
|
|
|
|
|
|
@router.get("/entwuerfe", response_class=HTMLResponse)
|
|
def entwuerfe(request: Request, db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
items, gesamt = crud.items_suchen(db, status=Status.draft.value, limit=100)
|
|
return vorlagen.TemplateResponse(
|
|
request, "entwuerfe.html", _umgebung(request, db, items=items, gesamt=gesamt)
|
|
)
|
|
|
|
|
|
@router.get("/kleid/{item_id}/bearbeiten", response_class=HTMLResponse)
|
|
def bearbeiten_seite(item_id: int, request: Request, db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Nicht gefunden.")
|
|
return vorlagen.TemplateResponse(
|
|
request, "bearbeiten.html", _formular_umgebung(request, db, item=item, vorgabe=item)
|
|
)
|
|
|
|
|
|
@router.post("/kleid/{item_id}/bearbeiten")
|
|
def bearbeiten(
|
|
request: Request,
|
|
item_id: int,
|
|
csrf: str = Form(...),
|
|
title: str = Form(""),
|
|
description: str = Form(""),
|
|
size: str = Form(...),
|
|
category_id: str = Form(...),
|
|
gender: str = Form("unisex"),
|
|
season: str = Form("all_year"),
|
|
condition: str = Form("good"),
|
|
dateien: list[UploadFile] = File(default=[]),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Nicht gefunden.")
|
|
|
|
war_entwurf = item.status == Status.draft.value
|
|
# Wie beim Erfassen: erst prüfen, dann Bilder - und ValueError (int,
|
|
# Pydantic) als Meldung statt als 500er.
|
|
try:
|
|
daten = ItemAendern(
|
|
title=title, description=description, size=size,
|
|
category_id=int(category_id), gender=gender, season=season,
|
|
condition=condition,
|
|
# Ein Entwurf wird durchs Nachtragen sichtbar.
|
|
status=Status.available if war_entwurf else None,
|
|
)
|
|
except ValueError:
|
|
return _weiter(f"/kleid/{item_id}/bearbeiten",
|
|
"Bitte die Angaben prüfen - Kategorie wählen, "
|
|
"Titel höchstens 100 Zeichen.", "fehler")
|
|
|
|
try:
|
|
namen = _bilder_speichern(dateien)
|
|
except images.BildFehler as e:
|
|
return _weiter(f"/kleid/{item_id}/bearbeiten", str(e), "fehler")
|
|
|
|
try:
|
|
crud.item_aendern(db, item, daten)
|
|
except IntegrityError:
|
|
db.rollback()
|
|
return _weiter(f"/kleid/{item_id}/bearbeiten",
|
|
"Grösse und Kategorie werden gebraucht.", "fehler")
|
|
|
|
for name in namen:
|
|
crud.bild_anhaengen(db, item, name)
|
|
|
|
if war_entwurf:
|
|
_zuletzt_merken(request, {
|
|
"size": size, "category_id": category_id, "gender": gender,
|
|
"season": season, "condition": condition,
|
|
})
|
|
# Direkt zum nächsten Entwurf - das ist der eigentliche Arbeitsablauf.
|
|
return _weiter("/entwuerfe", "Veröffentlicht.")
|
|
return _weiter(f"/kleid/{item_id}", "Gespeichert.")
|
|
|
|
|
|
@router.post("/kleid/{item_id}/loeschen")
|
|
def loeschen(request: Request, item_id: int, csrf: str = Form(...),
|
|
db: Session = Depends(get_db)):
|
|
security.betreiber_noetig(request)
|
|
_csrf_oder_fehler(request, csrf)
|
|
item = crud.item_holen(db, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Nicht gefunden.")
|
|
for name in crud.item_loeschen(db, item):
|
|
images.loeschen(einstellungen().upload_pfad, name)
|
|
return _weiter("/", "Gelöscht.")
|