"""Wichtel-Werkstatt – FastAPI-Anwendung. Routen: / Raum erstellen oder per Code beitreten /r/ Teilnehmeransicht (Registrierung / Status / Ergebnis) /r//admin Admin-Login bzw. Admin-Panel /r//recover/ Einmal-Link, um ein verlorenes Cookie zu ersetzen """ from __future__ import annotations import sqlite3 from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI, Form, Request from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from . import db from .draw import draw_assignment BASE_DIR = Path(__file__).resolve().parent COOKIE_MAX_AGE = 2 * 365 * 24 * 60 * 60 # 2 Jahre MIN_PASSWORD_LENGTH = 4 MIN_NAME_LENGTH = 1 MAX_NAME_LENGTH = 50 MAX_ROOM_NAME_LENGTH = 60 MIN_PARTICIPANTS_FOR_DRAW = 3 @asynccontextmanager async def lifespan(_: FastAPI): db.init_db() yield app = FastAPI(title="Wichtel-Werkstatt", lifespan=lifespan) app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static") templates = Jinja2Templates(directory=BASE_DIR / "templates") # ---------- Helpers ---------- def conn() -> sqlite3.Connection: return db.connect() def is_secure(request: Request) -> bool: """Secure-Cookie nur über HTTPS (direkt oder via X-Forwarded-Proto).""" forwarded = request.headers.get("x-forwarded-proto", "") return "https" in forwarded or request.url.scheme == "https" def set_cookie(response, request: Request, name: str, value: str, path: str) -> None: response.set_cookie( name, value, max_age=COOKIE_MAX_AGE, httponly=True, secure=is_secure(request), samesite="lax", path=path, ) def participant_cookie(room_id: str) -> str: return f"wichtel_p_{room_id}" def admin_cookie(room_id: str) -> str: return f"wichtel_a_{room_id}" def base_url(request: Request) -> str: return str(request.base_url).rstrip("/") def share_url(request: Request, code: str) -> str: return f"{base_url(request)}/r/{code}" def render_message(request: Request, title: str, text: str, status: int = 200) -> HTMLResponse: return templates.TemplateResponse( request, "message.html", {"title": title, "text": text}, status_code=status ) def room_not_found(request: Request) -> HTMLResponse: return render_message( request, "Raum nicht gefunden 🤔", "Diesen Wichtel-Raum gibt es nicht (mehr). Prüfe den Link oder Code – " "oder frage den Admin nach dem richtigen Einladungslink.", status=404, ) def get_admin_room(request: Request, room) -> bool: """True, wenn das Admin-Cookie gültig ist.""" token = request.cookies.get(admin_cookie(room["id"])) if not token or not room["admin_token_hash"]: return False return db.hash_token(token) == room["admin_token_hash"] def admin_panel_response(request: Request, room, recovery_link: str | None = None, recovery_for: str | None = None, message: str | None = None, message_kind: str = "ok"): with conn() as c: participants = db.list_participants(c, room["id"]) exclusions = db.list_exclusions(c, room["id"]) admin_participant = None ptoken = request.cookies.get(participant_cookie(room["id"])) if ptoken: admin_participant = db.get_participant_by_token(c, room["id"], ptoken) drawn = db.room_is_drawn(room) can_draw = (not drawn) and len(participants) >= MIN_PARTICIPANTS_FOR_DRAW return templates.TemplateResponse( request, "admin.html", { "room": room, "participants": participants, "exclusions": exclusions, "share_link": share_url(request, room["code"]), "drawn": drawn, "can_draw": can_draw, "min_draw": MIN_PARTICIPANTS_FOR_DRAW, "recovery_link": recovery_link, "recovery_for": recovery_for, "admin_participant": admin_participant, "message": message, "message_kind": message_kind, "base": base_url(request), }, ) # ---------- Startseite ---------- HOME_ERRORS = { "room_name": "Bitte gib einen Raumnamen an (höchstens 60 Zeichen).", "password": "Das Admin-Passwort muss mindestens 4 Zeichen lang sein.", "code": "Zu diesem Code wurde kein Raum gefunden.", } @app.get("/", response_class=HTMLResponse) def home(request: Request, error: str | None = None): return templates.TemplateResponse( request, "home.html", {"error": HOME_ERRORS.get(error or "")} ) @app.post("/rooms") def create_room(request: Request, room_name: str = Form(...), password: str = Form(...)): room_name = room_name.strip() if not room_name or len(room_name) > MAX_ROOM_NAME_LENGTH: return RedirectResponse("/?error=room_name", status_code=303) if len(password) < MIN_PASSWORD_LENGTH: return RedirectResponse("/?error=password", status_code=303) with conn() as c: room = db.create_room(c, room_name, password) token = db.new_token() db.set_admin_token(c, room["id"], token) response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303) set_cookie(response, request, admin_cookie(room["id"]), token, f"/r/{room['code']}") return response @app.post("/join") def join(request: Request, code: str = Form(...)): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return RedirectResponse("/?error=code", status_code=303) return RedirectResponse(f"/r/{room['code']}", status_code=303) # ---------- Teilnehmeransicht ---------- ROOM_ERRORS = { "name": "Bitte gib einen Namen an (höchstens 50 Zeichen).", "duplicate": "Dieser Name ist im Raum schon vergeben – wähle bitte einen anderen " "(z. B. mit Nachnamen).", } @app.get("/r/{code}", response_class=HTMLResponse) def room_page(request: Request, code: str, error: str | None = None): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) participant = None token = request.cookies.get(participant_cookie(room["id"])) if token: participant = db.get_participant_by_token(c, room["id"], token) assignment = None if participant and db.room_is_drawn(room): assignment = db.get_assignment_for(c, room["id"], participant["id"]) return templates.TemplateResponse( request, "room.html", { "room": room, "participant": participant, "assignment": assignment, "drawn": db.room_is_drawn(room), "error": ROOM_ERRORS.get(error or ""), }, ) @app.post("/r/{code}/register") def register(request: Request, code: str, name: str = Form(...)): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if db.room_is_drawn(room): return render_message( request, "Zu spät 🎄", "In diesem Raum wurde bereits ausgelost – eine Teilnahme ist nicht mehr möglich.", ) name = name.strip() if len(name) < MIN_NAME_LENGTH or len(name) > MAX_NAME_LENGTH: return RedirectResponse(f"/r/{room['code']}?error=name", status_code=303) token = db.new_token() try: participant = db.add_participant(c, room["id"], name, token) except sqlite3.IntegrityError: return RedirectResponse(f"/r/{room['code']}?error=duplicate", status_code=303) response = RedirectResponse(f"/r/{room['code']}", status_code=303) set_cookie(response, request, participant_cookie(room["id"]), token, f"/r/{room['code']}") return response # ---------- Admin ---------- @app.get("/r/{code}/admin", response_class=HTMLResponse) def admin_page(request: Request, code: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return templates.TemplateResponse( request, "admin_login.html", {"room": room, "error": None} ) return admin_panel_response(request, room) @app.post("/r/{code}/admin/login") def admin_login(request: Request, code: str, password: str = Form(...)): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not db.verify_password(password, room["password_hash"]): return templates.TemplateResponse( request, "admin_login.html", {"room": room, "error": True} ) token = db.new_token() db.set_admin_token(c, room["id"], token) response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303) set_cookie(response, request, admin_cookie(room["id"]), token, f"/r/{room['code']}") return response @app.post("/r/{code}/admin/logout") def admin_logout(request: Request, code: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) db.clear_admin_token(c, room["id"]) response = RedirectResponse(f"/r/{room['code']}/admin", status_code=303) response.delete_cookie(admin_cookie(room["id"]), path=f"/r/{room['code']}") return response @app.post("/r/{code}/admin/self-register") def admin_self_register(request: Request, code: str, name: str = Form(...)): """Der Admin nimmt selbst am Wichteln teil: legt ihn als Teilnehmer an und setzt zusätzlich den Teilnehmer-Cookie in seinem Browser.""" with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) if db.room_is_drawn(room): return admin_panel_response( request, room, message="Nach der Auslosung ist keine Teilnahme mehr möglich.", message_kind="error", ) name = name.strip() if len(name) < MIN_NAME_LENGTH or len(name) > MAX_NAME_LENGTH: return admin_panel_response( request, room, message="Bitte gib einen gültigen Namen an (höchstens 50 Zeichen).", message_kind="error", ) token = db.new_token() try: db.add_participant(c, room["id"], name, token) except sqlite3.IntegrityError: return admin_panel_response( request, room, message="Dieser Name ist bereits vergeben.", message_kind="error", ) response = admin_panel_response( request, room, message=f"Schön, dass du dabei bist: Du wichtelst als »{name}« mit! 🎅 " "Deine Teilnehmeransicht (und später dein Ergebnis) erreichst du " "über den Einladungslink.", message_kind="ok", ) set_cookie(response, request, participant_cookie(room["id"]), token, f"/r/{room['code']}") return response @app.post("/r/{code}/participants/{pid}/delete") def delete_participant(request: Request, code: str, pid: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) if db.room_is_drawn(room): return admin_panel_response( request, room, message="Nach der Auslosung können keine Teilnehmer mehr gelöscht werden.", message_kind="error", ) db.delete_participant(c, room["id"], pid) return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) @app.post("/r/{code}/exclusions/add") def add_exclusion(request: Request, code: str, from_id: str = Form(...), to_id: str = Form(...)): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) if db.room_is_drawn(room): return admin_panel_response( request, room, message="Nach der Auslosung können keine Ausschlüsse mehr geändert werden.", message_kind="error", ) if from_id == to_id: return admin_panel_response( request, room, message="Ein Teilnehmer kann sich nicht selbst ausgeschlossen werden.", message_kind="error", ) db.add_exclusion(c, room["id"], from_id, to_id) return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) @app.post("/r/{code}/exclusions/{from_id}/{to_id}/delete") def delete_exclusion(request: Request, code: str, from_id: str, to_id: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) if db.room_is_drawn(room): return admin_panel_response( request, room, message="Nach der Auslosung können keine Ausschlüsse mehr geändert werden.", message_kind="error", ) db.remove_exclusion(c, room["id"], from_id, to_id) return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) @app.post("/r/{code}/draw") def draw(request: Request, code: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) if db.room_is_drawn(room): return admin_panel_response( request, room, message="Es wurde bereits ausgelost.", message_kind="error" ) participants = db.list_participants(c, room["id"]) if len(participants) < MIN_PARTICIPANTS_FOR_DRAW: return admin_panel_response( request, room, message=f"Mindestens {MIN_PARTICIPANTS_FOR_DRAW} Teilnehmer nötig – aktuell sind es {len(participants)}.", message_kind="error", ) ids = [p["id"] for p in participants] mapping = draw_assignment(ids, db.exclusion_pairs(c, room["id"])) if mapping is None: return admin_panel_response( request, room, message="Mit diesen Ausschlüssen ist keine gültige Auslosung möglich. " "Bitte entferne oder lockere Ausschlüsse.", message_kind="error", ) db.save_assignments(c, room["id"], mapping) room = db.get_room_by_id(c, room["id"]) return admin_panel_response( request, room, message="🎉 Die Auslosung ist abgeschlossen! Die Teilnehmer sehen ihr Ergebnis, " "wenn sie ihren Einladungslink öffnen.", message_kind="ok", ) # ---------- Recovery-Links (Cookie verloren) ---------- @app.post("/r/{code}/recovery/{pid}") def create_recovery_link(request: Request, code: str, pid: str): """Erzeugt einen Einmal-Link, mit dem der Teilnehmer sein Cookie zurückbekommt. Bewusst auch nach der Auslosung erlaubt: Der Admin sieht dabei weiterhin nicht, wer wen gezogen hat – der Link stellt nur die Sicht des Teilnehmers selbst wieder her. """ with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) if not get_admin_room(request, room): return RedirectResponse(f"/r/{room['code']}/admin", status_code=303) participant = db.get_participant(c, pid) if participant is None or participant["room_id"] != room["id"]: return admin_panel_response( request, room, message="Teilnehmer nicht gefunden.", message_kind="error" ) token = db.new_token() db.create_recovery_link(c, room["id"], pid, token) link = f"{share_url(request, room['code'])}/recover/{token}" return admin_panel_response( request, room, recovery_link=link, recovery_for=participant["name"], message=None, ) @app.get("/r/{code}/recover/{token}", response_class=HTMLResponse) def redeem_recovery_link(request: Request, code: str, token: str): with conn() as c: room = db.get_room_by_code(c, code) if room is None: return room_not_found(request) link = db.get_recovery_link(c, token) if link is None or link["room_id"] != room["id"]: return render_message( request, "Link ungültig 🔒", "Dieser Wiederherstellungslink ist ungültig. Bitte den Admin um einen neuen Link.", status=404, ) if link["used_at"] is not None: return render_message( request, "Link bereits verwendet 🔒", "Dieser Wiederherstellungslink wurde schon benutzt. " "Falls du erneut Zugriff brauchst, bitte den Admin um einen neuen Link.", ) new_token = db.new_token() db.set_participant_token(c, link["participant_id"], new_token) db.mark_recovery_link_used(c, link["token_hash"]) response = RedirectResponse(f"/r/{room['code']}", status_code=303) set_cookie(response, request, participant_cookie(room["id"]), new_token, f"/r/{room['code']}") return response @app.exception_handler(sqlite3.Error) def sqlite_error(request: Request, exc: sqlite3.Error): return render_message( request, "Ups ❄", "Es ist ein Fehler beim Speichern aufgetreten. Bitte versuche es noch einmal.", status=500, )