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) <noreply@anthropic.com>
This commit is contained in:
EiSiMo 2026-04-23 09:35:02 +02:00
parent d06dfdaca1
commit 64439fd42e
3 changed files with 63 additions and 8 deletions

View file

@ -18,7 +18,9 @@ logger = logging.getLogger("web.notifications")
EventType = str # 'match' | 'apply_ok' | 'apply_fail' 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: try:
r = requests.post( r = requests.post(
f"https://api.telegram.org/bot{token}/sendMessage", 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}, "disable_web_page_preview": True},
timeout=10, timeout=10,
) )
if not r.ok: 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]) logger.warning("telegram send failed: %s %s", r.status_code, r.text[:200])
return False return False, detail
return True
except requests.RequestException as e: except requests.RequestException as e:
logger.warning("telegram unreachable: %s", e) logger.warning("telegram unreachable: %s", e)
return False return False, str(e)
def _should_notify(notif, event: EventType) -> bool: def _should_notify(notif, event: EventType) -> bool:
@ -57,7 +63,7 @@ def notify_user(user_id: int, event: EventType, *,
token = notif["telegram_bot_token"] token = notif["telegram_bot_token"]
chat = notif["telegram_chat_id"] chat = notif["telegram_chat_id"]
if token and chat: if token and chat:
_telegram_send(token, chat, body_markdown or body_plain) telegram_send(token, chat, body_markdown or body_plain)
except Exception: except Exception:
logger.exception("notify_user failed for user=%s event=%s", user_id, event) logger.exception("notify_user failed for user=%s event=%s", user_id, event)

View file

@ -1,5 +1,7 @@
"""Einstellungen (settings) tab: profile, filter info, notifications, partner, """Einstellungen (settings) tab: profile, filter info, notifications, partner,
account, plus the related action endpoints.""" account, plus the related action endpoints."""
from urllib.parse import quote
from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.responses import HTMLResponse, RedirectResponse
@ -7,6 +9,7 @@ import db
from auth import current_user, hash_password, require_csrf, require_user from auth import current_user, hash_password, require_csrf, require_user
from common import base_context, client_ip, templates from common import base_context, client_ip, templates
from matching import row_to_dict from matching import row_to_dict
from notifications import telegram_send
router = APIRouter() router = APIRouter()
@ -39,6 +42,8 @@ def tab_settings(request: Request, section: str):
ctx["filters"] = row_to_dict(db.get_filters(u["id"])) ctx["filters"] = row_to_dict(db.get_filters(u["id"]))
elif section == "benachrichtigungen": elif section == "benachrichtigungen":
ctx["notifications"] = db.get_notifications(u["id"]) 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": elif section == "partner":
ctx["partner"] = db.get_partner_user(u["id"]) ctx["partner"] = db.get_partner_user(u["id"])
ctx["partner_profile"] = db.get_profile(ctx["partner"]["id"]) if ctx["partner"] else None 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) 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") @router.post("/actions/account/password")
async def action_password( async def action_password(
request: Request, request: Request,

View file

@ -3,6 +3,18 @@
Wähle einen Kanal und entscheide, welche Events dich erreichen sollen. Wähle einen Kanal und entscheide, welche Events dich erreichen sollen.
</p> </p>
{% if notif_flash %}
<div class="chip mb-4
{% if notif_flash == 'test_ok' %}chip-ok
{% else %}chip-bad
{% endif %}">
{% 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 %}
</div>
{% endif %}
<form method="post" action="/actions/notifications" class="space-y-4 max-w-xl" <form method="post" action="/actions/notifications" class="space-y-4 max-w-xl"
autocomplete="off" data-lpignore="true" data-1p-ignore data-bwignore data-form-type="other"> autocomplete="off" data-lpignore="true" data-1p-ignore data-bwignore data-form-type="other">
<input type="hidden" name="csrf" value="{{ csrf }}"> <input type="hidden" name="csrf" value="{{ csrf }}">
@ -42,5 +54,9 @@
</label> </label>
</div> </div>
<div class="flex gap-2">
<button class="btn btn-primary" type="submit">Speichern</button> <button class="btn btn-primary" type="submit">Speichern</button>
<button class="btn btn-ghost" type="submit"
formaction="/actions/notifications/test" formnovalidate>Test senden</button>
</div>
</form> </form>