diff --git a/elementary/messages/messaging_integrations/teams_webhook.py b/elementary/messages/messaging_integrations/teams_webhook.py index 6c4fa7feb..533fe82a1 100644 --- a/elementary/messages/messaging_integrations/teams_webhook.py +++ b/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,