diff --git a/elementary/messages/messaging_integrations/teams_webhook.py b/elementary/messages/messaging_integrations/teams_webhook.py index 533fe82a1..1be4a700c 100644 --- a/elementary/messages/messaging_integrations/teams_webhook.py +++ b/elementary/messages/messaging_integrations/teams_webhook.py @@ -1,7 +1,7 @@ import json from datetime import datetime, timezone from http import HTTPStatus -from typing import Any, Dict, List, Optional +from typing import Any, Optional import requests from ratelimit import limits, sleep_and_retry @@ -26,7 +26,6 @@ Channel: TypeAlias = Optional[str] ONE_SECOND = 1 -TEAMS_PAYLOAD_SIZE_LIMIT = 27 * 1024 class TeamsWebhookHttpError(MessagingIntegrationError): @@ -38,8 +37,16 @@ def __init__(self, response: requests.Response): ) -def _build_payload(card: dict) -> dict: - return { +class TeamsWebhookPayloadTooLargeError(MessagingIntegrationError): + def __init__(self, response: requests.Response, payload_size: int | None = None): + self.status_code = response.status_code + self.response = response + size_info = f" ({payload_size} bytes)" if payload_size is not None else "" + super().__init__(f"Teams webhook payload size{size_info} exceeds limit") + + +def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response: + payload = { "type": "message", "attachments": [ { @@ -50,60 +57,6 @@ def _build_payload(card: dict) -> dict: ], } - -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, @@ -134,6 +87,7 @@ def send_message( body: MessageBody, ) -> MessageSendResult[EmptyMessageContext]: card = format_adaptive_card(body) + payload_json = json.dumps(card) try: response = send_adaptive_card(self.url, card) # For the old teams webhook version of Teams simply returning status code 200 @@ -145,6 +99,11 @@ def send_message( if response.status_code not in (HTTPStatus.OK, HTTPStatus.ACCEPTED) or ( response.status_code == HTTPStatus.OK and len(response.text) > 1 ): + if "HTTP error 413" in response.text: + raise TeamsWebhookPayloadTooLargeError( + response, payload_size=len(payload_json) + ) + raise MessagingIntegrationError( f"Could not post message to Teams via webhook. Status code: {response.status_code}, Error: {response.text}" )