Coverage for custom_components/supernotify/envelope.py: 99%
239 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +0000
1from __future__ import annotations
3import copy
4import logging
5import string
6import time
7import typing
8import uuid
9from typing import Any, cast
11from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE
12from homeassistant.helpers.template import is_template_string
13from jinja2 import TemplateError
15from custom_components.supernotify.people import Recipient
16from custom_components.supernotify.target import Target
18from .common import DupeCheckable
19from .const import (
20 ATTR_FORCE_RESEND,
21 ATTR_MEDIA,
22 ATTR_MEDIA_CAMERA_ENTITY_ID,
23 ATTR_MEDIA_CLIP_URL,
24 ATTR_MEDIA_SNAPSHOT_URL,
25 ATTR_MESSAGE_HTML,
26 ATTR_PRIORITY,
27 ATTR_SPOKEN_MESSAGE,
28 ATTR_TIMESTAMP,
29 PRIORITY_MEDIUM,
30)
31from .media_grab import grab_image
32from .model import (
33 ConditionVariables,
34 DataFilter,
35 DeliveryCustomization,
36 MessageOnlyPolicy,
37 SuppressionReason,
38 TargetRequired,
39 TransportFeature,
40)
41from .options import (
42 OPTION_DATA_KEYS_SELECT,
43 OPTION_MESSAGE_USAGE,
44 OPTION_SIMPLIFY_TEXT,
45 OPTION_STRIP_URLS,
46)
48if typing.TYPE_CHECKING:
49 from anyio import Path
50 from homeassistant.core import Context as HAContext
52 from custom_components.supernotify.common import CallRecord
54 from .context import Context
55 from .delivery import Delivery
56 from .notification import Notification
57 from .scenario import Scenario
59_LOGGER = logging.getLogger(__name__)
61HASH_PREP_TRANSLATION_TABLE = table = str.maketrans("", "", string.punctuation + string.digits)
64class Envelope(DupeCheckable):
65 """Wrap a notification with a specific set of targets and service data possibly customized for those targets"""
67 _SCENARIO_TEMPLATE_DIRECTIVE_KEYS = frozenset({"message_template", "title_template"})
69 def __init__(
70 self,
71 delivery: Delivery,
72 notification: Notification | None = None,
73 target: Target | None = None, # targets only for this delivery
74 data: dict[str, Any] | None = None,
75 context: Context | None = None, # notification data customized for this delivery
76 ha_context: HAContext | None = None, # calling HA service context, propagated to HA service calls
77 ) -> None:
78 self.target: Target = target or Target()
79 self.context: Context | None = context
80 self.ha_context: HAContext | None = ha_context
81 self.delivery_name: str = delivery.name
82 self.delivery: Delivery = delivery
83 self._notification = notification
84 self.notification_id = None
85 self.media = None
86 self.action_groups = None
87 self.priority = PRIORITY_MEDIUM
88 self._message: str | None = None
89 self._title: str | None = None
90 self.message_html: str | None = None
91 self.spoken_message: str | None = None
92 self.force_resend: bool = False
93 self.data: dict[str, Any] = {} # delivery/target/scenario/action data
94 self.actions: list[dict[str, Any]] = []
95 if notification:
96 delivery_config_data: dict[str, Any] = notification.delivery_data(delivery)
97 self._enabled_scenarios: dict[str, Scenario] = notification.enabled_scenarios
98 # in reverse of usual logic, delivery config wins over notification data for message and title
99 self._message = delivery_config_data.pop(ATTR_MESSAGE, notification.message)
100 self._title = delivery_config_data.pop(ATTR_TITLE, notification._title)
101 self.id = f"{notification.id}_{self.delivery_name}"
102 else:
103 # should be testing scenarios only
104 delivery_config_data = {}
105 self._enabled_scenarios = {}
106 self.id = str(uuid.uuid1())
107 if data:
108 self.data = copy.deepcopy(data)
109 if delivery_config_data:
110 # notification-level delivery override wins over scenario/delivery data
111 self.data |= delivery_config_data
112 else:
113 self.data = delivery_config_data or {}
115 if notification:
116 self.notification_id = notification.id
117 self.media = notification.media
118 self.action_groups = notification.action_groups
119 self.actions = notification.actions
120 self.priority = self.data.pop(ATTR_PRIORITY, notification.priority)
121 self.message_html = self.data.pop(ATTR_MESSAGE_HTML, notification.message_html)
122 self.spoken_message = self.data.pop(ATTR_SPOKEN_MESSAGE, notification.spoken_message)
123 self.force_resend = self.data.pop(ATTR_FORCE_RESEND, notification.force_resend)
125 self.timestamp_format: str | None = self.data.pop(ATTR_TIMESTAMP, None)
127 # from this point on `self.data` has no internal Supernotify fields
129 if notification and hasattr(notification, "condition_variables"): # yeuchh
130 self.condition_variables: ConditionVariables = notification.condition_variables
131 else:
132 self.condition_variables = ConditionVariables()
134 # Scenario/delivery overrides can express `data` values (e.g. volume,
135 # volume_level, method for alexa_announce) as Jinja2 templates. These
136 # must be resolved once, here, before any transport sees `self.data` -
137 # previously they were only rendered for the archive copy built in
138 # contents(), so the transport call itself received the raw template
139 # string (#64). Keep the pre-render copy so contents() can still show
140 # the raw template alongside the resolved value for debugging.
141 self._raw_data: dict[str, Any] = dict(self.data)
142 if self.context:
143 self.data = self._render_data_templates(self.data)
145 self.message = self._compute_message()
146 self.title = self._compute_title()
148 self.delivered: int = 0
149 self.error_count: int = 0
150 self.skipped: int = 0
151 self.skip_reason: SuppressionReason | None = None
152 self.calls: list[CallRecord] = []
153 self.failed_calls: list[CallRecord] = []
154 self.delivery_error: list[str] | None = None
156 def customize_data(self, data: dict[str, Any], prune_empty: bool = True) -> dict[str, Any]:
157 """Return data filtered by delivery data_keys_select option, pruning empty maps by default."""
158 if not data:
159 return data
160 rules = self.delivery.options.get(OPTION_DATA_KEYS_SELECT)
161 return DataFilter(rules).apply(data, prune_empty=prune_empty)
163 async def grab_image(self) -> Path | None:
164 """Grab an image from a camera, snapshot URL, MQTT Image etc"""
165 image_path: Path | None = None
166 if self._notification:
167 image_path = await grab_image(
168 self._notification, self.delivery, self._notification.context, ha_context=self.ha_context
169 )
170 return image_path
172 def core_action_data(self, force_message: bool = True) -> dict[str, Any]:
173 """Build the core set of `service_data` dict to pass to underlying notify service"""
174 # TODO: remove all logic, so only called to pre-populate `data`
175 data: dict[str, Any] = {}
176 # message is mandatory for notify platform
177 if self.message is None:
178 if force_message:
179 data[ATTR_MESSAGE] = ""
180 else:
181 data[ATTR_MESSAGE] = self.message
183 if self.timestamp_format and ATTR_MESSAGE in data:
184 data[ATTR_MESSAGE] = f"{data[ATTR_MESSAGE]} [{time.strftime(self.timestamp_format, time.localtime())}]"
185 if self.title is not None:
186 data[ATTR_TITLE] = self.title
187 return data
189 def contents(self, minimal: bool = True, **_kwargs: Any) -> dict[str, typing.Any]:
190 exclude_attrs: list[str] = ["_notification", "context", "ha_context", "condition_variables", "force_resend"]
191 if minimal:
192 exclude_attrs.append("delivery")
193 features: TransportFeature = self.delivery.transport.supported_features
194 if not features & TransportFeature.ACTIONS:
195 exclude_attrs.extend(["actions", "action_groups"])
196 if not features & TransportFeature.IMAGES and not features & TransportFeature.VIDEO:
197 exclude_attrs.append(ATTR_MEDIA)
198 if not features & TransportFeature.MESSAGE:
199 exclude_attrs.extend(["message_html", "message"])
200 if features & TransportFeature.SPOKEN:
201 exclude_attrs.append("message_html")
202 else:
203 exclude_attrs.append("spoken_message")
204 if not features & TransportFeature.TITLE:
205 exclude_attrs.append("title")
206 if self.delivery.target_required == TargetRequired.NEVER:
207 exclude_attrs.append("target")
209 json_ready = {k: v for k, v in self.__dict__.items() if k not in exclude_attrs and not k.startswith("_")}
210 json_ready["data"] = self._resolve_data_templates(self._raw_data)
211 json_ready["calls"] = [call.contents() for call in self.calls]
212 json_ready["failedcalls"] = [call.contents() for call in self.failed_calls]
213 return json_ready
215 def __eq__(self, other: Any | None) -> bool: # ruff: ignore[any-type]
216 """Specialized equality check for subset of attributes"""
217 if other is None or not isinstance(other, Envelope):
218 return False
219 return bool(
220 self.target == other.target
221 and self.delivery_name == other.delivery_name
222 and self.data == other.data
223 and self.notification_id == other.notification_id
224 )
226 def __repr__(self) -> str:
227 """Return a concise string representation of the Envelope.
229 The returned string includes the envelope's message, title, and delivery name
230 in the form: Envelope(message={message},title={title},delivery={delivery_name}).
232 Primarily intended for debugging and logging; note that attribute values are
233 inserted directly and may not be quoted or escaped.
234 """
235 return f"Envelope(message={self.message},title={self.title},delivery={self.delivery_name})"
237 def record_recipient_notifications(self, recorded_person_ids: set[str]) -> None:
238 """Update every involved recipient's notify.recipient_<name> entity (if it has one),
239 so its state (or, on HA < 2026.3, its last_notified attribute - see
240 RecipientNotifyEntity.record_notification()) reflects delivery regardless of which
241 target form the caller used - a plain person_id, an email/phone/mobile override,
242 notify.recipient_<name> itself, or anything else Recipient.initialize() folds into the
243 same Target. Delivery target selection keeps person_ids, and generate_targets() narrows
244 them to the recipients each envelope actually reaches (see _attach_person_ids()), so
245 that's the one reliable link back from an arbitrary envelope to the Recipient objects
246 it reached - see RecipientNotifyEntity.record_notification() for
247 why this call is needed at all rather than leaving it to HA's own NotifyEntity state
248 tracking.
250 `recorded_person_ids` are the recipients already recorded by the notification's other
251 envelopes, which this adds to, so a recipient reached by several deliveries is only
252 recorded once - each is a state write, and would otherwise show up in the logbook as
253 several identical entries at the same moment."""
254 if self.target is None or self.context is None:
255 return
256 for person_id in self.target.person_ids:
257 if person_id in recorded_person_ids:
258 continue
259 recipient: Recipient | None = self.context.people_registry.people.get(person_id)
260 if recipient is not None:
261 recorded_person_ids.add(person_id)
262 recipient.on_notification(self.ha_context)
264 def _compute_title(self, ignore_usage: bool = False) -> str | None:
265 # message and title reverse the usual defaulting, delivery config overrides runtime call
267 title: str | None = None
268 message_usage = self.delivery.option_str(OPTION_MESSAGE_USAGE)
269 if not ignore_usage and message_usage.upper() in (MessageOnlyPolicy.USE_TITLE, MessageOnlyPolicy.COMBINE_TITLE):
270 # Message sourced from title text, title field dropped
271 title = None
272 else:
273 title = self.delivery.title if self.delivery.title is not None else self._title
274 title = self._render_scenario_templates(title, "title_template", "notification_title")
275 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
276 title = self.delivery.transport.simplify(title, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
278 if title is None:
279 return None
280 return str(title)
282 def _spoken_message(self, msg: str | None) -> str | None:
283 """Alternative message only for spoken voice transports"""
284 return self.spoken_message if self.spoken_message is not None else msg
286 def _compute_message(self) -> str | None:
287 # message and title reverse the usual defaulting, delivery config overrides runtime call
289 # self._message could be top level `message` or `message` set in delivery override
290 msg: str | None = self.delivery.message if self.delivery.message is not None else self._message
291 if self.delivery.transport.supported_features & TransportFeature.SPOKEN:
292 msg = self._spoken_message(msg)
294 if msg and self.context and is_template_string(msg):
295 try:
296 context_vars = cast("dict[str,Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
297 template = self.context.hass_api.template(msg)
298 msg = template.async_render(variables=context_vars)
299 except Exception as e:
300 _LOGGER.warning("SUPERNOTIFY Rendering delivery message template for %s failed: %s", self.delivery_name, e)
302 message_usage: str = str(self.delivery.option_str(OPTION_MESSAGE_USAGE))
303 if message_usage.upper() == MessageOnlyPolicy.USE_TITLE:
304 title = self._compute_title(ignore_usage=True)
305 if title:
306 msg = title
307 elif message_usage.upper() == MessageOnlyPolicy.COMBINE_TITLE:
308 title = self._compute_title(ignore_usage=True)
309 if title:
310 msg = f"{title} {msg}"
312 msg = self._render_scenario_templates(msg, "message_template", "notification_message")
313 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
314 msg = self.delivery.transport.simplify(msg, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
316 if msg is None: # keep mypy happy
317 return None
318 return str(msg)
320 def _render_scenario_templates(self, original: str | None, template_field: str, matching_ctx: str) -> str | None:
321 """Apply templating to a field, like message or title"""
322 rendered = original if original is not None else ""
323 delivery_configs: list[DeliveryCustomization] = list(
324 filter(None, (scenario.delivery_config(self.delivery.name) for scenario in self._enabled_scenarios.values()))
325 )
326 template_formats: list[str] = [
327 dc.data_value(template_field)
328 for dc in delivery_configs
329 if dc is not None and dc.data_value(template_field) is not None
330 ]
331 if template_formats and self.context:
332 if self.condition_variables:
333 context_vars: dict[str, Any] = cast("dict[str,Any]", self.condition_variables.as_dict())
334 else:
335 context_vars = {}
336 for template_format in template_formats:
337 context_vars[matching_ctx] = rendered
338 try:
339 template = self.context.hass_api.template(template_format)
340 rendered = template.async_render(variables=context_vars)
341 except TemplateError as e:
342 self.error_count += 1
343 _LOGGER.warning(
344 "SUPERNOTIFY Rendering template %s for %s failed: %s", template_field, self.delivery.name, e
345 )
346 return rendered
347 return original
349 # DupeCheckable implementation
351 def hash(self) -> int:
352 """Alpha hash to reduce noise from messages with timestamps or incrementing counts"""
354 def alphaize(v: str | None) -> str | None:
355 return v.translate(HASH_PREP_TRANSLATION_TABLE) if v else v
357 message: str | None
358 if self.delivery.transport.supported_features & TransportFeature.SPOKEN:
359 message = self._spoken_message(self._message)
360 else:
361 message = self._message
362 media = self.media or {}
363 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
364 media_url = media.get(ATTR_MEDIA_CLIP_URL) or media.get(ATTR_MEDIA_SNAPSHOT_URL)
365 return hash((
366 alphaize(message),
367 alphaize(self.delivery.name),
368 self.target.hash_resolved(),
369 alphaize(self._title),
370 camera_entity_id,
371 media_url,
372 ))
374 def _render_data_templates(self, data: dict[str, Any]) -> dict[str, Any]:
375 """Render Jinja2 templates in delivery `data` before the transport call.
377 Scenario or delivery-level overrides can set `data` values as Jinja2
378 template strings (see `_resolve_data_templates` for the archive/
379 diagnostics equivalent). Unlike that method, this returns a plain
380 resolved dict with no `<key>_template` breadcrumbs, so it is safe to
381 pass straight through to a transport's underlying HA service call.
382 """
383 if not data or not self.context:
384 return data
385 context_vars = cast("dict[str, Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
386 rendered: dict[str, Any] = {}
387 for key, value in data.items():
388 if key in self._SCENARIO_TEMPLATE_DIRECTIVE_KEYS:
389 # message_template/title_template are directives consumed by
390 # _render_scenario_templates() with its own chained render
391 # context, not literal data values - rendering them here
392 # would use the wrong (stale) context and produce unused junk.
393 rendered[key] = value
394 elif isinstance(value, str) and "{{" in value:
395 try:
396 rendered[key] = self.context.hass_api.template(value).async_render(variables=context_vars)
397 except Exception as e:
398 _LOGGER.warning(
399 "SUPERNOTIFY Rendering delivery data template for %s.%s failed: %s", self.delivery_name, key, e
400 )
401 rendered[key] = value
402 else:
403 rendered[key] = value
404 return rendered
406 def _resolve_data_templates(self, data: dict[str, Any]) -> dict[str, Any]:
407 """Resolve Jinja2 templates in data dict for archive readability.
409 Returns a copy of data with template strings replaced by their
410 resolved values. Raw template string is preserved alongside as
411 <key>_template for debugging. Non-template values are unchanged.
412 """
413 if not data or not self.context:
414 return data
415 resolved: dict[str, Any] = {}
416 context_vars = cast("dict[str, Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
417 for key, value in data.items():
418 if isinstance(value, str) and "{{" in value:
419 try:
420 rendered = self.context.hass_api.template(value).async_render(variables=context_vars)
421 resolved[key] = rendered
422 resolved[f"{key}_template"] = value
423 except Exception as e:
424 _LOGGER.debug("SUPERNOTIFY Could not resolve template for %s in %s: %s", key, self.delivery_name, e)
425 resolved[key] = value
426 else:
427 resolved[key] = value
428 return resolved