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
{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()