Coverage for custom_components/supernotify/transports/telegram.py: 100%
182 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1"""Telegram transport for SuperNotify.
3Sends push notifications via Telegram using Home Assistant's telegram_bot integration.
4Supports text messages, photos (with optional captions), and inline action buttons.
5Uses telegram_bot service for granular control over formatting, media types, and protection.
7Targets are chat ids, as `telegram_chat_id` targets, or the `notify` entity the integration
8creates for each chat, which also identifies the bot when there's more than one.
10Supported data keys (all optional):
11 telegram_parse_mode str "HTML" | "Markdown" | "MarkdownV2" (default: plain text)
12 telegram_disable_notification bool Override silent mode (overrides priority mapping)
13 telegram_protect_content bool Accepted, but has no effect (see below)
14 telegram_chat_id str|int|list Override target chat ids (default: None)
15 telegram_reply_to_message_id int Reply to message ID (default: None)
16 telegram_inline_keyboard list Custom action buttons in [[{text, callback_data}]] format
17 telegram_attach_image bool Attach camera snapshot (default: False)
18 telegram_image_as_document bool Send image as document without compression (default: False)
20Notes on the HA telegram_bot service schema:
21- Chat ids are sent as `chat_id`, not `target`, which HA has deprecated, raising a repair
22 issue for every call using it.
23- The schema rejects unknown keys, and a notification's `data` reaches every delivery, so
24 only the keys in PASSTHROUGH_KEYS are passed on from it.
25- `parse_mode` accepts only lowercase values: 'html', 'markdown',
26 'markdownv2', 'plain_text'. We normalise the user-provided value to
27 lowercase before forwarding.
28- `protect_content` is a Telegram Bot API parameter but the HA
29 `telegram_bot` integration only accepts it for `send_media_group`, not the
30 services used here. We accept the data key for forward-compatibility but
31 do NOT forward it to the service.
32- `inline_keyboard` for HA is a list of rows where each row is a list of
33 `[label, callback_or_url]` 2-element lists (NOT dicts with
34 `text`/`callback_data` keys). Example: `[[["OK","/ok"],["Cancel","/cancel"]]]`.
35"""
37from __future__ import annotations
39import html
40import logging
41from typing import TYPE_CHECKING, Any, ClassVar, cast
43from custom_components.supernotify.common import boolify
44from custom_components.supernotify.const import (
45 ACTION_LEVEL_NOTIFICATION,
46 ATTR_PHONE,
47 ATTR_TELEGRAM_CHAT_ID,
48 CONF_LEVEL,
49 TRANSPORT_TELEGRAM,
50)
51from custom_components.supernotify.model import DebugTrace, SuppressionReason, TransportConfig, TransportFeature
52from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption
53from custom_components.supernotify.target import TargetEntityCategory
54from custom_components.supernotify.transport import Transport
56if TYPE_CHECKING:
57 from custom_components.supernotify.envelope import Envelope
58 from custom_components.supernotify.hass_api import HomeAssistantAPI
59from homeassistant.helpers.typing import ConfigType
61_LOGGER = logging.getLogger(__name__)
63HA_TELEGRAM_BOT_DOMAIN = "telegram_bot"
65_PRIORITY_MAP = {
66 "critical": False, # notify (sound + vibration)
67 "high": False, # notify (sound)
68 "medium": False, # notify (sound)
69 "low": True, # silent
70 "minimum": True, # silent
71}
73# Telegram Bot API limits
74_MAX_MESSAGE_LENGTH = 4096
75_MAX_CAPTION_LENGTH = 1024
77# telegram_bot service keys a notification's own `data` may set, beyond those the transport sets itself
78PASSTHROUGH_KEYS = (
79 "config_entry_id",
80 "disable_web_page_preview",
81 "keyboard",
82 "message_tag",
83 "message_thread_id",
84 "one_time_keyboard",
85 "resize_keyboard",
86)
87# characters with a meaning in each markdown flavour, escaped in a title
88_MARKDOWN_SPECIAL = {"markdown": "_*`[", "markdownv2": "_*[]()~`>#+-=|{}.!\\"}
91def _escape_html(text: str) -> str:
92 """Escape HTML special characters for Telegram HTML parse mode."""
93 return html.escape(text) if text else ""
96def _escape_markdown(text: str, parse_mode: str) -> str:
97 """Escape markdown special characters, which Telegram otherwise fails to parse"""
98 special = _MARKDOWN_SPECIAL[parse_mode]
99 return "".join(f"\\{c}" if c in special else c for c in text)
102def _fit(text: str, limit: int, parse_mode: str) -> tuple[str, str]:
103 """Cut text to Telegram's limit, as plain text if it was formatted, since a cut can leave
104 a tag or markup open, which Telegram rejects"""
105 if len(text) <= limit:
106 return text, parse_mode
107 _LOGGER.debug("SUPERNOTIFY telegram: message truncated to %d chars", limit)
108 return text[:limit], "plain_text"
111def _normalise_inline_keyboard(keyboard: list) -> list:
112 """Normalise a user-supplied inline keyboard to the HA telegram_bot shape.
114 Accepts:
115 - HA-native: list of rows where each row is a list of `[label, callback]`
116 2-element lists. Returned unchanged.
117 - Telegram Bot API native: list of rows where each row is a list of
118 `{"text": label, "callback_data": callback}` dicts. Converted to the
119 HA shape.
120 - Mixed rows are tolerated (each button is normalised independently).
122 Returns the keyboard in the HA shape, or `[]` if the input is malformed.
123 """
124 if not isinstance(keyboard, list):
125 return []
126 out: list = []
127 for row in keyboard:
128 if not isinstance(row, list):
129 continue
130 out_row: list = []
131 for btn in row:
132 if isinstance(btn, list) and len(btn) >= 2:
133 # Already in HA shape: [label, callback_or_url]
134 out_row.append([str(btn[0]), str(btn[1])])
135 elif isinstance(btn, dict):
136 label = btn.get("text") or btn.get("title") or btn.get("label")
137 callback = btn.get("callback_data") or btn.get("action") or btn.get("url")
138 if label and callback:
139 out_row.append([str(label), str(callback)])
140 if out_row:
141 out.append(out_row)
142 return out
145def _build_inline_keyboard(actions: list) -> list:
146 """Convert SuperNotify actions to HA telegram_bot inline keyboard format.
148 SuperNotify actions are dicts with keys "title" (button label)
149 and "action" (callback identifier). The HA telegram_bot service expects
150 a list of rows where each row is a list of [label, callback_or_url]
151 2-element lists (NOT dicts with text/callback_data keys).
152 Example shape: [[["OK", "/ack_ok"], ["Open HA", "/open"]]]
154 Limits to max 5 buttons (single row) and 64-byte callback_data.
155 """
156 if not actions or not isinstance(actions, list):
157 return []
159 row = []
160 for i, action in enumerate(actions):
161 if i >= 5:
162 _LOGGER.warning("SUPERNOTIFY telegram: more than 5 actions, truncating to 5")
163 break
165 if not isinstance(action, dict):
166 _LOGGER.warning("SUPERNOTIFY telegram: action not a dict, skipped")
167 continue
169 if action.get(CONF_LEVEL) == ACTION_LEVEL_NOTIFICATION:
170 # opens on tapping the notification, which Telegram has no way to do, and not a button
171 continue
173 # SuperNotify action keys: "title" for label, "action" for callback id.
174 # Also tolerate Telegram-native "text"/"callback_data" for users who
175 # craft the action list manually.
176 label = action.get("title") or action.get("text") or action.get("label")
177 callback = action.get("action") or action.get("callback_data") or action.get("id")
179 if not label or not callback:
180 _LOGGER.warning("SUPERNOTIFY telegram: action missing 'title' or 'action' key, skipped")
181 continue
183 # Truncate callback_data to 64 bytes UTF-8 per Telegram API
184 callback_str = str(callback)
185 encoded = callback_str.encode("utf-8")
186 if len(encoded) > 64:
187 callback_str = encoded[:64].decode("utf-8", errors="ignore")
188 _LOGGER.warning("SUPERNOTIFY telegram: callback_data truncated to 64 bytes: %s", callback_str)
190 # HA format: [label, callback_or_url] 2-element list
191 row.append([str(label), callback_str])
193 return [row] if row else []
196class TelegramTransport(Transport):
197 """Notify via Telegram using Home Assistant telegram_bot integration."""
199 name = TRANSPORT_TELEGRAM
200 integration_domain = HA_TELEGRAM_BOT_DOMAIN
201 declared_options: ClassVar[list[DeliveryOption]] = [*MEDIA_OPTIONS]
203 def __init__(self, *args: Any, **kwargs: Any) -> None:
204 super().__init__(*args, **kwargs)
206 @property
207 def supported_features(self) -> TransportFeature:
208 return (
209 TransportFeature.MESSAGE
210 | TransportFeature.TITLE
211 | TransportFeature.IMAGES
212 | TransportFeature.ACTIONS
213 | TransportFeature.SNAPSHOT_IMAGE
214 )
216 @property
217 def default_config(self) -> TransportConfig:
218 config = super().default_config
219 config.delivery_defaults.action = "telegram_bot.send_message"
220 return config
222 @property
223 def unique_target_categories(self) -> list[str | TargetEntityCategory]:
224 # a numeric chat ID has no shape distinct enough for automatic matching (see
225 # deliver()'s own numeric-ness check below), so it's only ever reachable here via
226 # explicit qualification - telegram_chat_id is nonetheless a dedicated category name
227 # no other transport uses, so a value explicitly qualified with it is unambiguous.
228 # The notify entity for each chat is the integration's own, claimed ahead of the
229 # generic notify_entity transport, like html5's
230 return [ATTR_TELEGRAM_CHAT_ID, TargetEntityCategory(domain="notify", platform=HA_TELEGRAM_BOT_DOMAIN)]
232 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
233 return hass_api.find_config_entry_data(HA_TELEGRAM_BOT_DOMAIN) is not None
235 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
236 return {self.name: {}}
238 def validate_action(self, action: str | None) -> bool:
239 """Validate that action is one of the supported telegram_bot services."""
240 if not action:
241 return False
242 return action in (
243 "telegram_bot.send_message",
244 "telegram_bot.send_photo",
245 "telegram_bot.send_document",
246 )
248 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
249 _LOGGER.debug("SUPERNOTIFY telegram %s", envelope.message)
251 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
253 # Pop Telegram-specific data keys
254 parse_mode: str | None = raw_data.pop("telegram_parse_mode", None)
255 disable_notification_override = raw_data.pop("telegram_disable_notification", None)
256 # `telegram_protect_content` is accepted but currently NOT forwarded
257 # because the HA telegram_bot service schema rejects it. Pop to keep
258 # it out of the residual raw_data merge below.
259 raw_data.pop("telegram_protect_content", None)
260 chat_id_override = raw_data.pop("telegram_chat_id", None)
261 reply_to_message_id = raw_data.pop("telegram_reply_to_message_id", None)
262 custom_keyboard = raw_data.pop("telegram_inline_keyboard", None)
263 attach_image = boolify(raw_data.pop("telegram_attach_image", False), default=False)
264 image_as_document = boolify(raw_data.pop("telegram_image_as_document", False), default=False)
266 # Chat ids from the data override, else the envelope's own `telegram_chat_id` targets, else
267 # the delivery's own configured target, where `phone` was once used for a chat id
268 raw_chat_ids: list[Any] = []
269 if chat_id_override:
270 raw_chat_ids = chat_id_override if isinstance(chat_id_override, list) else [chat_id_override]
271 elif envelope.target and envelope.target.targets.get(ATTR_TELEGRAM_CHAT_ID):
272 raw_chat_ids = envelope.target.targets[ATTR_TELEGRAM_CHAT_ID]
273 elif envelope.delivery and envelope.delivery.target:
274 delivery_targets = envelope.delivery.target.targets
275 raw_chat_ids = delivery_targets.get(ATTR_TELEGRAM_CHAT_ID) or delivery_targets.get(ATTR_PHONE) or []
277 # telegram_bot expects integers. Group chat IDs are negative; channel usernames
278 # (@channel) would be strings but are not accepted by the telegram_bot schema
279 chat_ids: list[int] = []
280 for raw_chat_id in raw_chat_ids:
281 try:
282 chat_ids.append(int(cast("str | int", raw_chat_id)))
283 except TypeError, ValueError:
284 _LOGGER.warning("SUPERNOTIFY telegram: chat_id %r is not numeric (expected int), skipped", raw_chat_id)
285 entity_ids: list[str] = envelope.target.domain_entity_ids("notify") if envelope.target else []
287 if not chat_ids and not entity_ids:
288 _LOGGER.warning("SUPERNOTIFY telegram: no chat_id or telegram notify entity to send to")
289 envelope.skipped = 1
290 envelope.skip_reason = SuppressionReason.NO_TARGET
291 return False
293 # Validate and normalise parse_mode. The HA telegram_bot service
294 # schema accepts only lowercase values: html / markdown / markdownv2
295 # / plain_text. We accept the camel-cased aliases the user may type
296 # (HTML, Markdown, MarkdownV2) and normalise.
297 # If the user does NOT specify a parse_mode, we default to plain_text
298 # rather than relying on the telegram_bot service default (markdown),
299 # because plain text bodies often contain `_` or `*` characters
300 # (e.g. "protect_content") that markdown would interpret as opening
301 # italic/bold and trigger "Can't parse entities" errors at Telegram.
302 if parse_mode:
303 normalised = str(parse_mode).lower()
304 if normalised not in ("html", "markdown", "markdownv2", "plain_text"):
305 _LOGGER.warning("SUPERNOTIFY telegram: invalid parse_mode '%s', ignoring", parse_mode)
306 parse_mode = "plain_text"
307 else:
308 parse_mode = normalised
309 else:
310 parse_mode = "plain_text"
312 # Map priority to disable_notification boolean
313 if disable_notification_override is not None:
314 disable_notification = boolify(disable_notification_override, default=False)
315 else:
316 disable_notification = _PRIORITY_MAP.get(envelope.priority or "medium", False)
318 # Build message text with title if present
319 message_text = envelope.message or ""
320 if envelope.title:
321 if parse_mode == "html":
322 # The body may already contain HTML the user wrote intentionally
323 # (e.g. <b>...</b>). Only escape the title (which is typically
324 # plain text) and prepend it bolded.
325 message_text = f"<b>{_escape_html(envelope.title)}</b>\n\n{message_text}"
326 elif parse_mode in ("markdown", "markdownv2"):
327 message_text = f"*{_escape_markdown(envelope.title, parse_mode)}*\n\n{message_text}"
328 else:
329 message_text = f"{envelope.title}\n\n{message_text}"
331 # Convert actions to inline keyboard. The HA telegram_bot service
332 # expects rows of [label, callback_or_url] 2-element lists. If the
333 # user passes a Telegram Bot API native dict shape
334 # ([[{text, callback_data}, ...], ...]), normalise it on the fly.
335 inline_keyboard = None
336 if custom_keyboard:
337 inline_keyboard = _normalise_inline_keyboard(custom_keyboard)
338 elif envelope.actions:
339 inline_keyboard = _build_inline_keyboard(envelope.actions)
341 # Recipients: chat ids, and notify entities, which also tell HA which bot to use.
342 # send_message/send_photo/send_document all accept the same keys.
343 action_data: dict[str, Any] = {}
344 if chat_ids:
345 action_data["chat_id"] = chat_ids
346 if entity_ids:
347 action_data["entity_id"] = entity_ids
349 # Determine if we have an image to attach
350 image_path = None
351 if attach_image:
352 image_path = await self.safe_grab_image(envelope)
353 _LOGGER.debug("SUPERNOTIFY telegram: image grabbed at %s", image_path)
355 # Select service and build service-specific payload
356 service_action = "telegram_bot.send_message"
358 if image_path and image_as_document:
359 # Send image as document (no compression). The telegram_bot HA
360 # service schema uses `file` for both send_photo and send_document
361 # (local path or http(s) URL).
362 service_action = "telegram_bot.send_document"
363 action_data["file"] = str(image_path)
364 if message_text:
365 action_data["caption"], action_data["parse_mode"] = _fit(message_text, _MAX_CAPTION_LENGTH, parse_mode)
366 elif image_path:
367 # Send image as photo with caption (`file` is the schema field
368 # name; the telegram_bot service infers content-type).
369 service_action = "telegram_bot.send_photo"
370 action_data["file"] = str(image_path)
371 if message_text:
372 action_data["caption"], action_data["parse_mode"] = _fit(message_text, _MAX_CAPTION_LENGTH, parse_mode)
373 else:
374 # Send text message
375 action_data["message"], action_data["parse_mode"] = _fit(message_text, _MAX_MESSAGE_LENGTH, parse_mode)
377 # Add optional parameters
378 if disable_notification:
379 action_data["disable_notification"] = True
381 # protect_content is intentionally NOT forwarded - the HA telegram_bot
382 # service schema only accepts it for send_media_group. Kept as a
383 # documented data key for forward-compatibility once HA exposes it.
385 if reply_to_message_id:
386 action_data["reply_to_message_id"] = reply_to_message_id
388 if inline_keyboard:
389 action_data["inline_keyboard"] = inline_keyboard
391 # Only what the telegram_bot schema accepts, since it rejects any other key, and a
392 # notification's own `data` is shared with every delivery
393 dropped: list[str] = sorted(k for k in raw_data if k not in PASSTHROUGH_KEYS)
394 if dropped:
395 _LOGGER.debug("SUPERNOTIFY telegram: dropping data keys telegram_bot doesn't accept: %s", dropped)
396 action_data.update({k: v for k, v in raw_data.items() if k in PASSTHROUGH_KEYS})
398 # Use the base-class call_action() to invoke the dynamically chosen
399 # telegram_bot service (send_message / send_photo / send_document).
400 # call_action() handles call-record tracking, error capture, and the
401 # envelope.delivered/calls bookkeeping that SuperNotify uses to
402 # decide success vs. fallback. `implied_target=True` tells SuperNotify
403 # the target is implied by the in-payload chat_id/entity_id (so the
404 # `TargetRequired.ALWAYS` check does not skip the delivery).
405 return await self.call_action(
406 envelope,
407 qualified_action=service_action,
408 action_data=action_data,
409 implied_target=True,
410 )