135 lines
3.9 KiB
Python
135 lines
3.9 KiB
Python
"""Chat: zwei feste Kanäle pro Team (CLAUDE.md).
|
||
|
||
- ankuendigungen: nur Admin postet, alle lesen.
|
||
- team: alle posten.
|
||
|
||
Nur Text. Nachrichten chronologisch. Neue Nachrichten holt der Client per
|
||
kleinem fetch-Polling (JSON-Endpunkt /chat/<art>/seit/<id>) – das ist die einzige
|
||
in v1 erlaubte JavaScript-Ausnahme. Kein WebSocket, kein Framework. Alles
|
||
team-gescoped (team_query) und CSRF-geschützt.
|
||
"""
|
||
|
||
from datetime import timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from flask import (
|
||
Blueprint,
|
||
abort,
|
||
flash,
|
||
jsonify,
|
||
redirect,
|
||
render_template,
|
||
request,
|
||
url_for,
|
||
)
|
||
|
||
from .. import db
|
||
from ..auth import current_user, login_required
|
||
from ..models import Benachrichtigung, Kanal, Nachricht
|
||
from ..notifications import benachrichtigen_team
|
||
from ..team_scope import get_active_team_id, team_query
|
||
|
||
bp = Blueprint("chat", __name__, url_prefix="/chat")
|
||
|
||
LABELS = {Kanal.ART_ANKUENDIGUNGEN: "Ankündigungen", Kanal.ART_TEAM: "Team"}
|
||
MAX_LAENGE = 2000
|
||
_BERLIN = ZoneInfo("Europe/Berlin")
|
||
|
||
|
||
def _zeit(dt):
|
||
"""erstellt_am (in UTC, aus SQLite meist naiv) → 'DD.MM. HH:MM' in Berliner Zeit."""
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(_BERLIN).strftime("%d.%m. %H:%M")
|
||
|
||
|
||
def _kanal_holen(art):
|
||
if art not in (Kanal.ART_ANKUENDIGUNGEN, Kanal.ART_TEAM):
|
||
abort(404)
|
||
kanal = team_query(Kanal).filter_by(art=art).first()
|
||
if kanal is None:
|
||
abort(404)
|
||
return kanal
|
||
|
||
|
||
def _serialisieren(nachricht, user_id):
|
||
return {
|
||
"id": nachricht.id,
|
||
"autor": nachricht.autor.anzeigename,
|
||
"zeit": _zeit(nachricht.erstellt_am),
|
||
"text": nachricht.text,
|
||
"eigene": nachricht.autor_id == user_id,
|
||
}
|
||
|
||
|
||
@bp.route("/<art>")
|
||
@login_required
|
||
def kanal(art):
|
||
kanal = _kanal_holen(art)
|
||
nachrichten = (
|
||
team_query(Nachricht)
|
||
.filter_by(kanal_id=kanal.id)
|
||
.order_by(Nachricht.erstellt_am)
|
||
.all()
|
||
)
|
||
darf_posten = (not kanal.nur_admin_darf_posten) or current_user().ist_admin
|
||
letzte_id = nachrichten[-1].id if nachrichten else 0
|
||
return render_template(
|
||
"chat.html",
|
||
kanal=kanal,
|
||
label=LABELS[kanal.art],
|
||
nachrichten=nachrichten,
|
||
zeit=_zeit,
|
||
darf_posten=darf_posten,
|
||
letzte_id=letzte_id,
|
||
)
|
||
|
||
|
||
@bp.route("/<art>", methods=("POST",))
|
||
@login_required
|
||
def posten(art):
|
||
kanal = _kanal_holen(art)
|
||
# Ankündigungen: nur Admin darf posten (alle dürfen lesen).
|
||
if kanal.nur_admin_darf_posten and not current_user().ist_admin:
|
||
abort(403)
|
||
|
||
text = (request.form.get("text") or "").strip()
|
||
if not text:
|
||
flash("Bitte einen Text eingeben.")
|
||
return redirect(url_for("chat.kanal", art=art))
|
||
if len(text) > MAX_LAENGE:
|
||
flash(f"Nachricht zu lang (max. {MAX_LAENGE} Zeichen).")
|
||
return redirect(url_for("chat.kanal", art=art))
|
||
|
||
db.session.add(
|
||
Nachricht(
|
||
team_id=get_active_team_id(),
|
||
kanal_id=kanal.id,
|
||
autor_id=current_user().id,
|
||
text=text,
|
||
)
|
||
)
|
||
# Alle außer dem Autor über die neue Nachricht informieren.
|
||
benachrichtigen_team(
|
||
f"Neue Nachricht in „{LABELS[kanal.art]}“ von {current_user().anzeigename}",
|
||
url=url_for("chat.kanal", art=art),
|
||
typ=Benachrichtigung.TYP_NACHRICHT,
|
||
ausser_user_ids={current_user().id},
|
||
)
|
||
db.session.commit()
|
||
return redirect(url_for("chat.kanal", art=art))
|
||
|
||
|
||
@bp.route("/<art>/seit/<int:seit>")
|
||
@login_required
|
||
def neu(art, seit):
|
||
"""JSON: alle Nachrichten dieses Kanals mit id > seit (für das Polling)."""
|
||
kanal = _kanal_holen(art)
|
||
user_id = current_user().id
|
||
nachrichten = (
|
||
team_query(Nachricht)
|
||
.filter(Nachricht.kanal_id == kanal.id, Nachricht.id > seit)
|
||
.order_by(Nachricht.erstellt_am)
|
||
.all()
|
||
)
|
||
return jsonify(nachrichten=[_serialisieren(n, user_id) for n in nachrichten])
|