Truncate oversized Teams webhook payloads to prevent HTTP 413 errors

elementary/messages/messaging_integrations/teams_webhook.py

@@ -1,6 +1,7 @@
+import json
 from datetime import datetime, timezone
 from http import HTTPStatus
-from typing import Any, Optional
+from typing import Any, Dict, List, Optional
 
 import requests
 from ratelimit import limits, sleep_and_retry
@@ -25,6 +26,7 @@
 
 Channel: TypeAlias = Optional[str]
 ONE_SECOND = 1
+TEAMS_PAYLOAD_SIZE_LIMIT = 27 * 1024
 
 
 class TeamsWebhookHttpError(MessagingIntegrationError):
@@ -36,8 +38,8 @@ def __init__(self, response: requests.Response):
         )
 
 
-def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
-    payload = {
+def _build_payload(card: dict) -> dict:
+    return {
         "type": "message",
         "attachments": [
             {
@@ -48,6 +50,60 @@ def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
         ],
     }
 
+
+def _truncation_notice_item() -> Dict[str, Any]:
+    return {
+        "type": "TextBlock",
+        "text": "_... Content truncated due to message size limits._",
+        "wrap": True,
+        "isSubtle": True,
+    }
+
+
+def _minimal_card(card: dict) -> dict:
+    return {
+        **card,
+        "body": [
+            {
+                "type": "TextBlock",
+                "text": "Alert content too large to display in Teams.",
+                "wrap": True,
+                "weight": "bolder",
+            }
+        ],
+    }
+
+
+def _truncate_card(card: dict) -> dict:
+    body: List[Dict[str, Any]] = list(card.get("body", []))
+    if not body:
+        return card
+
+    while len(body) > 1:
+        payload = _build_payload({**card, "body": body + [_truncation_notice_item()]})
+        if len(json.dumps(payload)) <= TEAMS_PAYLOAD_SIZE_LIMIT:
+            break
+        body.pop()
+
+    truncated = {**card, "body": body + [_truncation_notice_item()]}
+    if len(json.dumps(_build_payload(truncated))) > TEAMS_PAYLOAD_SIZE_LIMIT:
+        return _minimal_card(card)
+    return truncated
+
+
+def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
+    payload = _build_payload(card)
+    payload_json = json.dumps(payload)
+    if len(payload_json) > TEAMS_PAYLOAD_SIZE_LIMIT:
+        logger.warning(
+            "Teams webhook payload size (%d bytes) exceeds limit (%d bytes), "
+            "truncating card body",
+            len(payload_json),
+            TEAMS_PAYLOAD_SIZE_LIMIT,
+        )
+        card = _truncate_card(card)
+        payload = _build_payload(card)
+
     response = requests.post(
         webhook_url,
         json=payload,