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