"""Einstellungen (settings) tab: profile, filter info, notifications, partner, account, plus the related action endpoints.""" from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse import db from auth import current_user, hash_password, require_csrf, require_user from common import base_context, client_ip, templates from matching import row_to_dict router = APIRouter() VALID_SECTIONS = ("profil", "filter", "benachrichtigungen", "partner", "account") @router.get("/einstellungen", response_class=HTMLResponse) def tab_settings_root(request: Request): return RedirectResponse("/einstellungen/profil", status_code=303) @router.get("/einstellungen/{section}", response_class=HTMLResponse) def tab_settings(request: Request, section: str): u = current_user(request) if not u: return RedirectResponse("/login", status_code=303) # Benutzer verwaltung lives under /admin/benutzer since the admin tab rework. if section == "benutzer": return RedirectResponse("/admin/benutzer", status_code=301) if section not in VALID_SECTIONS: raise HTTPException(404) ctx = base_context(request, u, "einstellungen") ctx["section"] = section if section == "profil": ctx["profile"] = db.get_profile(u["id"]) elif section == "filter": ctx["filters"] = row_to_dict(db.get_filters(u["id"])) elif section == "benachrichtigungen": ctx["notifications"] = db.get_notifications(u["id"]) elif section == "partner": ctx["partner"] = db.get_partner_user(u["id"]) ctx["partner_profile"] = db.get_profile(ctx["partner"]["id"]) if ctx["partner"] else None ctx["incoming_requests"] = db.partnership_incoming(u["id"]) ctx["outgoing_requests"] = db.partnership_outgoing(u["id"]) ctx["partner_flash"] = request.query_params.get("flash") or "" return templates.TemplateResponse("einstellungen.html", ctx) @router.post("/actions/profile") async def action_profile(request: Request, user=Depends(require_user)): form = await request.form() require_csrf(user["id"], form.get("csrf", "")) def _b(name): return form.get(name, "").lower() in ("true", "on", "yes", "1") def _i(name): try: return int(form.get(name) or 0) except ValueError: return 0 # Field names are intentionally opaque ("contact_addr", "immomio_login", # "immomio_secret") to keep password managers — specifically Bitwarden — # from recognising the form as a login/identity form and autofilling. db.update_profile(user["id"], { "salutation": form.get("salutation", ""), "firstname": form.get("firstname", ""), "lastname": form.get("lastname", ""), "email": form.get("contact_addr", ""), "telephone": form.get("telephone", ""), "street": form.get("street", ""), "house_number": form.get("house_number", ""), "postcode": form.get("postcode", ""), "city": form.get("city", ""), "is_possessing_wbs": 1 if _b("is_possessing_wbs") else 0, "wbs_type": form.get("wbs_type", "0"), "wbs_valid_till": form.get("wbs_valid_till", "1970-01-01"), "wbs_rooms": _i("wbs_rooms"), "wbs_adults": _i("wbs_adults"), "wbs_children": _i("wbs_children"), "is_prio_wbs": 1 if _b("is_prio_wbs") else 0, "immomio_email": form.get("immomio_login", ""), "immomio_password": form.get("immomio_secret", ""), }) db.log_audit(user["username"], "profile.updated", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/profil", status_code=303) @router.post("/actions/notifications") async def action_notifications(request: Request, user=Depends(require_user)): form = await request.form() require_csrf(user["id"], form.get("csrf", "")) def _b(n): return 1 if form.get(n, "").lower() in ("on", "true", "1", "yes") else 0 channel = form.get("channel", "ui") if channel not in ("ui", "telegram"): channel = "ui" db.update_notifications(user["id"], { "channel": channel, "telegram_bot_token": form.get("telegram_bot_token", ""), "telegram_chat_id": form.get("telegram_chat_id", ""), "notify_on_match": _b("notify_on_match"), "notify_on_apply_success": _b("notify_on_apply_success"), "notify_on_apply_fail": _b("notify_on_apply_fail"), }) db.log_audit(user["username"], "notifications.updated", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/benachrichtigungen", status_code=303) @router.post("/actions/account/password") async def action_password( request: Request, old_password: str = Form(""), new_password: str = Form(""), new_password_repeat: str = Form(""), csrf: str = Form(...), user=Depends(require_user), ): require_csrf(user["id"], csrf) if not new_password or new_password != new_password_repeat: return RedirectResponse("/einstellungen/account?err=mismatch", status_code=303) if len(new_password) < 10: return RedirectResponse("/einstellungen/account?err=tooshort", status_code=303) row = db.get_user_by_username(user["username"]) from auth import verify_hash if not row or not verify_hash(row["password_hash"], old_password): return RedirectResponse("/einstellungen/account?err=wrongold", status_code=303) db.set_user_password(user["id"], hash_password(new_password)) db.log_audit(user["username"], "password.changed", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/account?ok=1", status_code=303) @router.post("/actions/partner/request") async def action_partner_request( request: Request, partner_username: str = Form(...), csrf: str = Form(...), user=Depends(require_user), ): require_csrf(user["id"], csrf) target = db.get_user_by_username((partner_username or "").strip()) if not target or target["id"] == user["id"]: return RedirectResponse("/einstellungen/partner?flash=nouser", status_code=303) req_id = db.partnership_request(user["id"], target["id"]) if req_id is None: return RedirectResponse("/einstellungen/partner?flash=exists", status_code=303) db.log_audit(user["username"], "partner.requested", f"target={target['username']}", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/partner?flash=sent", status_code=303) @router.post("/actions/partner/accept") async def action_partner_accept( request: Request, request_id: int = Form(...), csrf: str = Form(...), user=Depends(require_user), ): require_csrf(user["id"], csrf) if not db.partnership_accept(request_id, user["id"]): return RedirectResponse("/einstellungen/partner?flash=accept_failed", status_code=303) db.log_audit(user["username"], "partner.accepted", f"request={request_id}", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/partner?flash=accepted", status_code=303) @router.post("/actions/partner/decline") async def action_partner_decline( request: Request, request_id: int = Form(...), csrf: str = Form(...), user=Depends(require_user), ): require_csrf(user["id"], csrf) db.partnership_decline(request_id, user["id"]) db.log_audit(user["username"], "partner.declined", f"request={request_id}", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/partner?flash=declined", status_code=303) @router.post("/actions/partner/unlink") async def action_partner_unlink( request: Request, csrf: str = Form(...), user=Depends(require_user), ): require_csrf(user["id"], csrf) db.partnership_unlink(user["id"]) db.log_audit(user["username"], "partner.unlinked", user_id=user["id"], ip=client_ip(request)) return RedirectResponse("/einstellungen/partner?flash=unlinked", status_code=303)