import random from fastapi import FastAPI, Request, Form, HTTPException from fastapi.responses import RedirectResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import JSONResponse from starlette.middleware.sessions import SessionMiddleware import uvicorn from sqlalchemy import text from db.models import engine from db.models import get_postpaid_user from db.models import get_postpaid_user_by_username from db.models import set_postpaid_user_money from db.models import drink_postpaid_user from db.models import toggle_activate_postpaid_user from db.models import get_prepaid_user from db.models import get_prepaid_user_by_username from db.models import create_prepaid_user from db.models import drink_prepaid_user from db.models import toggle_activate_prepaid_user from db.models import set_prepaid_user_money from db.models import del_user_prepaid from db.models import get_last_drink from db.models import revert_last_drink from db.models import update_drink_type from auth import oidc import os from dotenv import load_dotenv ADMIN_GROUP = "Getraenkeliste Verantwortliche" FS_GROUP = "Getraenkeliste Postpaid" app = FastAPI() load_dotenv() SECRET_KEY = os.getenv("SECRET_KEY", "my_secret_key") app.add_middleware(SessionMiddleware, secret_key=str(SECRET_KEY)) app.include_router(oidc.router) app.mount("/static", StaticFiles(directory="static"), name="static") templates = Jinja2Templates(directory="templates") @app.get("/", response_class=HTMLResponse) def home(request: Request): # Check if user is logged in and has a valid session user_db_id = request.session.get("user_db_id") user_authentik = request.session.get("user_authentik") if not user_db_id or not user_authentik: return RedirectResponse(url="/login", status_code=303) user_db_id = request.session.get("user_db_id") user_authentik = request.session.get("user_authentik") if not user_db_id or not user_authentik: raise HTTPException(status_code=404, detail="User not found") # if user is Admin, load all postpaid users users = None db_users_prepaid = None if ADMIN_GROUP in user_authentik["groups"]: with engine.connect() as conn: t = text("SELECT id FROM users_postpaid") result = conn.execute(t).fetchall() if result: users = [] for row in result: user_db = get_postpaid_user(row[0]) if user_db: users.append(user_db) # if user is in Fachschaft, load all prepaid users prepaid_users_from_curr_user = [] if FS_GROUP in user_authentik["groups"]: with engine.connect() as conn: t = text("SELECT id FROM users_prepaid") result = conn.execute(t).fetchall() if result: db_users_prepaid = [] for row in result: prepaid_user = get_prepaid_user(row[0]) if prepaid_user: db_users_prepaid.append(prepaid_user) # additionally load all prepaid users from the current user t = text("SELECT id, username, user_key, money, last_drink FROM users_prepaid WHERE postpaid_user_id = :user_db_id") result = conn.execute(t, {"user_db_id": user_db_id}).fetchall() if result: prepaid_users_from_curr_user = [] for row in result: prepaid_user = get_prepaid_user(row[0]) if prepaid_user: prepaid_users_from_curr_user.append(prepaid_user) # load current user from database user_is_postpaid = get_is_postpaid(user_authentik) if user_is_postpaid: db_user = get_postpaid_user(user_db_id) else: db_user = get_prepaid_user(user_db_id) # get last drink for current user, if not less than 60 seconds ago last_drink = get_last_drink(user_db_id, user_is_postpaid, 60) return templates.TemplateResponse("index.html", { "request": request, "user": user_authentik, "users": users, "user_db_id": user_db_id, "db_user": db_user, "db_users_prepaid": db_users_prepaid, "prepaid_users_from_curr_user": prepaid_users_from_curr_user, "last_drink": last_drink, "avail_drink_types": ["Paulaner Spezi", "Mio Mate", "Club Mate", "Sonstiges"], }) @app.get("/login", response_class=HTMLResponse) def login_form(request: Request): """ Renders the login form template. Args: request (Request): The incoming HTTP request object. Returns: TemplateResponse: The rendered login.html template with the request context. """ return templates.TemplateResponse("login.html", {"request": request}) @app.post("/set_money_postpaid") def set_money_postpaid(request: Request, username = Form(...), money: float = Form(...)): """ Handles a POST request to set the postpaid money balance for a specified user. Args: request (Request): The incoming HTTP request, containing session information for authentication. username (str, Form): The username of the user whose postpaid balance is to be set (provided via form data). money (float, Form): The new balance amount to set for the user (provided via form data). Raises: HTTPException: - 403 if the current user is not authenticated as an admin. - 404 if the specified user does not exist. Returns: RedirectResponse: Redirects to the home page ("/") with a 303 status code upon successful update. """ user_authentik = request.session.get("user_authentik") if not user_authentik or ADMIN_GROUP not in user_authentik["groups"]: raise HTTPException(status_code=403, detail="Nicht erlaubt") user = get_postpaid_user_by_username(username) requested_user_id = user["id"] set_postpaid_user_money(requested_user_id, money*100) return RedirectResponse(url="/", status_code=303) @app.post("/drink") async def drink(request: Request): """ Handles a drink purchase request for a user. Checks if the user is authenticated and belongs to the admin group. If not, raises a 403 error. Verifies that the user's database ID is present in the session; if not, raises a 404 error. Retrieves the current user's balance and processes the drink purchase. Redirects the user to the homepage after processing. Args: request (Request): The incoming HTTP request containing session data. Raises: HTTPException: If the user is not authenticated as an admin (403). HTTPException: If the user's database ID is not found in the session (404). Returns: RedirectResponse: Redirects to the homepage after processing the drink purchase. """ user_authentik = request.session.get("user_authentik") if not user_authentik or FS_GROUP not in user_authentik["groups"]: raise HTTPException(status_code=403, detail="Not allowed") user_db_id = request.session.get("user_db_id") if not user_db_id: raise HTTPException(status_code=404, detail="User not found") drink_postpaid_user(user_db_id) return RedirectResponse(url="/", status_code=303) @app.post("/payup") def payup(request: Request, username: str = Form(...), money: float = Form(...)): user_auth = request.session.get("user_authentik") if not user_auth or ADMIN_GROUP not in user_auth["groups"]: raise HTTPException(status_code=403, detail="Not allowed") user_db_id = get_postpaid_user_by_username(username)["id"] if not user_db_id: raise HTTPException(status_code=404, detail="User not found") if money < 0 or money > 1000: raise HTTPException(status_code=400, detail="Money must be between 0 and 1000") curr_user_money = get_postpaid_user(user_db_id)["money"] set_postpaid_user_money(user_db_id, curr_user_money + money*100) current_user_db_id = request.session.get("user_db_id") if not current_user_db_id: raise HTTPException(status_code=404, detail="Current user not found") current_user_money = get_postpaid_user(current_user_db_id)["money"] set_postpaid_user_money(current_user_db_id, current_user_money - money*100) return RedirectResponse(url="/", status_code=303) @app.post("/toggle_activated_user_postpaid") def toggle_activated_user_postpaid(request: Request, username: str = Form(...)): user_auth = request.session.get("user_authentik") if not user_auth or ADMIN_GROUP not in user_auth["groups"]: raise HTTPException(status_code=403, detail="Not allowed") user_db_id = get_postpaid_user_by_username(username)["id"] if not user_db_id: raise HTTPException(status_code=404, detail="User not found") toggle_activate_postpaid_user(user_db_id) return RedirectResponse(url="/", status_code=303) @app.post("/add_prepaid_user") def add_prepaid_user(request: Request, username: str = Form(...), start_money: float = Form(...)): active_user_auth = request.session.get("user_authentik") active_user_db_id = request.session.get("user_db_id") if not active_user_auth or FS_GROUP not in active_user_auth["groups"]: raise HTTPException(status_code=403, detail="Not allowed") if not active_user_db_id: raise HTTPException(status_code=404, detail="Current user not found") if not username: raise HTTPException(status_code=400, detail="Username is empty") user_exists = False try: get_postpaid_user_by_username(username) user_exists = True get_prepaid_user_by_username(username) user_exists = True except HTTPException: pass if user_exists: raise HTTPException(status_code=400, detail="User already exists") if start_money < 0 or start_money > 100: raise HTTPException(status_code=400, detail="Start money must be between 0 and 100") if len(username) < 3 or len(username) > 20: raise HTTPException(status_code=400, detail="Username must be between 3 and 20 characters") create_prepaid_user(username, active_user_db_id, int(start_money*100)) prev_money = get_postpaid_user(active_user_db_id)["money"] set_postpaid_user_money(active_user_db_id, prev_money - int(start_money*100)) return RedirectResponse(url="/", status_code=303) @app.post("/drink_prepaid") def drink_prepaid(request: Request): user_db_id = request.session.get("user_db_id") if not user_db_id: raise HTTPException(status_code=404, detail="User not found") user_authentik = request.session.get("user_authentik") if not user_authentik: raise HTTPException(status_code=404, detail="User not found") if not user_authentik["prepaid"]: raise HTTPException(status_code=403, detail="Not allowed") drink_prepaid_user(user_db_id) return RedirectResponse(url="/", status_code=303) @app.post("/toggle_activated_user_prepaid") def toggle_activated_user_prepaid(request: Request, username: str = Form(...)): user_auth = request.session.get("user_authentik") if not user_auth or ADMIN_GROUP not in user_auth["groups"]: raise HTTPException(status_code=403, detail="Not allowed") user_db_id = get_prepaid_user_by_username(username)["id"] if not user_db_id: raise HTTPException(status_code=404, detail="User not found") toggle_activate_prepaid_user(user_db_id) return RedirectResponse(url="/", status_code=303) @app.post("/add_money_prepaid_user") def add_money_prepaid_user(request: Request, username: str = Form(...), money: float = Form(...)): curr_user_auth = request.session.get("user_authentik") if not curr_user_auth or FS_GROUP not in curr_user_auth["groups"]: raise HTTPException(status_code=403, detail="Not allowed") curr_user_db_id = request.session.get("user_db_id") if not curr_user_db_id: raise HTTPException(status_code=404, detail="Logged In User not found") prepaid_user_dict = get_prepaid_user_by_username(username) prepaid_user_db_id = prepaid_user_dict["id"] if not prepaid_user_db_id: raise HTTPException(status_code=404, detail="Prepaid User not found") if money < 0 or money > 100: raise HTTPException(status_code=400, detail="Money must be between 0 and 100") curr_user_money = get_postpaid_user(curr_user_db_id)["money"] prepaid_user_money = prepaid_user_dict["money"] set_postpaid_user_money(curr_user_db_id, curr_user_money - money*100) set_prepaid_user_money(prepaid_user_db_id, prepaid_user_money + money*100, curr_user_db_id) return RedirectResponse(url="/", status_code=303) @app.post("/del_prepaid_user") def delete_prepaid_user(request: Request, username: str = Form(...)): # check if user is in ADMIN_GROUP user_auth = request.session.get("user_authentik") if not user_auth or ADMIN_GROUP not in user_auth["groups"]: raise HTTPException(status_code=403, detail="Nicht erlaubt") user_to_del = get_prepaid_user_by_username(username) if not user_to_del["id"]: raise HTTPException(status_code=404, detail="User not found") del_user_prepaid(user_to_del["id"]) return RedirectResponse(url="/", status_code=303) @app.post("/del_last_drink") def del_last_drink(request: Request): user_db_id = request.session.get("user_db_id") if not user_db_id: raise HTTPException(status_code=404, detail="User not found") user_authentik = request.session.get("user_authentik") if not user_authentik: raise HTTPException(status_code=404, detail="User not found") last_drink = get_last_drink(user_db_id, True, 60) if not last_drink: return RedirectResponse(url="/", status_code=303) user_is_postpaid = get_is_postpaid(user_authentik) revert_last_drink(user_db_id, user_is_postpaid, last_drink["id"]) return RedirectResponse(url="/", status_code=303) @app.post("/update_drink_post") def update_drink_post(request: Request, drink_type: str = Form(...)): user_db_id = request.session.get("user_db_id") if not user_db_id: raise HTTPException(status_code=404, detail="User not found") user_authentik = request.session.get("user_authentik") if not user_authentik: raise HTTPException(status_code=404, detail="User not found") last_drink = get_last_drink(user_db_id, True, 60) if not last_drink: return RedirectResponse(url="/", status_code=303) if not drink_type: raise HTTPException(status_code=400, detail="Drink type is empty") update_drink_type(user_db_id, get_is_postpaid(user_authentik), last_drink["id"], drink_type) return RedirectResponse(url="/", status_code=303) def get_is_postpaid(user_authentik) -> bool: try: if user_authentik["prepaid"]: user_is_postpaid = False else: user_is_postpaid = True except KeyError: user_is_postpaid = True return user_is_postpaid if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)