Coverage for custom_components/supernotify/transports/telegram.py: 100%
187 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +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.
7Supported data keys (all optional):
8 telegram_parse_mode str "HTML" | "Markdown" (default: None, plain text)
9 telegram_disable_notification bool Override silent mode (overrides priority mapping)
10 telegram_protect_content bool Block forward/save (default: False)
11 telegram_chat_id str|int Override target chat_id (default: None)
12 telegram_reply_to_message_id int Reply to message ID (default: None)
13 telegram_inline_keyboard list Custom action buttons in [[{text, callback_data}]] format
14 telegram_attach_image bool Attach camera snapshot (default: False)
15 telegram_image_as_document bool Send image as document without compression (default: False)
17Notes on the HA telegram_bot service schema:
18- `parse_mode` accepts only lowercase values: 'html', 'markdown',
19 'markdownv2', 'plain_text'. We normalise the user-provided value to
20 lowercase before forwarding.
21- `protect_content` is a Telegram Bot API parameter but the HA
22 `telegram_bot` integration does not currently expose it as a service-data
23 key (voluptuous rejects it as `extra keys not allowed`). We accept the
24 data key for forward-compatibility but do NOT forward it to the service.
25- `inline_keyboard` for HA is a list of rows where each row is a list of
26 `[label, callback_or_url]` 2-element lists (NOT dicts with
27 `text`/`callback_data` keys). Example: `[[["OK","/ok"],["Cancel","/cancel"]]]`.
28"""
30from __future__ import annotations
32import html
33import logging
34from typing import TYPE_CHECKING, Any, ClassVar, cast
36from custom_components.supernotify.common import boolify
37from custom_components.supernotify.const import TRANSPORT_TELEGRAM
38from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature
39from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption
40from custom_components.supernotify.transport import Transport
42if TYPE_CHECKING:
43 from custom_components.supernotify.envelope import Envelope
44 from custom_components.supernotify.hass_api import HomeAssistantAPI
45from homeassistant.helpers.typing import ConfigType
47_LOGGER = logging.getLogger(__name__)
49HA_TELEGRAM_BOT_DOMAIN = "telegram_bot"
51_PRIORITY_MAP = {
52 "critical": False, # notify (sound + vibration)
53 "high": False, # notify (sound)
54 "medium": False, # notify (sound)
55 "low": True, # silent
56 "minimum": True, # silent
57}
59# Telegram Bot API limits
60_MAX_MESSAGE_LENGTH = 4096
61_MAX_CAPTION_LENGTH = 1024
64def _escape_html(text: str) -> str:
65 """Escape HTML special characters for Telegram HTML parse mode."""
66 return html.escape(text) if text else ""
69def _normalise_inline_keyboard(keyboard: list) -> list:
70 """Normalise a user-supplied inline keyboard to the HA telegram_bot shape.
72 Accepts:
73 - HA-native: list of rows where each row is a list of `[label, callback]`
74 2-element lists. Returned unchanged.
75 - Telegram Bot API native: list of rows where each row is a list of
76 `{"text": label, "callback_data": callback}` dicts. Converted to the
77 HA shape.
78 - Mixed rows are tolerated (each button is normalised independently).
80 Returns the keyboard in the HA shape, or `[]` if the input is malformed.
81 """
82 if not isinstance(keyboard, list):
83 return []
84 out: list = []
85 for row in keyboard:
86 if not isinstance(row, list):
87 continue
88 out_row: list = []
89 for btn in row:
90 if isinstance(btn, list) and len(btn) >= 2:
91 # Already in HA shape: [label, callback_or_url]
92 out_row.append([str(btn[0]), str(btn[1])])
93 elif isinstance(btn, dict):
94 label = btn.get("text") or btn.get("title") or btn.get("label")
95 callback = btn.get("callback_data") or btn.get("action") or btn.get("url")
96 if label and callback:
97 out_row.append([str(label), str(callback)])
98 if out_row:
99 out.append(out_row)
100 return out
103def _build_inline_keyboard(actions: list) -> list:
104 """Convert SuperNotify actions to HA telegram_bot inline keyboard format.
106 SuperNotify actions are dicts with keys "title" (button label)
107 and "action" (callback identifier). The HA telegram_bot service expects
108 a list of rows where each row is a list of [label, callback_or_url]
109 2-element lists (NOT dicts with text/callback_data keys).
110 Example shape: [[["OK", "/ack_ok"], ["Open HA", "/open"]]]
112 Limits to max 5 buttons (single row) and 64-byte callback_data.
113 """
114 if not actions or not isinstance(actions, list):
115 return []
117 row = []
118 for i, action in enumerate(actions):
119 if i >= 5:
120 _LOGGER.warning("SUPERNOTIFY telegram: more than 5 actions, truncating to 5")
121 break
123 if not isinstance(action, dict):
124 _LOGGER.warning("SUPERNOTIFY telegram: action not a dict, skipped")
125 continue
127 # SuperNotify action keys: "title" for label, "action" for callback id.
128 # Also tolerate Telegram-native "text"/"callback_data" for users who
129 # craft the action list manually.
130 label = action.get("title") or action.get("text") or action.get("label")
131 callback = action.get("action") or action.get("callback_data") or action.get("id")
133 if not label or not callback:
134 _LOGGER.warning("SUPERNOTIFY telegram: action missing 'title' or 'action' key, skipped")
135 continue
137 # Truncate callback_data to 64 bytes UTF-8 per Telegram API
138 callback_str = str(callback)
139 encoded = callback_str.encode("utf-8")
140 if len(encoded) > 64:
141 callback_str = encoded[:64].decode("utf-8", errors="ignore")
142 _LOGGER.warning("SUPERNOTIFY telegram: callback_data truncated to 64 bytes: %s", callback_str)
144 # HA format: [label, callback_or_url] 2-element list
145 row.append([str(label), callback_str])
147 return [row] if row else []
150class TelegramTransport(Transport):
151 """Notify via Telegram using Home Assistant telegram_bot integration."""
153 name = TRANSPORT_TELEGRAM
154 declared_options: ClassVar[list[DeliveryOption]] = [*MEDIA_OPTIONS]
156 def __init__(self, *args: Any, **kwargs: Any) -> None:
157 super().__init__(*args, **kwargs)
159 @property
160 def supported_features(self) -> TransportFeature:
161 return (
162 TransportFeature.MESSAGE
163 | TransportFeature.TITLE
164 | TransportFeature.IMAGES
165 | TransportFeature.ACTIONS
166 | TransportFeature.SNAPSHOT_IMAGE
167 )
169 @property
170 def default_config(self) -> TransportConfig:
171 config = TransportConfig()
172 config.delivery_defaults.action = "telegram_bot.send_message"
173 config.delivery_defaults.target_required = TargetRequired.ALWAYS
174 config.delivery_defaults.inclusion = self.inclusion_mode
175 return config
177 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
178 return hass_api.find_config_entry_data(HA_TELEGRAM_BOT_DOMAIN) is not None
180 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
181 return {self.name: {}}
183 def validate_action(self, action: str | None) -> bool:
184 """Validate that action is one of the supported telegram_bot services."""
185 if not action:
186 return False
187 return action in (
188 "telegram_bot.send_message",
189 "telegram_bot.send_photo",
190 "telegram_bot.send_document",
191 )
193 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
194 _LOGGER.debug("SUPERNOTIFY telegram %s", envelope.message)
196 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
198 # Pop Telegram-specific data keys
199 parse_mode: str | None = raw_data.pop("telegram_parse_mode", None)
200 disable_notification_override = raw_data.pop("telegram_disable_notification", None)
201 # `telegram_protect_content` is accepted but currently NOT forwarded
202 # because the HA telegram_bot service schema rejects it. Pop to keep
203 # it out of the residual raw_data merge below.
204 raw_data.pop("telegram_protect_content", None)
205 chat_id_override = raw_data.pop("telegram_chat_id", None)
206 reply_to_message_id = raw_data.pop("telegram_reply_to_message_id", None)
207 custom_keyboard = raw_data.pop("telegram_inline_keyboard", None)
208 attach_image = boolify(raw_data.pop("telegram_attach_image", False), default=False)
209 image_as_document = boolify(raw_data.pop("telegram_image_as_document", False), default=False)
211 # Resolve target chat_id.
212 # `envelope.delivery.target` is a SuperNotify `Target` object with a
213 # `.targets` attribute (dict[category, list[id]]), where category is
214 # ATTR_PHONE, ATTR_EMAIL, ATTR_MOBILE_APP_ID, etc. For Telegram we
215 # prefer `phone` (Telegram chat IDs are stored there by convention).
216 # Also accept legacy raw shapes: dict, list, or scalar string/int.
217 raw_target: Any = chat_id_override
218 if not raw_target and envelope.delivery:
219 # TODO: this should probably be envelope.target like all the other transports
220 raw_target = envelope.delivery.target
222 # Target object: extract first id from preferred categories
223 if hasattr(raw_target, "targets") and isinstance(raw_target.targets, dict):
224 categorised = raw_target.targets
225 # Prefer phone (Telegram convention), then any non-empty list
226 preferred = ("phone", "chat_id")
227 picked = None
228 for cat in preferred:
229 if categorised.get(cat):
230 picked = categorised[cat][0]
231 break
232 if not picked:
233 for v in categorised.values():
234 if isinstance(v, list) and v:
235 picked = v[0]
236 break
237 raw_target = picked
239 # Legacy dict shape (pre-Target object): pick first list value
240 if isinstance(raw_target, dict):
241 for v in raw_target.values():
242 if isinstance(v, list) and v:
243 raw_target = v[0]
244 break
245 if v:
246 raw_target = v
247 break
248 elif isinstance(raw_target, list) and raw_target:
249 raw_target = raw_target[0]
251 if not raw_target:
252 _LOGGER.warning("SUPERNOTIFY telegram: chat_id not configured in delivery data")
253 self.record_error("chat_id not configured", "deliver")
254 return False
256 # telegram_bot service expects int (or list of int). Group chat IDs are
257 # negative integers; channel usernames (@channel) would be string but
258 # are not currently supported by this transport.
259 try:
260 chat_id: int = int(cast("str | int", raw_target))
261 except (TypeError, ValueError):
262 _LOGGER.warning("SUPERNOTIFY telegram: chat_id %r is not numeric (expected int)", raw_target)
263 self.record_error(f"chat_id {raw_target!r} not numeric", "deliver")
264 return False
266 # Validate and normalise parse_mode. The HA telegram_bot service
267 # schema accepts only lowercase values: html / markdown / markdownv2
268 # / plain_text. We accept the camel-cased aliases the user may type
269 # (HTML, Markdown, MarkdownV2) and normalise.
270 # If the user does NOT specify a parse_mode, we default to plain_text
271 # rather than relying on the telegram_bot service default (markdown),
272 # because plain text bodies often contain `_` or `*` characters
273 # (e.g. "protect_content") that markdown would interpret as opening
274 # italic/bold and trigger "Can't parse entities" errors at Telegram.
275 if parse_mode:
276 normalised = str(parse_mode).lower()
277 if normalised not in ("html", "markdown", "markdownv2", "plain_text"):
278 _LOGGER.warning("SUPERNOTIFY telegram: invalid parse_mode '%s', ignoring", parse_mode)
279 parse_mode = "plain_text"
280 else:
281 parse_mode = normalised
282 else:
283 parse_mode = "plain_text"
285 # Map priority to disable_notification boolean
286 if disable_notification_override is not None:
287 disable_notification = boolify(disable_notification_override, default=False)
288 else:
289 disable_notification = _PRIORITY_MAP.get(envelope.priority or "medium", False)
291 # Build message text with title if present
292 message_text = envelope.message or ""
293 if envelope.title:
294 if parse_mode == "html":
295 # The body may already contain HTML the user wrote intentionally
296 # (e.g. <b>...</b>). Only escape the title (which is typically
297 # plain text) and prepend it bolded.
298 message_text = f"<b>{_escape_html(envelope.title)}</b>\n\n{message_text}"
299 elif parse_mode in ("markdown", "markdownv2"):
300 message_text = f"*{envelope.title}*\n\n{message_text}"
301 else:
302 message_text = f"{envelope.title}\n\n{message_text}"
304 # Truncate message to Telegram limit
305 if len(message_text) > _MAX_MESSAGE_LENGTH:
306 message_text = message_text[:_MAX_MESSAGE_LENGTH]
307 _LOGGER.debug("SUPERNOTIFY telegram: message truncated to %d chars", _MAX_MESSAGE_LENGTH)
309 # Convert actions to inline keyboard. The HA telegram_bot service
310 # expects rows of [label, callback_or_url] 2-element lists. If the
311 # user passes a Telegram Bot API native dict shape
312 # ([[{text, callback_data}, ...], ...]), normalise it on the fly.
313 inline_keyboard = None
314 if custom_keyboard:
315 inline_keyboard = _normalise_inline_keyboard(custom_keyboard)
316 elif envelope.actions:
317 inline_keyboard = _build_inline_keyboard(envelope.actions)
319 # Build base action data. The HA telegram_bot service schema uses
320 # `target` (list of int chat IDs) as the primary recipient field; we
321 # pass a single-element list for one chat. send_message/send_photo/
322 # send_document all accept the same `target` key.
323 action_data: dict[str, Any] = {"target": [chat_id]}
325 # Determine if we have an image to attach
326 image_path = None
327 if attach_image:
328 try:
329 image_path = await envelope.grab_image()
330 _LOGGER.debug("SUPERNOTIFY telegram: image grabbed at %s", image_path)
331 except Exception as e:
332 _LOGGER.warning("SUPERNOTIFY telegram: failed to grab image: %s", e)
333 image_path = None
335 # Select service and build service-specific payload
336 service_action = "telegram_bot.send_message"
338 if image_path and image_as_document:
339 # Send image as document (no compression). The telegram_bot HA
340 # service schema uses `file` for both send_photo and send_document
341 # (local path or http(s) URL).
342 service_action = "telegram_bot.send_document"
343 action_data["file"] = str(image_path)
344 if message_text:
345 action_data["caption"] = message_text[:_MAX_CAPTION_LENGTH]
346 action_data["parse_mode"] = parse_mode
347 elif image_path:
348 # Send image as photo with caption (`file` is the schema field
349 # name; the telegram_bot service infers content-type).
350 service_action = "telegram_bot.send_photo"
351 action_data["file"] = str(image_path)
352 if message_text:
353 action_data["caption"] = message_text[:_MAX_CAPTION_LENGTH]
354 action_data["parse_mode"] = parse_mode
355 else:
356 # Send text message
357 action_data["message"] = message_text
358 action_data["parse_mode"] = parse_mode
360 # Add optional parameters
361 if disable_notification:
362 action_data["disable_notification"] = True
364 # protect_content is intentionally NOT forwarded - the HA
365 # telegram_bot service schema does not accept this key (voluptuous
366 # rejects it as `extra keys not allowed`). Kept as a documented data
367 # key for forward-compatibility once HA exposes it.
369 if reply_to_message_id:
370 action_data["reply_to_message_id"] = reply_to_message_id
372 if inline_keyboard:
373 action_data["inline_keyboard"] = inline_keyboard
375 # Merge remaining generic data keys
376 action_data.update(raw_data)
378 # Use the base-class call_action() to invoke the dynamically chosen
379 # telegram_bot service (send_message / send_photo / send_document).
380 # call_action() handles call-record tracking, error capture, and the
381 # envelope.delivered/calls bookkeeping that SuperNotify uses to
382 # decide success vs. fallback. `implied_target=True` tells SuperNotify
383 # the target is implied by the in-payload chat_id (so the
384 # `TargetRequired.ALWAYS` check does not skip the delivery).
385 return await self.call_action(
386 envelope,
387 qualified_action=service_action,
388 action_data=action_data,
389 implied_target=True,
390 )