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>
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
from html import escape
|
|
from typing import Mapping, Sequence
|
|
|
|
from ..constants import CONTENT_MODE_FULL
|
|
from ..models import KIND_FLAPPING, KIND_RECOVERY, KIND_REMINDER, Event, display_state, is_host_service
|
|
from .delta import HOST_RECOVERY_HEAD, reminder_text, recovery_text, since_text, state_label
|
|
|
|
TELEGRAM_LIMIT = 4096
|
|
OUTPUT_MAX = 300
|
|
AI_MAX_LINES = 8
|
|
AI_LINE_MAX = 300
|
|
ELLIPSIS = "…"
|
|
AI_UNAVAILABLE = "KI nicht verfügbar"
|
|
THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL | re.IGNORECASE)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Analysis:
|
|
text: str | None = None
|
|
failed: bool = False
|
|
|
|
|
|
def clip(text: str, limit: int) -> str:
|
|
text = text.strip()
|
|
return text if len(text) <= limit else text[:limit - 1].rstrip() + ELLIPSIS
|
|
|
|
|
|
def clip_ai_text(text: str) -> str:
|
|
lines = [line.strip() for line in THINK_RE.sub("", text).splitlines() if line.strip()]
|
|
return "\n".join(clip(line, AI_LINE_MAX) for line in lines[:AI_MAX_LINES])
|
|
|
|
|
|
def _label(event: Event) -> str:
|
|
if event.kind == KIND_FLAPPING:
|
|
return f"FLAPPING ({display_state(event.service, event.state)})"
|
|
return state_label(event)
|
|
|
|
|
|
def _detail(event: Event, full: bool) -> str:
|
|
output = clip(event.output, OUTPUT_MAX) if full else ""
|
|
if event.kind == KIND_RECOVERY:
|
|
return recovery_text(event)
|
|
if event.kind == KIND_REMINDER:
|
|
return f"{reminder_text(event)}: {output}" if full else reminder_text(event)
|
|
return output
|
|
|
|
|
|
def _host_recovery_line(event: Event) -> str:
|
|
return f"<b>{escape(HOST_RECOVERY_HEAD)}</b>{escape(recovery_text(event).removeprefix(HOST_RECOVERY_HEAD))}"
|
|
|
|
|
|
def _event_line(event: Event, full: bool = True) -> str:
|
|
host = is_host_service(event.service)
|
|
if host and event.kind == KIND_RECOVERY:
|
|
return _host_recovery_line(event)
|
|
since = since_text(event) if event.kind not in (KIND_RECOVERY, KIND_REMINDER, KIND_FLAPPING) else ""
|
|
detail = _detail(event, full)
|
|
label = f"<b>{escape(_label(event))}</b>"
|
|
head = label if host else f"{label} {escape(event.service)}"
|
|
head = f"{head}, {escape(since)}" if since else head
|
|
return f"{head}: {escape(detail)}" if (detail or full) else head
|
|
|
|
|
|
def _analysis_block(event: Event, analysis: Analysis, single: bool) -> str:
|
|
title = "Analyse:" if single else f"Analyse {escape(event.service)}:"
|
|
body = AI_UNAVAILABLE if analysis.failed else escape(analysis.text or "")
|
|
return f"\n<b>{title}</b>\n{body}"
|
|
|
|
|
|
def format_message(events: Sequence[Event], analyses: Mapping[int, Analysis] | None = None,
|
|
mode: str = CONTENT_MODE_FULL) -> str:
|
|
full = mode == CONTENT_MODE_FULL
|
|
analyses = analyses if (analyses and full) else {}
|
|
single = len(events) == 1
|
|
blocks = [f"<b>{escape(events[0].host)}</b>"]
|
|
blocks += [_event_line(event, full) for event in events]
|
|
blocks += [_analysis_block(events[i], analyses[i], single) for i in sorted(analyses)
|
|
if i < len(events)]
|
|
return _fit(blocks)
|
|
|
|
|
|
def _fit(blocks: list[str]) -> str:
|
|
text = ""
|
|
for block in blocks:
|
|
candidate = f"{text}\n{block}" if text else block
|
|
if len(candidate) > TELEGRAM_LIMIT - len(ELLIPSIS) - 1:
|
|
return f"{text}\n{ELLIPSIS}" if text else ELLIPSIS
|
|
text = candidate
|
|
return text
|