Add Telegram / Discord / Teams / PagerDuty notification sinks

Author: JE-Chen

automation_file/__init__.py

@@ -88,11 +88,15 @@
     zip_info,
 )
 from automation_file.notify import (
+    DiscordSink,
     EmailSink,
     NotificationException,
     NotificationManager,
     NotificationSink,
+    PagerDutySink,
     SlackSink,
+    TeamsSink,
+    TelegramSink,
     WebhookSink,
     notification_manager,
     notify_send,
@@ -348,11 +352,15 @@ def __getattr__(name: str) -> Any:
     "progress_registry",
     "register_progress_ops",
     # Notifications
+    "DiscordSink",
     "EmailSink",
     "NotificationException",
     "NotificationManager",
     "NotificationSink",
+    "PagerDutySink",
     "SlackSink",
+    "TeamsSink",
+    "TelegramSink",
     "WebhookSink",
     "notification_manager",
     "notify_send",

automation_file/notify/__init__.py

@@ -17,18 +17,26 @@
     register_notify_ops,
 )
 from automation_file.notify.sinks import (
+    DiscordSink,
     EmailSink,
     NotificationSink,
+    PagerDutySink,
     SlackSink,
+    TeamsSink,
+    TelegramSink,
     WebhookSink,
 )
 
 __all__ = [
+    "DiscordSink",
     "EmailSink",
     "NotificationException",
     "NotificationManager",
     "NotificationSink",
+    "PagerDutySink",
     "SlackSink",
+    "TeamsSink",
+    "TelegramSink",
     "WebhookSink",
     "notification_manager",
     "notify_send",

automation_file/notify/sinks.py

@@ -194,6 +194,218 @@ def __repr__(self) -> str:
         )
 
 
+class TelegramSink(NotificationSink):
+    """Telegram Bot API sink.
+
+    ``bot_token`` is a secret — it's combined into the request URL and never
+    logged. ``chat_id`` identifies the target channel or user.
+    """
+
+    _LEVEL_PREFIX: ClassVar[dict[str, str]] = {
+        "info": "",
+        "warning": "⚠️ ",
+        "error": "🚨 ",
+    }
+    _API_HOST: ClassVar[str] = "https://api.telegram.org"
+
+    def __init__(
+        self,
+        bot_token: str,
+        chat_id: str | int,
+        *,
+        name: str = "telegram",
+        timeout: float = _DEFAULT_TIMEOUT,
+    ) -> None:
+        if not bot_token:
+            raise NotificationException("telegram sink requires a bot_token")
+        if chat_id in (None, ""):
+            raise NotificationException("telegram sink requires a chat_id")
+        self._bot_token = bot_token
+        self.chat_id = chat_id
+        self.name = name
+        self.timeout = timeout
+        self._url = f"{self._API_HOST}/bot{bot_token}/sendMessage"
+        validate_http_url(self._url)
+
+    def send(self, subject: str, body: str, level: str = "info") -> None:
+        self._check_level(level)
+        prefix = self._LEVEL_PREFIX[level]
+        text = f"{prefix}{subject}\n{self._truncate(body)}"
+        payload = {"chat_id": self.chat_id, "text": text, "disable_web_page_preview": True}
+        try:
+            response = requests.post(
+                self._url,
+                data=json.dumps(payload).encode("utf-8"),
+                headers={"Content-Type": "application/json"},
+                timeout=self.timeout,
+                allow_redirects=False,
+            )
+        except requests.RequestException as err:
+            raise NotificationException(f"telegram sink {self.name!r} post failed: {err}") from err
+        if response.status_code >= 400:
+            raise NotificationException(
+                f"telegram sink {self.name!r} returned HTTP {response.status_code}"
+            )
+
+    def __repr__(self) -> str:
+        return f"TelegramSink(name={self.name!r}, chat_id={self.chat_id!r})"
+
+
+class DiscordSink(NotificationSink):
+    """Discord incoming-webhook sink — POSTs ``{content}`` to the webhook URL."""
+
+    _LEVEL_PREFIX: ClassVar[dict[str, str]] = {
+        "info": "",
+        "warning": ":warning: ",
+        "error": ":rotating_light: ",
+    }
+    # Discord caps message content at 2000 characters.
+    _MAX_CONTENT: ClassVar[int] = 1900
+
+    def __init__(
+        self,
+        webhook_url: str,
+        *,
+        name: str = "discord",
+        timeout: float = _DEFAULT_TIMEOUT,
+    ) -> None:
+        validate_http_url(webhook_url)
+        self._url = webhook_url
+        self.name = name
+        self.timeout = timeout
+
+    def send(self, subject: str, body: str, level: str = "info") -> None:
+        self._check_level(level)
+        prefix = self._LEVEL_PREFIX[level]
+        content = f"{prefix}**{subject}**\n{self._truncate(body)}"
+        if len(content) > self._MAX_CONTENT:
+            content = content[: self._MAX_CONTENT] + "…"
+        try:
+            response = requests.post(
+                self._url,
+                data=json.dumps({"content": content}).encode("utf-8"),
+                headers={"Content-Type": "application/json"},
+                timeout=self.timeout,
+                allow_redirects=False,
+            )
+        except requests.RequestException as err:
+            raise NotificationException(f"discord sink {self.name!r} post failed: {err}") from err
+        if response.status_code >= 400:
+            raise NotificationException(
+                f"discord sink {self.name!r} returned HTTP {response.status_code}"
+            )
+
+
+class TeamsSink(NotificationSink):
+    """Microsoft Teams incoming-webhook sink using the legacy MessageCard schema."""
+
+    _LEVEL_COLOR: ClassVar[dict[str, str]] = {
+        "info": "2E86DE",
+        "warning": "E67E22",
+        "error": "C0392B",
+    }
+
+    def __init__(
+        self,
+        webhook_url: str,
+        *,
+        name: str = "teams",
+        timeout: float = _DEFAULT_TIMEOUT,
+    ) -> None:
+        validate_http_url(webhook_url)
+        self._url = webhook_url
+        self.name = name
+        self.timeout = timeout
+
+    def send(self, subject: str, body: str, level: str = "info") -> None:
+        self._check_level(level)
+        payload = {
+            "@type": "MessageCard",
+            "@context": "https://schema.org/extensions",
+            "summary": subject,
+            "themeColor": self._LEVEL_COLOR[level],
+            "title": subject,
+            "text": self._truncate(body),
+        }
+        try:
+            response = requests.post(
+                self._url,
+                data=json.dumps(payload).encode("utf-8"),
+                headers={"Content-Type": "application/json"},
+                timeout=self.timeout,
+                allow_redirects=False,
+            )
+        except requests.RequestException as err:
+            raise NotificationException(f"teams sink {self.name!r} post failed: {err}") from err
+        if response.status_code >= 400:
+            raise NotificationException(
+                f"teams sink {self.name!r} returned HTTP {response.status_code}"
+            )
+
+
+class PagerDutySink(NotificationSink):
+    """PagerDuty Events API v2 sink.
+
+    ``routing_key`` is the integration key for a PagerDuty service and is
+    treated as a secret. Each ``send`` enqueues a ``trigger`` event unless
+    the level is explicitly ``info``, in which case it sends ``acknowledge``
+    semantics via ``event_action='trigger'`` + ``severity='info'``.
+    """
+
+    _ENQUEUE_URL: ClassVar[str] = "https://events.pagerduty.com/v2/enqueue"
+    _LEVEL_SEVERITY: ClassVar[dict[str, str]] = {
+        "info": "info",
+        "warning": "warning",
+        "error": "error",
+    }
+
+    def __init__(
+        self,
+        routing_key: str,
+        *,
+        source: str = "automation_file",
+        name: str = "pagerduty",
+        timeout: float = _DEFAULT_TIMEOUT,
+    ) -> None:
+        if not routing_key:
+            raise NotificationException("pagerduty sink requires a routing_key")
+        self._routing_key = routing_key
+        self.source = source
+        self.name = name
+        self.timeout = timeout
+        validate_http_url(self._ENQUEUE_URL)
+
+    def send(self, subject: str, body: str, level: str = "info") -> None:
+        self._check_level(level)
+        payload = {
+            "routing_key": self._routing_key,
+            "event_action": "trigger",
+            "payload": {
+                "summary": subject,
+                "source": self.source,
+                "severity": self._LEVEL_SEVERITY[level],
+                "custom_details": {"body": self._truncate(body)},
+            },
+        }
+        try:
+            response = requests.post(
+                self._ENQUEUE_URL,
+                data=json.dumps(payload).encode("utf-8"),
+                headers={"Content-Type": "application/json"},
+                timeout=self.timeout,
+                allow_redirects=False,
+            )
+        except requests.RequestException as err:
+            raise NotificationException(f"pagerduty sink {self.name!r} post failed: {err}") from err
+        if response.status_code >= 400:
+            raise NotificationException(
+                f"pagerduty sink {self.name!r} returned HTTP {response.status_code}"
+            )
+
+    def __repr__(self) -> str:
+        return f"PagerDutySink(name={self.name!r}, source={self.source!r})"
+
+
 def _describe(sink: NotificationSink) -> dict[str, Any]:
     info: dict[str, Any] = {"name": sink.name, "type": type(sink).__name__}
     if isinstance(sink, WebhookSink):
@@ -202,6 +414,10 @@ def _describe(sink: NotificationSink) -> dict[str, Any]:
         info["host"] = sink.host
         info["port"] = sink.port
         info["recipients"] = sink.recipients
+    elif isinstance(sink, TelegramSink):
+        info["chat_id"] = sink.chat_id
+    elif isinstance(sink, PagerDutySink):
+        info["source"] = sink.source
     return info