mirror of
https://forgejo.altau.su/lego/lego-monitoring.git
synced 2026-03-10 12:45:19 +00:00
73 lines
2 KiB
Python
73 lines
2 KiB
Python
from dataclasses import dataclass
|
||
from typing import Optional
|
||
|
||
import nio
|
||
|
||
from alerting.enum import AlertType, Severity
|
||
from misc import cvars
|
||
from misc.config import get_config
|
||
|
||
|
||
@dataclass
|
||
class Alert:
|
||
alert_type: AlertType
|
||
message: str
|
||
severity: Severity
|
||
html_message: Optional[str] = None
|
||
|
||
|
||
async def get_client() -> nio.AsyncClient:
|
||
"""
|
||
Returns a Matrix client.
|
||
It is better to call get_client once and use it for multiple send_alert calls
|
||
"""
|
||
matrix_cfg = cvars.config.get().matrix
|
||
client = nio.AsyncClient(matrix_cfg.homeserver)
|
||
client.access_token = matrix_cfg.access_token
|
||
client.user_id = matrix_cfg.user_id
|
||
client.device_id = matrix_cfg.device_id
|
||
return client
|
||
|
||
|
||
def format_message(alert: Alert) -> str:
|
||
match alert.severity:
|
||
case Severity.INFO:
|
||
severity_emoji = "ℹ️"
|
||
case Severity.WARNING:
|
||
severity_emoji = "⚠️"
|
||
case Severity.CRITICAL:
|
||
severity_emoji = "🆘"
|
||
message = f"{severity_emoji} {alert.alert_type} Alert\n{alert.message}"
|
||
if alert.html_message:
|
||
html_message = f"{severity_emoji} {alert.alert_type} Alert<br>{alert.html_message}"
|
||
return message, html_message
|
||
else:
|
||
return message, None
|
||
|
||
|
||
async def send_alert(alert: Alert) -> None:
|
||
try:
|
||
client = cvars.matrix_client.get()
|
||
except LookupError: # being called standalone
|
||
cvars.config.set(get_config())
|
||
temp_client = True
|
||
client = await get_client()
|
||
cvars.matrix_client.set(client)
|
||
else:
|
||
temp_client = False
|
||
room_id = cvars.config.get().matrix.room_id
|
||
message, html_message = format_message(alert)
|
||
content = {
|
||
"msgtype": "m.text",
|
||
"body": message,
|
||
}
|
||
if html_message:
|
||
content["format"] = "org.matrix.custom.html"
|
||
content["formatted_body"] = html_message
|
||
await client.room_send(
|
||
room_id=room_id,
|
||
message_type="m.room.message",
|
||
content=content,
|
||
)
|
||
if temp_client:
|
||
await client.close()
|