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>
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from dataclasses import dataclass
|
|
from typing import Callable, Sequence
|
|
|
|
from ..chat import BACKEND_MATRIX, BACKEND_TELEGRAM, Button, Buttons, callback_data
|
|
from ..constants import CHECKMK_DT_ENABLED, FLAG_ON, MATRIX_ALLOWLIST, TELEGRAM_ALLOWLIST, UI_BASE_URL
|
|
from ..models import KIND_RECOVERY, KIND_SYSTEM, Event, is_host_service
|
|
from .content import alert_link
|
|
|
|
ALERT_BUTTON_MINUTES = 120
|
|
LABEL_MUTE = "2 h stumm"
|
|
LABEL_DEEP = "Tiefenanalyse"
|
|
LABEL_DOWNTIME = "Downtime 2 h"
|
|
LABEL_DETAILS = "Details"
|
|
MATRIX_HINT = "Antwort mit !mute 2h zum Stummschalten"
|
|
MATRIX_LINK_PREFIX = "Details: "
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Extras:
|
|
buttons: Buttons = ()
|
|
hint: str = ""
|
|
|
|
|
|
ExtrasProvider = Callable[[str], Extras | None]
|
|
|
|
|
|
def actionable(events: Sequence[Event]) -> bool:
|
|
return (len(events) == 1 and events[0].kind not in (KIND_RECOVERY, KIND_SYSTEM)
|
|
and not events[0].suppressed)
|
|
|
|
|
|
def telegram_buttons(alert_id: int, downtime: bool) -> Buttons:
|
|
rows = [(Button(LABEL_MUTE, callback_data.encode(callback_data.ACTION_MUTE, alert_id, ALERT_BUTTON_MINUTES)),
|
|
Button(LABEL_DEEP, callback_data.encode(callback_data.ACTION_DEEP, alert_id)))]
|
|
if downtime:
|
|
rows.append((Button(LABEL_DOWNTIME, callback_data.encode(
|
|
callback_data.ACTION_DOWNTIME, alert_id, ALERT_BUTTON_MINUTES)),))
|
|
return tuple(rows)
|
|
|
|
|
|
def with_link(extras: Extras, backend: str, link: str | None) -> Extras:
|
|
if not link:
|
|
return extras
|
|
if backend == BACKEND_TELEGRAM:
|
|
button = Button(LABEL_DETAILS, url=link)
|
|
rows = (*extras.buttons[:-1], (*extras.buttons[-1], button)) if extras.buttons else ((button,),)
|
|
return Extras(buttons=rows, hint=extras.hint)
|
|
if backend == BACKEND_MATRIX:
|
|
line = f"{MATRIX_LINK_PREFIX}{link}"
|
|
return Extras(buttons=extras.buttons, hint=f"{extras.hint}\n{line}" if extras.hint else line)
|
|
return extras
|
|
|
|
|
|
class AlertActions:
|
|
def __init__(self, store):
|
|
self._store = store
|
|
|
|
def build(self, backend: str, events: Sequence[Event], alert_ids: Sequence[int]) -> Extras:
|
|
link = alert_link(self._store.get(UI_BASE_URL), alert_ids[0] if alert_ids else None)
|
|
return with_link(self._actions(backend, events, alert_ids), backend, link)
|
|
|
|
def _actions(self, backend: str, events: Sequence[Event], alert_ids: Sequence[int]) -> Extras:
|
|
if not actionable(events) or not alert_ids:
|
|
return Extras()
|
|
if backend == BACKEND_TELEGRAM and self._store.get(TELEGRAM_ALLOWLIST):
|
|
downtime = (self._store.get(CHECKMK_DT_ENABLED) == FLAG_ON
|
|
and not is_host_service(events[0].service))
|
|
return Extras(buttons=telegram_buttons(alert_ids[0], downtime))
|
|
if backend == BACKEND_MATRIX and self._store.get(MATRIX_ALLOWLIST):
|
|
return Extras(hint=MATRIX_HINT)
|
|
return Extras()
|