54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from ...redaction import redact
|
|
from ...timeutil import payload_with_display_times
|
|
from .base import DeliveryResult
|
|
from .http import classify_response
|
|
|
|
|
|
def _format_message(event_type: str, payload: dict, time_zone: str) -> str:
|
|
payload = payload_with_display_times(payload, time_zone)
|
|
icon = "[FIRING]" if event_type == "firing" else "[RESOLVED]"
|
|
sev = payload.get("severity", "unknown")
|
|
agent = payload.get("agent_id", "?")
|
|
check = payload.get("check_id", "?")
|
|
status = payload.get("status", "?")
|
|
incident_key = payload.get("incident_key", "?")
|
|
output = payload.get("output") or ""
|
|
lines = [
|
|
f"{icon} {sev} {agent}/{check}",
|
|
f"status: {status}",
|
|
f"incident_key: {incident_key}",
|
|
f"observed_at: {payload.get('observed_at_display') or payload.get('observed_at', '?')}",
|
|
]
|
|
if output:
|
|
lines.append(f"output: {output[:300]}")
|
|
return redact("\n".join(lines)) or ""
|
|
|
|
|
|
class TelegramNotifier:
|
|
name = "telegram"
|
|
|
|
def __init__(self, client: httpx.AsyncClient, token: str, chat_id: str, time_zone: str) -> None:
|
|
self._client = client
|
|
self._token = token
|
|
self._chat_id = chat_id
|
|
self._time_zone = time_zone
|
|
|
|
async def deliver(self, event_type: str, payload: dict) -> DeliveryResult:
|
|
url = f"https://api.telegram.org/bot{self._token}/sendMessage"
|
|
body = {
|
|
"chat_id": self._chat_id,
|
|
"text": _format_message(event_type, payload, self._time_zone),
|
|
}
|
|
try:
|
|
resp = await self._client.post(url, json=body)
|
|
except httpx.HTTPError as exc:
|
|
return DeliveryResult(ok=False, permanent=False, error=type(exc).__name__)
|
|
ok, perm = classify_response(resp.status_code)
|
|
if ok:
|
|
return DeliveryResult(ok=True)
|
|
return DeliveryResult(ok=False, permanent=perm, error=f"http_{resp.status_code}")
|