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>
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
import re
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from .. import audit
|
|
from ..chat import MATRIX_ROOM_ID_RE, MATRIX_USER_ID_RE, MatrixAdapter, mask_secret
|
|
from ..constants import (FLAG_OFF, FLAG_ON, MATRIX_ALLOWLIST, MATRIX_CONTENT_MODE, MATRIX_ENABLED,
|
|
MATRIX_HOMESERVER, MATRIX_ROOM_ID, MATRIX_TOKEN, MATRIX_USER_ID)
|
|
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 = "matrix"
|
|
MESSAGE_PATH = f"/settings/{SLUG}/message"
|
|
TEST_MESSAGE_TEXT = "Testnachricht von checkbot"
|
|
ERROR_DETAIL_MAX = 300
|
|
|
|
WHITESPACE_RE = re.compile(r"\s")
|
|
|
|
|
|
def validate(values: dict) -> list[str]:
|
|
errors = []
|
|
homeserver, user_id = values.get(MATRIX_HOMESERVER, ""), values.get(MATRIX_USER_ID, "")
|
|
token, room_id = values.get(MATRIX_TOKEN, ""), values.get(MATRIX_ROOM_ID, "")
|
|
if values.get(MATRIX_ENABLED) == FLAG_ON:
|
|
for value, label in ((homeserver, "Homeserver-URL"), (user_id, "Bot-Benutzer-ID"),
|
|
(token, "Access-Token"), (room_id, "Standard-Raum-ID")):
|
|
if not value:
|
|
errors.append(f"{label} fehlt.")
|
|
if homeserver and not URL_RE.match(homeserver):
|
|
errors.append("Homeserver-URL muss mit http:// oder https:// beginnen.")
|
|
if user_id and not MATRIX_USER_ID_RE.match(user_id):
|
|
errors.append("Benutzer-ID muss die Form @name:server haben.")
|
|
if token and WHITESPACE_RE.search(token):
|
|
errors.append("Access-Token darf keine Leerzeichen enthalten.")
|
|
if room_id and not MATRIX_ROOM_ID_RE.match(room_id):
|
|
errors.append("Raum-ID muss mit ! beginnen (nicht der Alias).")
|
|
allowlist = values.get(MATRIX_ALLOWLIST, "")
|
|
if allowlist:
|
|
entries = [e.strip() for e in allowlist.split(",")]
|
|
if not all(MATRIX_USER_ID_RE.match(e) for e in entries):
|
|
errors.append("Allowlist darf nur Matrix-IDs (@name:server), durch Kommas getrennt, enthalten.")
|
|
return errors + validate_content_mode(values, MATRIX_CONTENT_MODE)
|
|
|
|
|
|
def _adapter(values: dict) -> MatrixAdapter:
|
|
for key, label in ((MATRIX_HOMESERVER, "Homeserver-URL"), (MATRIX_USER_ID, "Bot-Benutzer-ID"),
|
|
(MATRIX_TOKEN, "Access-Token")):
|
|
if not values.get(key):
|
|
raise ValueError(f"{label} fehlt.")
|
|
return MatrixAdapter(values[MATRIX_HOMESERVER], values[MATRIX_USER_ID], values[MATRIX_TOKEN],
|
|
values.get(MATRIX_ROOM_ID, ""))
|
|
|
|
|
|
async def test_connection(values: dict) -> str:
|
|
adapter = _adapter(values)
|
|
try:
|
|
return await adapter.check()
|
|
finally:
|
|
await adapter.aclose()
|
|
|
|
|
|
SECTION = Section(
|
|
slug=SLUG,
|
|
title="Matrix",
|
|
fields=(
|
|
Field(MATRIX_ENABLED, "Matrix aktivieren", kind="checkbox", default=FLAG_OFF),
|
|
Field(MATRIX_HOMESERVER, "Homeserver-URL", placeholder="https://matrix.example.de"),
|
|
Field(MATRIX_USER_ID, "Bot-Benutzer-ID", placeholder="@checkbot:example.de"),
|
|
Field(MATRIX_TOKEN, "Access-Token des Bot-Users", kind="password", secret=True),
|
|
Field(MATRIX_ROOM_ID, "Standard-Raum-ID (unverschlüsselt)", placeholder="!abcdef:example.de"),
|
|
Field(MATRIX_ALLOWLIST, "Erlaubte Matrix-IDs (kommagetrennt)", placeholder="@admin:example.de"),
|
|
content_mode_field(MATRIX_CONTENT_MODE,
|
|
"Mit \"kurz\" enthalten Nachrichten keinen Plugin-Output und keine KI-Texte; "
|
|
"die KI-Analyse wird dann nicht nachgereicht."),
|
|
),
|
|
validate=validate,
|
|
test=test_connection,
|
|
optional=True,
|
|
enabled_key=MATRIX_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(MATRIX_ROOM_ID):
|
|
return JSONResponse({"ok": False, "detail": "Standard-Raum-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[MATRIX_TOKEN])
|
|
return JSONResponse({"ok": False, "detail": detail[:ERROR_DETAIL_MAX]})
|
|
finally:
|
|
await adapter.aclose()
|
|
audit.record(request.app.state.db, user.username, "matrix.test_message", values[MATRIX_ROOM_ID])
|
|
return JSONResponse({"ok": True, "detail": "Testnachricht gesendet."})
|