some actual alerts and telegram client

src/lego_monitoring/alerting/alerts.py

@@ -0,0 +1,73 @@
+from dataclasses import dataclass
+
+from telethon import TelegramClient
+from telethon.sessions import MemorySession
+
+from ..core import cvars
+from .enum import AlertType, Severity
+
+
+@dataclass
+class Alert:
+    alert_type: AlertType
+    message: str
+    severity: Severity
+
+
+async def get_client() -> TelegramClient:
+    config = cvars.config.get()
+    api_id, api_hash, bot_token = config.telegram.creds.split(",")
+    client = await TelegramClient(MemorySession(), api_id, api_hash).start(bot_token=bot_token)
+    client.parse_mode = "html"
+    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}"
+    return message
+
+
+async def send_alert(alert: Alert) -> None:
+    try:
+        client = cvars.tg_client.get()
+    except LookupError:  # being called standalone
+        # cvars.config.set(get_config())
+        # temp_client = True
+        # client = await get_client()
+        # cvars.matrix_client.set(client)
+        raise NotImplementedError  # TODO
+    else:
+        ...  # temp_client = False
+    room_id = cvars.config.get().telegram.room_id
+    message = format_message(alert)
+    await client.send_message(entity=room_id, message=message)
+    # if temp_client:
+    #     await client.close()
+
+
+async def send_start_alert() -> None:
+    config = cvars.config.get()
+    await send_alert(
+        Alert(
+            alert_type=AlertType.BOOT,
+            message=f"Service running with enabled checkers: {', '.join(config.enabled_checker_sets)}",
+            severity=Severity.INFO,
+        )
+    )
+
+
+async def send_stop_alert() -> None:
+    await send_alert(
+        Alert(
+            alert_type=AlertType.BOOT,
+            message="Service stopping.",
+            severity=Severity.INFO,
+        )
+    )

src/lego_monitoring/alerting/enum.py

@@ -0,0 +1,23 @@
+from enum import StrEnum
+
+
+class AlertType(StrEnum):
+    BOOT = "BOOT"
+    TEST = "TEST"
+    # ERROR = "ERROR"
+    # RAM = "RAM"
+    # CPU = "CPU"
+    # TEMP = "TEMP"
+    # VULN = "VULN"
+    # LOGIN = "LOGIN"
+    # SMART = "SMART"  # TODO
+    # RAID = "RAID"
+    # DISKS = "DISKS"
+    # UPS = "UPS"
+    # UPDATE = "UPDATE"
+
+
+class Severity(StrEnum):
+    INFO = "INFO"
+    WARNING = "WARNING"
+    CRITICAL = "CRITICAL"