From 64439fd42e3b3895ffaf76f097e8b01f1de57e63 Mon Sep 17 00:00:00 2001 From: EiSiMo Date: Thu, 23 Apr 2026 09:35:02 +0200 Subject: [PATCH] feat(notifications): add Telegram test button New "Test senden" button next to Speichern posts current form credentials (not DB) to /actions/notifications/test, which fires a test message and redirects back with a flash chip showing the outcome (including the Telegram API's error description on failure). telegram_send is now public and returns (ok, detail) so the UI can surface real error messages ("chat not found", "Unauthorized", etc.). Co-Authored-By: Claude Opus 4.7 (1M context) --- web/notifications.py | 20 ++++++++----- web/routes/einstellungen.py | 33 ++++++++++++++++++++++ web/templates/_settings_notifications.html | 18 +++++++++++- 3 files changed, 63 insertions(+), 8 deletions(-) diff --git a/web/notifications.py b/web/notifications.py index 59268ea..baa32da 100644 --- a/web/notifications.py +++ b/web/notifications.py @@ -18,7 +18,9 @@ logger = logging.getLogger("web.notifications") EventType = str # 'match' | 'apply_ok' | 'apply_fail' -def _telegram_send(token: str, chat_id: str, text: str) -> bool: +def telegram_send(token: str, chat_id: str, text: str) -> tuple[bool, str]: + """Send a Telegram message. Returns (ok, detail) — detail is the Telegram + API's `description` on failure, or a short error string, or "ok".""" try: r = requests.post( f"https://api.telegram.org/bot{token}/sendMessage", @@ -26,13 +28,17 @@ def _telegram_send(token: str, chat_id: str, text: str) -> bool: "disable_web_page_preview": True}, timeout=10, ) - if not r.ok: - logger.warning("telegram send failed: %s %s", r.status_code, r.text[:200]) - return False - return True + if r.ok: + return True, "ok" + try: + detail = r.json().get("description") or f"HTTP {r.status_code}" + except ValueError: + detail = f"HTTP {r.status_code}" + logger.warning("telegram send failed: %s %s", r.status_code, r.text[:200]) + return False, detail except requests.RequestException as e: logger.warning("telegram unreachable: %s", e) - return False + return False, str(e) def _should_notify(notif, event: EventType) -> bool: @@ -57,7 +63,7 @@ def notify_user(user_id: int, event: EventType, *, token = notif["telegram_bot_token"] chat = notif["telegram_chat_id"] if token and chat: - _telegram_send(token, chat, body_markdown or body_plain) + telegram_send(token, chat, body_markdown or body_plain) except Exception: logger.exception("notify_user failed for user=%s event=%s", user_id, event) diff --git a/web/routes/einstellungen.py b/web/routes/einstellungen.py index 6b5e8d0..d87d5e7 100644 --- a/web/routes/einstellungen.py +++ b/web/routes/einstellungen.py @@ -1,5 +1,7 @@ """Einstellungen (settings) tab: profile, filter info, notifications, partner, account, plus the related action endpoints.""" +from urllib.parse import quote + from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse @@ -7,6 +9,7 @@ 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 +from notifications import telegram_send router = APIRouter() @@ -39,6 +42,8 @@ def tab_settings(request: Request, section: str): ctx["filters"] = row_to_dict(db.get_filters(u["id"])) elif section == "benachrichtigungen": ctx["notifications"] = db.get_notifications(u["id"]) + ctx["notif_flash"] = request.query_params.get("flash") or "" + ctx["notif_flash_detail"] = request.query_params.get("detail") or "" 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 @@ -105,6 +110,34 @@ async def action_notifications(request: Request, user=Depends(require_user)): return RedirectResponse("/einstellungen/benachrichtigungen", status_code=303) +@router.post("/actions/notifications/test") +async def action_notifications_test(request: Request, user=Depends(require_user)): + """Send a test Telegram message using the credentials currently in the + form (not the DB), so unsaved edits can be verified before saving.""" + form = await request.form() + require_csrf(user["id"], form.get("csrf", "")) + token = (form.get("telegram_bot_token") or "").strip() + chat_id = (form.get("telegram_chat_id") or "").strip() + if not token or not chat_id: + return RedirectResponse( + "/einstellungen/benachrichtigungen?flash=test_missing", + status_code=303, + ) + ok, detail = telegram_send( + token, chat_id, f"*lazyflat* Testnachricht ✅\nfür `{user['username']}`", + ) + db.log_audit( + user["username"], "notifications.tested", + f"ok={ok} detail={detail[:120]}", + user_id=user["id"], ip=client_ip(request), + ) + flash = "test_ok" if ok else "test_fail" + target = f"/einstellungen/benachrichtigungen?flash={flash}" + if not ok: + target += f"&detail={quote(detail[:200])}" + return RedirectResponse(target, status_code=303) + + @router.post("/actions/account/password") async def action_password( request: Request, diff --git a/web/templates/_settings_notifications.html b/web/templates/_settings_notifications.html index 23f952e..b6721be 100644 --- a/web/templates/_settings_notifications.html +++ b/web/templates/_settings_notifications.html @@ -3,6 +3,18 @@ Wähle einen Kanal und entscheide, welche Events dich erreichen sollen.

+{% if notif_flash %} +
+ {% if notif_flash == 'test_ok' %}Testnachricht gesendet. + {% elif notif_flash == 'test_missing' %}Bot-Token und Chat-ID eintragen, um zu testen. + {% elif notif_flash == 'test_fail' %}Test fehlgeschlagen{% if notif_flash_detail %}: {{ notif_flash_detail }}{% endif %}. + {% endif %} +
+{% endif %} +
@@ -42,5 +54,9 @@ - +
+ + +