Coverage for custom_components / supernotify / envelope.py: 89%
194 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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 .common import DupeCheckable
16from .const import (
17 ATTR_MEDIA,
18 ATTR_MESSAGE_HTML,
19 ATTR_PRIORITY,
20 ATTR_SPOKEN_MESSAGE,
21 ATTR_TIMESTAMP,
22 OPTION_DATA_KEYS_SELECT,
23 OPTION_MESSAGE_USAGE,
24 OPTION_SIMPLIFY_TEXT,
25 OPTION_STRIP_URLS,
26 PRIORITY_MEDIUM,
27)
28from .media_grab import grab_image
29from .model import (
30 ConditionVariables,
31 DataFilter,
32 DeliveryCustomization,
33 MessageOnlyPolicy,
34 SuppressionReason,
35 Target,
36 TargetRequired,
37 TransportFeature,
38)
40if typing.TYPE_CHECKING:
41 from anyio import Path
43 from custom_components.supernotify.common import CallRecord
45 from .context import Context
46 from .delivery import Delivery
47 from .notification import Notification
48 from .scenario import Scenario
50_LOGGER = logging.getLogger(__name__)
52HASH_PREP_TRANSLATION_TABLE = table = str.maketrans("", "", string.punctuation + string.digits)
55class Envelope(DupeCheckable):
56 """Wrap a notification with a specific set of targets and service data possibly customized for those targets"""
58 def __init__(
59 self,
60 delivery: Delivery,
61 notification: Notification | None = None,
62 target: Target | None = None, # targets only for this delivery
63 data: dict[str, Any] | None = None,
64 context: Context | None = None, # notification data customized for this delivery
65 ) -> None:
66 self.target: Target = target or Target()
67 self.context: Context | None = context
68 self.delivery_name: str = delivery.name
69 self.delivery: Delivery = delivery
70 self._notification = notification
71 self.notification_id = None
72 self.media = None
73 self.action_groups = None
74 self.priority = PRIORITY_MEDIUM
75 self._message: str | None = None
76 self._title: str | None = None
77 self.message_html: str | None = None
78 self.data: dict[str, Any] = {}
79 self.actions: list[dict[str, Any]] = []
80 if notification:
81 delivery_config_data: dict[str, Any] = notification.delivery_data(delivery)
82 self._enabled_scenarios: dict[str, Scenario] = notification.enabled_scenarios
83 self._message = delivery_config_data.pop(ATTR_MESSAGE, notification.message)
84 self._title = delivery_config_data.pop(ATTR_TITLE, notification._title)
85 self.id = f"{notification.id}_{self.delivery_name}"
86 else:
87 delivery_config_data = {}
88 self._enabled_scenarios = {}
89 self.id = str(uuid.uuid1())
90 if data:
91 self.data = copy.deepcopy(data)
92 if delivery_config_data:
93 # notification-level delivery override wins over scenario/delivery data
94 self.data |= delivery_config_data
95 else:
96 self.data = delivery_config_data if delivery_config_data else {}
98 if notification:
99 self.notification_id = notification.id
100 self.media = notification.media
101 self.action_groups = notification.action_groups
102 self.actions = notification.actions
103 self.priority = self.data.get(ATTR_PRIORITY, notification.priority)
104 self.message_html = self.data.get(ATTR_MESSAGE_HTML, notification.message_html)
105 if notification and hasattr(notification, "condition_variables"): # yeuchh
106 self.condition_variables = notification.condition_variables
107 else:
108 self.condition_variables = ConditionVariables()
110 self.message = self._compute_message()
111 self.title = self._compute_title()
113 self.delivered: int = 0
114 self.error_count: int = 0
115 self.skipped: int = 0
116 self.skip_reason: SuppressionReason | None = None
117 self.calls: list[CallRecord] = []
118 self.failed_calls: list[CallRecord] = []
119 self.delivery_error: list[str] | None = None
121 def customize_data(self, data: dict[str, Any], prune_empty: bool = True) -> dict[str, Any]:
122 """Return data filtered by delivery data_keys_select option, pruning empty maps by default."""
123 if not data:
124 return data
125 rules = self.delivery.options.get(OPTION_DATA_KEYS_SELECT)
126 return DataFilter(rules).apply(data, prune_empty=prune_empty)
128 async def grab_image(self) -> Path | None:
129 """Grab an image from a camera, snapshot URL, MQTT Image etc"""
130 image_path: Path | None = None
131 if self._notification:
132 image_path = await grab_image(self._notification, self.delivery, self._notification.context)
133 return image_path
135 def core_action_data(self, force_message: bool = True) -> dict[str, Any]:
136 """Build the core set of `service_data` dict to pass to underlying notify service"""
137 # TODO: remove all logic, so only called to pre-populate `data`
138 data: dict[str, Any] = {}
139 # message is mandatory for notify platform
140 if self.message is None:
141 if force_message:
142 data[ATTR_MESSAGE] = ""
143 else:
144 data[ATTR_MESSAGE] = self.message
145 timestamp = self.data.get(ATTR_TIMESTAMP)
146 if timestamp and ATTR_MESSAGE in data:
147 data[ATTR_MESSAGE] = f"{data[ATTR_MESSAGE]} [{time.strftime(timestamp, time.localtime())}]"
148 if self.title is not None:
149 data[ATTR_TITLE] = self.title
150 return data
152 def contents(self, minimal: bool = True, **_kwargs: Any) -> dict[str, typing.Any]:
153 exclude_attrs: list[str] = ["_notification", "context", "condition_variables"]
154 if minimal:
155 exclude_attrs.append("delivery")
156 features: TransportFeature = self.delivery.transport.supported_features
157 if not features & TransportFeature.ACTIONS:
158 exclude_attrs.extend(["actions", "action_groups"])
159 if not features & TransportFeature.IMAGES and not features & TransportFeature.VIDEO:
160 exclude_attrs.append(ATTR_MEDIA)
161 if not features & TransportFeature.MESSAGE:
162 exclude_attrs.extend(["message_html", "message"])
163 if not features & TransportFeature.TITLE:
164 exclude_attrs.append("title")
165 if self.delivery.target_required == TargetRequired.NEVER:
166 exclude_attrs.append("target")
168 json_ready = {k: v for k, v in self.__dict__.items() if k not in exclude_attrs and not k.startswith("_")}
169 json_ready["data"] = self._resolve_data_templates(self.data)
170 json_ready["calls"] = [call.contents() for call in self.calls]
171 json_ready["failedcalls"] = [call.contents() for call in self.failed_calls]
172 return json_ready
174 def __eq__(self, other: Any | None) -> bool:
175 """Specialized equality check for subset of attributes"""
176 if other is None or not isinstance(other, Envelope):
177 return False
178 return bool(
179 self.target == other.target
180 and self.delivery_name == other.delivery_name
181 and self.data == other.data
182 and self.notification_id == other.notification_id
183 )
185 def __repr__(self) -> str:
186 """Return a concise string representation of the Envelope.
188 The returned string includes the envelope's message, title, and delivery name
189 in the form: Envelope(message=<message>,title=<title>,delivery=<delivery_name>).
191 Primarily intended for debugging and logging; note that attribute values are
192 inserted directly and may not be quoted or escaped.
193 """
194 return f"Envelope(message={self.message},title={self.title},delivery={self.delivery_name})"
196 def _compute_title(self, ignore_usage: bool = False) -> str | None:
197 # message and title reverse the usual defaulting, delivery config overrides runtime call
199 title: str | None = None
200 message_usage = self.delivery.option_str(OPTION_MESSAGE_USAGE)
201 if not ignore_usage and message_usage.upper() in (MessageOnlyPolicy.USE_TITLE, MessageOnlyPolicy.COMBINE_TITLE):
202 title = None
203 else:
204 title = self.delivery.title if self.delivery.title is not None else self._title
205 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
206 title = self.delivery.transport.simplify(title, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
207 title = self._render_scenario_templates(title, "title_template", "notification_title")
208 if title is None:
209 return None
210 return str(title)
212 def _spoken_message(self) -> str | None:
213 """Alternative message only for spoken voice transports"""
214 if (
215 self._notification
216 and self._notification.extra_data
217 and ATTR_SPOKEN_MESSAGE in self._notification.extra_data
218 and self.delivery.transport.supported_features & TransportFeature.SPOKEN
219 ):
220 return str(self._notification.extra_data[ATTR_SPOKEN_MESSAGE])
221 return None
223 def _compute_message(self) -> str | None:
224 # message and title reverse the usual defaulting, delivery config overrides runtime call
226 # self._message could be top level `message` or `message` set in delivery override
227 msg: str | None = self.delivery.message if self.delivery.message is not None else self._message
228 msg = self._spoken_message() or msg
230 if msg and self.context and is_template_string(msg):
231 try:
232 context_vars = cast("dict[str,Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
233 template = self.context.hass_api.template(msg)
234 msg = template.async_render(variables=context_vars)
235 except Exception as e:
236 _LOGGER.warning("SUPERNOTIFY Rendering delivery message template for %s failed: %s", self.delivery_name, e)
238 message_usage: str = str(self.delivery.option_str(OPTION_MESSAGE_USAGE))
239 if message_usage.upper() == MessageOnlyPolicy.USE_TITLE:
240 title = self._compute_title(ignore_usage=True)
241 if title:
242 msg = title
243 elif message_usage.upper() == MessageOnlyPolicy.COMBINE_TITLE:
244 title = self._compute_title(ignore_usage=True)
245 if title:
246 msg = f"{title} {msg}"
248 if self.delivery.option_bool(OPTION_SIMPLIFY_TEXT) is True or self.delivery.option_bool(OPTION_STRIP_URLS) is True:
249 msg = self.delivery.transport.simplify(msg, strip_urls=self.delivery.option_bool(OPTION_STRIP_URLS))
251 msg = self._render_scenario_templates(msg, "message_template", "notification_message")
252 if msg is None: # keep mypy happy
253 return None
254 return str(msg)
256 def _render_scenario_templates(self, original: str | None, template_field: str, matching_ctx: str) -> str | None:
257 """Apply templating to a field, like message or title"""
258 rendered = original if original is not None else ""
259 delivery_configs: list[DeliveryCustomization] = list(
260 filter(None, (scenario.delivery_config(self.delivery.name) for scenario in self._enabled_scenarios.values()))
261 )
262 template_formats: list[str] = [
263 dc.data_value(template_field)
264 for dc in delivery_configs
265 if dc is not None and dc.data_value(template_field) is not None
266 ]
267 if template_formats and self.context:
268 if self.condition_variables:
269 context_vars: dict[str, Any] = cast("dict[str,Any]", self.condition_variables.as_dict())
270 else:
271 context_vars = {}
272 for template_format in template_formats:
273 context_vars[matching_ctx] = rendered
274 try:
275 template = self.context.hass_api.template(template_format)
276 rendered = template.async_render(variables=context_vars)
277 except TemplateError as e:
278 self.error_count += 1
279 _LOGGER.warning(
280 "SUPERNOTIFY Rendering template %s for %s failed: %s", template_field, self.delivery.name, e
281 )
282 return rendered
283 return original
285 # DupeCheckable implementation
287 def hash(self) -> int:
288 """Alpha hash to reduce noise from messages with timestamps or incrementing counts"""
290 def alphaize(v: str | None) -> str | None:
291 return v.translate(HASH_PREP_TRANSLATION_TABLE) if v else v
293 message: str | None = self._spoken_message() or self._message
294 return hash((alphaize(message), alphaize(self.delivery.name), self.target.hash_resolved(), alphaize(self._title)))
296 def _resolve_data_templates(self, data: dict[str, Any]) -> dict[str, Any]:
297 """Resolve Jinja2 templates in data dict for archive readability.
299 Returns a copy of data with template strings replaced by their
300 resolved values. Raw template string is preserved alongside as
301 <key>_template for debugging. Non-template values are unchanged.
302 """
303 if not data or not self.context:
304 return data
305 resolved: dict[str, Any] = {}
306 context_vars = cast("dict[str, Any]", self.condition_variables.as_dict()) if self.condition_variables else {}
307 for key, value in data.items():
308 if isinstance(value, str) and "{{" in value:
309 try:
310 rendered = self.context.hass_api.template(value).async_render(variables=context_vars)
311 resolved[key] = rendered
312 resolved[f"{key}_template"] = value
313 except Exception as e:
314 _LOGGER.debug("SUPERNOTIFY Could not resolve template for %s in %s: %s", key, self.delivery_name, e)
315 resolved[key] = value
316 else:
317 resolved[key] = value
318 return resolved