Coverage for custom_components/supernotify/envelope.py: 99%
274 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import copy
4import hashlib
5import json
6import logging
7import string
8import time
9import typing
10import uuid
11from typing import Any, cast
13from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
14from homeassistant.helpers.template import is_template_string
15from jinja2 import TemplateError
17from custom_components.supernotify.people import Recipient
18from custom_components.supernotify.target import Target
20from .common import DupeCheckable, canonical_priority
21from .const import (
22 ACTION_LEVEL_NOTIFICATION,
23 ATTR_ACTION,
24 ATTR_FORCE_RESEND,
25 ATTR_MEDIA,
26 ATTR_MEDIA_CAMERA_ENTITY_ID,
27 ATTR_MEDIA_CLIP_URL,
28 ATTR_MESSAGE_HTML,
29 ATTR_PRIORITY,
30 ATTR_SPOKEN_MESSAGE,
31 ATTR_TIMESTAMP,
32 CONF_LEVEL,
33 CONF_TITLE,
34 CONF_URI,
35 PRIORITY_MEDIUM,
36)
37from .media_grab import grab_image
38from .model import (
39 ConditionVariables,
40 DataFilter,
41 DeliveryCustomization,
42 MessageOnlyPolicy,
43 SuppressionReason,
44 TargetRequired,
45 TransportFeature,
46)
47from .options import (
48 OPTION_DATA_KEYS_SELECT,
49 OPTION_MESSAGE_USAGE,
50 OPTION_SIMPLIFY_TEXT,
51 OPTION_STRIP_URLS,
52)
54if typing.TYPE_CHECKING:
55 from anyio import Path
56 from homeassistant.core import Context as HAContext
58 from custom_components.supernotify.common import CallRecord
60 from .context import Context
61 from .delivery import Delivery
62 from .notification import Notification
63 from .scenario import Scenario
65_LOGGER = logging.getLogger(__name__)
67HASH_PREP_TRANSLATION_TABLE = table = str.maketrans("", "", string.punctuation + string.digits)
70class Envelope(DupeCheckable):
71 """Wrap a notification with a specific set of targets and service data possibly customized for those targets"""
73 _SCENARIO_TEMPLATE_DIRECTIVE_KEYS = frozenset({"message_template", "title_template"})
75 def __init__(
76 self,
77 delivery: Delivery,
78 notification: Notification | None = None,
79 target: Target | None = None, # targets only for this delivery
80 data: dict[str, Any] | None = None,
81 context: Context | None = None, # notification data customized for this delivery
82 ha_context: HAContext | None = None, # calling HA service context, propagated to HA service calls
83 ) -> None:
84 self.target: Target = target or Target()
85 self.context: Context | None = context
86 self.ha_context: HAContext | None = ha_context
87 self.delivery_name: str = delivery.name
88 self.delivery: Delivery = delivery
89 self._notification = notification
90 self.notification_id = None
91 self.media = None
92 self.action_groups = None
93 self.priority = PRIORITY_MEDIUM
94 self._message: str | None = None
95 self._title: str | None = None
96 self.message_html: str | None = None
97 self.spoken_message: str | None = None
98 self.force_resend: bool = False
99 self.data: dict[str, Any] = {} # delivery/target/scenario/action data
100 self.actions: list[dict[str, Any]] = []
101 if notification:
102 delivery_config_data: dict[str, Any] = notification.delivery_data(delivery)
103 self._enabled_scenarios: dict[str, Scenario] = notification.enabled_scenarios
104 # in reverse of usual logic, delivery config wins over notification data for message and title
105 self._message = delivery_config_data.pop(ATTR_MESSAGE, notification.message)
106 self._title = delivery_config_data.pop(ATTR_TITLE, notification._title)
107 self.id = f"{notification.id}_{self.delivery_name}"
108 else:
109 # should be testing scenarios only
110 delivery_config_data = {}
111 self._enabled_scenarios = {}
112 self.id = str(uuid.uuid1())
113 if data:
114 self.data = copy.deepcopy(data)
115 if delivery_config_data:
116 # notification-level delivery override wins over scenario/delivery data
117 self.data |= delivery_config_data
118 else:
119 self.data = delivery_config_data or {}
121 if notification:
122 self.notification_id = notification.id
123 self.media = notification.media
124 self.action_groups = notification.action_groups
125 self.actions = notification.actions
126 self.priority = canonical_priority(self.data.pop(ATTR_PRIORITY, notification.priority))
127 self.message_html = self.data.pop(ATTR_MESSAGE_HTML, notification.message_html)
128 self.spoken_message = self.data.pop(ATTR_SPOKEN_MESSAGE, notification.spoken_message)
129 self.force_resend = self.data.pop(ATTR_FORCE_RESEND, notification.force_resend)
131 self.timestamp_format: str | None = self.data.pop(ATTR_TIMESTAMP, None)
133 # from this point on `self.data` has no internal Supernotify fields
135 if notification and hasattr(notification, "condition_variables"): # yeuchh
136 self.condition_variables: ConditionVariables = notification.condition_variables
137 else:
138 self.condition_variables = ConditionVariables()
140 # Scenario/delivery overrides can express `data` values (e.g. volume,
141 # volume_level, method for alexa_announce) as Jinja2 templates. These
142 # must be resolved once, here, before any transport sees `self.data` -
143 # previously they were only rendered for the archive copy built in
144 # contents(), so the transport call itself received the raw template
145 # string (#64). Keep the pre-render copy so contents() can still show
146 # the raw template alongside the resolved value for debugging.
147 self._raw_data: dict[str, Any] = dict(self.data)
148 if self.context:
149 self.data = self._render_data_templates(self.data)
151 self.message = self._compute_message()
152 self.title = self._compute_title()
154 self.delivered: int = 0
155 self.error_count: int = 0
156 self.skipped: int = 0
157 self.skip_reason: SuppressionReason | None = None
158 self.calls: list[CallRecord] = []
159 self.failed_calls: list[CallRecord] = []
160 self.delivery_error: list[str] | None = None
162 def customize_data(self, data: dict[str, Any], prune_empty: bool = True) -> dict[str, Any]:
163 """Return data filtered by delivery data_keys_select option, pruning empty maps by default."""
164 if not data:
165 return data
166 rules = self.delivery.options.get(OPTION_DATA_KEYS_SELECT)
167 return DataFilter(rules).apply(data, prune_empty=prune_empty)
169 async def grab_image(self) -> Path | None:
170 """Grab an image from a camera, snapshot URL, MQTT Image etc"""
171 image_path: Path | None = None
172 if self._notification:
173 image_path = await grab_image(
174 self._notification, self.delivery, self._notification.context, ha_context=self.ha_context
175 )
176 return image_path
178 def render_action(self, action: dict[str, Any]) -> dict[str, Any]:
179 """Mobile action with its `action_template`, `title_template` and `uri_template` rendered, and `level` dropped"""
180 rendered: dict[str, Any] = {k: v for k, v in action.items() if k != CONF_LEVEL}
181 for field in (ATTR_ACTION, CONF_TITLE, CONF_URI):
182 template_format = rendered.pop(f"{field}_template", None)
183 if template_format is None or self.context is None:
184 continue
185 try:
186 context_vars = cast("dict[str,Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
187 rendered[field] = self.context.hass_api.template(template_format).async_render(variables=context_vars)
188 except TemplateError as e:
189 self.error_count += 1
190 _LOGGER.warning("SUPERNOTIFY Rendering %s_template for %s failed: %s", field, self.delivery_name, e)
191 return rendered
193 def notification_uri(self, absolute: bool = False) -> str | None:
194 """URI to open when the notification itself is tapped, from a `level: notification` mobile action
196 Where there's more than one, an action on the notification wins, then the action groups
197 in the order chosen, which is those on the notification followed by each scenario in turn.
198 `absolute` is for where a path relative to Home Assistant, like `/lovelace/cameras`, won't do.
199 """
200 groups: dict[str, list[dict[str, Any]]] = self.context.mobile_actions if self.context else {}
201 group_names: list[str] = list(groups) if self.action_groups is None else self.action_groups
202 for action in [*self.actions, *(action for name in group_names for action in groups.get(name, []))]:
203 if action.get(CONF_LEVEL) == ACTION_LEVEL_NOTIFICATION:
204 uri: str | None = self.render_action(action).get(CONF_URI)
205 if uri and absolute and self.context and str(uri).startswith("/"):
206 return self.context.hass_api.abs_url(str(uri))
207 if uri:
208 return str(uri)
209 return None
211 def core_action_data(self, force_message: bool = True) -> dict[str, Any]:
212 """Build the core set of `service_data` dict to pass to underlying notify service"""
213 # TODO: remove all logic, so only called to pre-populate `data`
214 data: dict[str, Any] = {}
215 # message is mandatory for notify platform
216 if self.message is None:
217 if force_message:
218 data[ATTR_MESSAGE] = ""
219 else:
220 data[ATTR_MESSAGE] = self.message
222 if self.timestamp_format and ATTR_MESSAGE in data:
223 data[ATTR_MESSAGE] = f"{data[ATTR_MESSAGE]} [{time.strftime(self.timestamp_format, time.localtime())}]"
224 if self.title is not None:
225 data[ATTR_TITLE] = self.title
226 return data
228 def contents(self, minimal: bool = True, **_kwargs: Any) -> dict[str, typing.Any]:
229 exclude_attrs: list[str] = ["_notification", "context", "ha_context", "condition_variables", "force_resend"]
230 if minimal:
231 exclude_attrs.append("delivery")
232 features: TransportFeature = self.delivery.transport.supported_features
233 if not features & TransportFeature.ACTIONS:
234 exclude_attrs.extend(["actions", "action_groups"])
235 if not features & TransportFeature.IMAGES and not features & TransportFeature.VIDEO:
236 exclude_attrs.append(ATTR_MEDIA)
237 if not features & TransportFeature.MESSAGE:
238 exclude_attrs.extend(["message_html", "message"])
239 if features & TransportFeature.SPOKEN:
240 exclude_attrs.append("message_html")
241 else:
242 exclude_attrs.append("spoken_message")
243 if not features & TransportFeature.TITLE:
244 exclude_attrs.append("title")
245 if self.delivery.target_required == TargetRequired.NEVER:
246 exclude_attrs.append("target")
248 json_ready = {k: v for k, v in self.__dict__.items() if k not in exclude_attrs and not k.startswith("_")}
249 json_ready["data"] = self._resolve_data_templates(self._raw_data)
250 json_ready["calls"] = [call.contents() for call in self.calls]
251 json_ready["failedcalls"] = [call.contents() for call in self.failed_calls]
252 return json_ready
254 def __eq__(self, other: Any | None) -> bool: # ruff: ignore[any-type]
255 """Specialized equality check for subset of attributes"""
256 if other is None or not isinstance(other, Envelope):
257 return False
258 return bool(
259 self.target == other.target
260 and self.delivery_name == other.delivery_name
261 and self.data == other.data
262 and self.notification_id == other.notification_id
263 )
265 def __repr__(self) -> str:
266 """Return a concise string representation of the Envelope.
268 The returned string includes the envelope's message, title, and delivery name
269 in the form: Envelope(message={message},title={title},delivery={delivery_name}).
271 Primarily intended for debugging and logging; note that attribute values are
272 inserted directly and may not be quoted or escaped.
273 """
274 return f"Envelope(message={self.message},title={self.title},delivery={self.delivery_name})"
276 def record_recipient_notifications(self, recorded_person_ids: set[str]) -> None:
277 """Update every involved recipient's notify.recipient_<name> entity (if it has one),
278 so its state (or, on HA < 2026.3, its last_notified attribute - see
279 RecipientNotifyEntity.record_notification()) reflects delivery regardless of which
280 target form the caller used - a plain person_id, an email/phone/mobile override,
281 notify.recipient_<name> itself, or anything else Recipient.initialize() folds into the
282 same Target. Delivery target selection keeps person_ids, and generate_targets() narrows
283 them to the recipients each envelope actually reaches (see _attach_person_ids()), so
284 that's the one reliable link back from an arbitrary envelope to the Recipient objects
285 it reached - see RecipientNotifyEntity.record_notification() for
286 why this call is needed at all rather than leaving it to HA's own NotifyEntity state
287 tracking.
289 `recorded_person_ids` are the recipients already recorded by the notification's other
290 envelopes, which this adds to, so a recipient reached by several deliveries is only
291 recorded once - each is a state write, and would otherwise show up in the logbook as
292 several identical entries at the same moment."""
293 if self.target is None or self.context is None:
294 return
295 for person_id in self.target.person_ids:
296 if person_id in recorded_person_ids:
297 continue
298 recipient: Recipient | None = self.context.people_registry.people.get(person_id)
299 if recipient is not None:
300 recorded_person_ids.add(person_id)
301 recipient.on_notification(self.ha_context)
303 def _compute_title(self, ignore_usage: bool = False) -> str | None:
304 # message and title reverse the usual defaulting, delivery config overrides runtime call
306 title: str | None = None
307 message_usage = self.delivery.option_str(OPTION_MESSAGE_USAGE)
308 if not ignore_usage and message_usage.upper() in (MessageOnlyPolicy.USE_TITLE, MessageOnlyPolicy.COMBINE_TITLE):
309 # Message sourced from title text, title field dropped
310 title = None
311 else:
312 title = self.delivery.title if self.delivery.title is not None else self._title
313 title = self._render_scenario_templates(title, "title_template", "notification_title")
314 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
315 title = self.delivery.transport.simplify(title, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
317 if title is None:
318 return None
319 return str(title)
321 def _spoken_message(self, msg: str | None) -> str | None:
322 """Alternative message only for spoken voice transports"""
323 return self.spoken_message if self.spoken_message is not None else msg
325 def _message_carries_title(self) -> bool:
326 """The message is all that's seen or heard - no title field, or spoken where any title is only on a screen"""
327 features = self.delivery.transport.supported_features
328 return not features & TransportFeature.TITLE or bool(features & TransportFeature.SPOKEN)
330 def _compute_message(self) -> str | None:
331 # message and title reverse the usual defaulting, delivery config overrides runtime call
333 # self._message could be top level `message` or `message` set in delivery override
334 msg: str | None = self.delivery.message if self.delivery.message is not None else self._message
335 if self.delivery.transport.supported_features & TransportFeature.SPOKEN:
336 msg = self._spoken_message(msg)
338 if msg and self.context and is_template_string(msg):
339 try:
340 context_vars = cast("dict[str,Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
341 template = self.context.hass_api.template(msg)
342 msg = template.async_render(variables=context_vars)
343 except Exception as e:
344 _LOGGER.warning("SUPERNOTIFY Rendering delivery message template for %s failed: %s", self.delivery_name, e)
346 message_usage: str = str(self.delivery.option_str(OPTION_MESSAGE_USAGE))
347 if message_usage.upper() == MessageOnlyPolicy.USE_TITLE:
348 title = self._compute_title(ignore_usage=True)
349 if title:
350 msg = title
351 elif message_usage.upper() == MessageOnlyPolicy.COMBINE_TITLE:
352 title = self._compute_title(ignore_usage=True)
353 if title:
354 msg = f"{title} {msg}" if msg else title
355 elif not msg and message_usage.upper() != MessageOnlyPolicy.MESSAGE_ONLY and self._message_carries_title():
356 msg = self._compute_title(ignore_usage=True)
358 msg = self._render_scenario_templates(msg, "message_template", "notification_message")
359 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
360 msg = self.delivery.transport.simplify(msg, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
362 if msg is None: # keep mypy happy
363 return None
364 return str(msg)
366 def _render_scenario_templates(self, original: str | None, template_field: str, matching_ctx: str) -> str | None:
367 """Apply templating to a field, like message or title"""
368 rendered = original if original is not None else ""
369 delivery_configs: list[DeliveryCustomization] = list(
370 filter(None, (scenario.delivery_config(self.delivery.name) for scenario in self._enabled_scenarios.values()))
371 )
372 template_formats: list[str] = [
373 dc.data_value(template_field)
374 for dc in delivery_configs
375 if dc is not None and dc.data_value(template_field) is not None
376 ]
377 if template_formats and self.context:
378 if self.condition_variables:
379 context_vars: dict[str, Any] = cast("dict[str,Any]", self.condition_variables.as_dict())
380 else:
381 context_vars = {}
382 for template_format in template_formats:
383 context_vars[matching_ctx] = rendered
384 try:
385 template = self.context.hass_api.template(template_format)
386 rendered = template.async_render(variables=context_vars)
387 except TemplateError as e:
388 self.error_count += 1
389 _LOGGER.warning(
390 "SUPERNOTIFY Rendering template %s for %s failed: %s", template_field, self.delivery.name, e
391 )
392 return rendered
393 return original
395 # DupeCheckable implementation
397 def hash(self) -> int:
398 """Alpha hash to reduce noise from messages with timestamps or incrementing counts"""
400 def alphaize(v: str | None) -> str | None:
401 return v.translate(HASH_PREP_TRANSLATION_TABLE) if v else v
403 message: str | None
404 if self.delivery.transport.supported_features & TransportFeature.SPOKEN:
405 message = self._spoken_message(self._message)
406 else:
407 message = self._message
408 media = self.media or {}
409 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
410 # snapshot_url is excluded: Frigate mints a unique thumbnail per detection even for
411 # the same ongoing scene, so including it defeats dupe detection for snapshot-only alerts
412 media_url = media.get(ATTR_MEDIA_CLIP_URL)
413 return hash((
414 alphaize(message),
415 alphaize(self.delivery.name),
416 self.target.hash_resolved(),
417 alphaize(self._title),
418 camera_entity_id,
419 media_url,
420 ))
422 def content_key(self) -> str:
423 """What this envelope would actually make the transport do - the transport, its action, the
424 resolved targets, and the exact text, data and media - but not which delivery it came from.
425 Two envelopes with the same key are the same delivery to the person receiving it, so a
426 fallback doesn't repeat what another delivery already sent.
428 A short blake2b digest rather than the built-in ``hash()``, so the key stays the same across
429 restarts and could be used in an id"""
430 media = self.media or {}
431 content: str = json.dumps(
432 [
433 self.delivery.transport.name,
434 self.delivery.action,
435 self.target.hash_resolved(),
436 self._message,
437 self._title,
438 self.message_html,
439 self.spoken_message,
440 self.data,
441 {k: media.get(k) for k in sorted(media)},
442 ],
443 sort_keys=True,
444 default=str,
445 )
446 return hashlib.blake2b(content.encode(), digest_size=8).hexdigest()
448 def _render_data_templates(self, data: dict[str, Any]) -> dict[str, Any]:
449 """Render Jinja2 templates in delivery `data` before the transport call.
451 Scenario or delivery-level overrides can set `data` values as Jinja2
452 template strings (see `_resolve_data_templates` for the archive/
453 diagnostics equivalent). Unlike that method, this returns a plain
454 resolved dict with no `<key>_template` breadcrumbs, so it is safe to
455 pass straight through to a transport's underlying HA service call.
456 """
457 if not data or not self.context:
458 return data
459 context_vars = cast("dict[str, Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
460 rendered: dict[str, Any] = {}
461 for key, value in data.items():
462 if key in self._SCENARIO_TEMPLATE_DIRECTIVE_KEYS:
463 # message_template/title_template are directives consumed by
464 # _render_scenario_templates() with its own chained render
465 # context, not literal data values - rendering them here
466 # would use the wrong (stale) context and produce unused junk.
467 rendered[key] = value
468 elif isinstance(value, str) and "{{" in value:
469 try:
470 rendered[key] = self.context.hass_api.template(value).async_render(variables=context_vars)
471 except Exception as e:
472 _LOGGER.warning(
473 "SUPERNOTIFY Rendering delivery data template for %s.%s failed: %s", self.delivery_name, key, e
474 )
475 rendered[key] = value
476 else:
477 rendered[key] = value
478 return rendered
480 def _resolve_data_templates(self, data: dict[str, Any]) -> dict[str, Any]:
481 """Resolve Jinja2 templates in data dict for archive readability.
483 Returns a copy of data with template strings replaced by their
484 resolved values. Raw template string is preserved alongside as
485 <key>_template for debugging. Non-template values are unchanged.
486 """
487 if not data or not self.context:
488 return data
489 resolved: dict[str, Any] = {}
490 context_vars = cast("dict[str, Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
491 for key, value in data.items():
492 if isinstance(value, str) and "{{" in value:
493 try:
494 rendered = self.context.hass_api.template(value).async_render(variables=context_vars)
495 resolved[key] = rendered
496 resolved[f"{key}_template"] = value
497 except Exception as e:
498 _LOGGER.debug("SUPERNOTIFY Could not resolve template for %s in %s: %s", key, self.delivery_name, e)
499 resolved[key] = value
500 else:
501 resolved[key] = value
502 return resolved