patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
107 lines
4.4 KiB
Python
107 lines
4.4 KiB
Python
import re
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from .. import audit
|
|
from ..chat import TELEGRAM_CHAT_ID_RE, TelegramAdapter, mask_secret
|
|
from ..constants import (DEFAULT_TELEGRAM_API_BASE, FLAG_OFF, FLAG_ON, TELEGRAM_ALLOWLIST,
|
|
TELEGRAM_API_BASE, TELEGRAM_CHAT_ID, TELEGRAM_CONTENT_MODE, TELEGRAM_ENABLED,
|
|
TELEGRAM_TOKEN)
|
|
from ..users import User
|
|
from .content_mode import content_mode_field, validate_content_mode
|
|
from .deps import require_admin
|
|
from .sections import URL_RE, Action, Field, Section, check, collect, effective
|
|
|
|
router = APIRouter()
|
|
|
|
SLUG = "telegram"
|
|
MESSAGE_PATH = f"/settings/{SLUG}/message"
|
|
TEST_MESSAGE_TEXT = "Testnachricht von checkbot"
|
|
ERROR_DETAIL_MAX = 300
|
|
|
|
TOKEN_RE = re.compile(r"^\d+:[A-Za-z0-9_-]+$")
|
|
ALLOWLIST_RE = re.compile(r"^\d+(\s*,\s*\d+)*$")
|
|
|
|
|
|
def validate(values: dict) -> list[str]:
|
|
errors = []
|
|
token, chat_id = values.get(TELEGRAM_TOKEN, ""), values.get(TELEGRAM_CHAT_ID, "")
|
|
if values.get(TELEGRAM_ENABLED) == FLAG_ON:
|
|
if not token:
|
|
errors.append("Bot-Token fehlt.")
|
|
if not chat_id:
|
|
errors.append("Standard-Chat-ID fehlt.")
|
|
if token and not TOKEN_RE.match(token):
|
|
errors.append("Bot-Token hat ein ungültiges Format.")
|
|
if chat_id and not TELEGRAM_CHAT_ID_RE.match(chat_id):
|
|
errors.append("Chat-ID muss numerisch sein oder mit @ beginnen.")
|
|
allowlist = values.get(TELEGRAM_ALLOWLIST, "")
|
|
if allowlist and not ALLOWLIST_RE.match(allowlist):
|
|
errors.append("Allowlist darf nur Benutzer-IDs (Ziffern) und Kommas enthalten.")
|
|
api_base = values.get(TELEGRAM_API_BASE, "")
|
|
if api_base and not URL_RE.match(api_base):
|
|
errors.append("API-Basis-URL muss mit http:// oder https:// beginnen.")
|
|
return errors + validate_content_mode(values, TELEGRAM_CONTENT_MODE)
|
|
|
|
|
|
def _adapter(values: dict) -> TelegramAdapter:
|
|
token = values.get(TELEGRAM_TOKEN, "")
|
|
if not token:
|
|
raise ValueError("Bot-Token fehlt.")
|
|
return TelegramAdapter(token, values.get(TELEGRAM_CHAT_ID, ""),
|
|
values.get(TELEGRAM_API_BASE) or DEFAULT_TELEGRAM_API_BASE)
|
|
|
|
|
|
async def test_connection(values: dict) -> str:
|
|
adapter = _adapter(values)
|
|
try:
|
|
return await adapter.check()
|
|
finally:
|
|
await adapter.aclose()
|
|
|
|
|
|
SECTION = Section(
|
|
slug=SLUG,
|
|
title="Telegram",
|
|
fields=(
|
|
Field(TELEGRAM_ENABLED, "Telegram aktivieren", kind="checkbox", default=FLAG_OFF),
|
|
Field(TELEGRAM_TOKEN, "Bot-Token", kind="password", secret=True),
|
|
Field(TELEGRAM_CHAT_ID, "Standard-Chat-ID", placeholder="123456789 oder @kanalname"),
|
|
Field(TELEGRAM_ALLOWLIST, "Erlaubte Benutzer-IDs (kommagetrennt)", placeholder="111111,222222"),
|
|
Field(TELEGRAM_API_BASE, "API-Basis-URL (optional)", placeholder=DEFAULT_TELEGRAM_API_BASE),
|
|
content_mode_field(TELEGRAM_CONTENT_MODE,
|
|
"Die Telegram-Server liegen außerhalb. Mit \"kurz\" verlassen weder Plugin-Output "
|
|
"noch KI-Texte das Haus; die KI-Analyse wird dann nicht nachgereicht."),
|
|
),
|
|
validate=validate,
|
|
test=test_connection,
|
|
optional=True,
|
|
enabled_key=TELEGRAM_ENABLED,
|
|
actions=(Action("Testnachricht senden", MESSAGE_PATH),),
|
|
)
|
|
|
|
|
|
@router.post(MESSAGE_PATH)
|
|
async def send_test_message(request: Request, user: User = Depends(require_admin)):
|
|
store = request.app.state.store
|
|
values = effective(store, collect(await request.form(), SECTION))
|
|
errors = check(SECTION, values)
|
|
if errors:
|
|
return JSONResponse({"ok": False, "detail": " ".join(errors)})
|
|
if not values.get(TELEGRAM_CHAT_ID):
|
|
return JSONResponse({"ok": False, "detail": "Standard-Chat-ID fehlt."})
|
|
try:
|
|
adapter = _adapter(values)
|
|
except ValueError as exc:
|
|
return JSONResponse({"ok": False, "detail": str(exc)})
|
|
try:
|
|
await adapter.send(adapter.default_target, TEST_MESSAGE_TEXT)
|
|
except Exception as exc:
|
|
detail = mask_secret(f"{type(exc).__name__}: {exc}", values[TELEGRAM_TOKEN])
|
|
return JSONResponse({"ok": False, "detail": detail[:ERROR_DETAIL_MAX]})
|
|
finally:
|
|
await adapter.aclose()
|
|
audit.record(request.app.state.db, user.username, "telegram.test_message", values[TELEGRAM_CHAT_ID])
|
|
return JSONResponse({"ok": True, "detail": "Testnachricht gesendet."})
|