Coverage for custom_components/supernotify/transport.py: 98%
173 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +0000
1from __future__ import annotations
3import datetime as dt
4import logging
5import re
6import time
7import unicodedata
8from abc import abstractmethod
9from traceback import format_exception
10from typing import TYPE_CHECKING, Any, ClassVar
11from urllib.parse import urlparse
13from homeassistant.components.notify.const import ATTR_TARGET
14from homeassistant.const import (
15 ATTR_ENTITY_ID,
16 ATTR_FRIENDLY_NAME,
17 ATTR_NAME,
18)
19from homeassistant.exceptions import IntegrationError
20from homeassistant.util import dt as dt_util
22from custom_components.supernotify.model import (
23 DebugTrace,
24 TargetRequired,
25 TransportConfig,
26 TransportFeature,
27)
28from custom_components.supernotify.target import Target, TargetEntityCategory
30from .common import CallRecord
31from .const import (
32 ATTR_ENABLED,
33 CONF_DELIVERY_DEFAULTS,
34 INCLUSION_EXPLICIT,
35)
36from .model import DeliveryConfig, SuppressionReason
37from .options import DeliveryOption
39if TYPE_CHECKING:
40 from homeassistant.helpers.typing import ConfigType
42 from .context import Context
43 from .delivery import Delivery, DeliveryRegistry
44 from .hass_api import HomeAssistantAPI
45 from .people import PeopleRegistry
47# Markup that spoken transports hand straight to the voice assistant. Simplification
48# strips angle brackets, so SSML has to be passed through untouched or the assistant
49# ends up speaking the tag names out loud.
50RE_MARKUP_TAG = re.compile(r"(</?[A-Za-z][\w.:-]*(?:\s[^<>]*?)?/?>)")
51RE_MARKUP_TAG_NAME = re.compile(r"</?([A-Za-z][\w.:-]*)")
52SSML_TAG_NAMES = frozenset({
53 "alexa:name",
54 "amazon:domain",
55 "amazon:effect",
56 "amazon:emotion",
57 "audio",
58 "break",
59 "emphasis",
60 "lang",
61 "mark",
62 "phoneme",
63 "prosody",
64 "say-as",
65 "speak",
66 "sub",
67 "voice",
68})
70# Sign characters kept even though their Unicode category (Sm) would otherwise be stripped,
71# so numeric values like "+3" or "-3" aren't left indistinguishable from "3".
72SIGN_CHARS = frozenset("+-=%")
74_LOGGER = logging.getLogger(__name__)
77class Transport:
78 """Base class for delivery transports.
80 Sub classes integrste with Home Assistant notification services
81 or alternative notification mechanisms.
82 """
84 name: str
85 declared_options: ClassVar[list[DeliveryOption]] = []
87 @abstractmethod
88 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
89 self.hass_api: HomeAssistantAPI = context.hass_api
90 self.people_registry: PeopleRegistry = context.people_registry
91 self.delivery_registry: DeliveryRegistry = context.delivery_registry
92 self.context: Context = context
93 transport_config = transport_config or {}
94 self.transport_config = TransportConfig(transport_config, class_config=self.default_config)
96 self.delivery_defaults: DeliveryConfig = self.transport_config.delivery_defaults
97 self.config_enabled = self.transport_config.enabled
98 self.enabled = self.config_enabled
99 self.alias = self.transport_config.alias
100 self.last_error_at: dt.datetime | None = None
101 self.last_error_in: str | None = None
102 self.last_error_message: str | None = None
103 self.error_count: int = 0
104 self._unavailable: bool = False
106 async def initialize(self) -> None:
107 """Async post-construction initialization"""
108 if self.name is None:
109 raise IntegrationError("Invalid nameless transport adaptor subclass")
111 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]:
112 return {}
114 @property
115 def supported_features(self) -> TransportFeature:
116 return TransportFeature.MESSAGE | TransportFeature.TITLE
118 @property
119 def targets(self) -> Target:
120 return self.delivery_defaults.target if self.delivery_defaults.target is not None else Target()
122 @property
123 def target_categories(self) -> list[str | TargetEntityCategory]:
124 """The target categories this transport understands, independent of any delivery.
126 A plain string names a category directly (e.g. `ATTR_EMAIL`); an `TargetEntityCategory`
127 declares that the `entity_id` category is accepted, but only for entities matching
128 its domain/platform constraints. Empty by default - a transport that doesn't declare
129 anything here relies entirely on `Delivery.select_targets()`'s other qualification
130 paths (its own name, its transport's name, or a delivery's own `OPTION_TARGET_CATEGORIES`
131 override), which is the deliberate design for `generic`, a bring-your-own-categories
132 transport. Queried via `Delivery.target_categories`, not directly - a `Transport`
133 never needs to know about delivery-level config, only the reverse.
134 """
135 return []
137 @property
138 def default_config(self) -> TransportConfig:
139 return TransportConfig()
141 @property
142 def inclusion_mode(self) -> list[str]:
143 """The `inclusion` an auto-configured delivery for this transport should use.
145 Explicit-only by default: most transports need a chat_id/channel/device_id the
146 notification author must supply, have targets too opaque or ambiguous to map to
147 a recipient/entity, or a channel too intrusive to fire on every notification.
148 Override to return `[INCLUSION_DEFAULT]` for the few transports that can
149 reasonably fire on every notification out of the box (e.g. email, mobile_push).
151 Pulled out as a separate property so can be reported in the Transport Configuration
152 section of the Developer documentation
153 """
154 return [INCLUSION_EXPLICIT]
156 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
157 """Whether this transport currently has what it needs to auto-configure a delivery.
159 Default implementation just defers to `build_standard_deliveries()` and checks for
160 a non-empty result - correct for any transport, but builds (and discards) the
161 `DeliveryConfig`s to answer what's otherwise a yes/no question. Override with a
162 standalone check (matching `build_standard_deliveries()`'s own condition) in a
163 transport where that's cheap and doesn't require mutating `self.delivery_defaults`
164 to find out - most transports that gate purely on hass_api state (a config entry, a
165 registered service, discovered entities) can. Skip the override where viability can
166 only be discovered by doing the same service/entity lookup
167 `build_standard_deliveries()` itself needs to build the config (e.g. `discord`,
168 `pushover`, `sms` - discovering *which* service is available - or `email`, which
169 also decides *how* to send based on what's found).
170 """
171 return bool(self.build_standard_deliveries(hass_api))
173 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
174 """Build every 'standard' (auto-generatable) delivery this transport contributes,
175 keyed by name: its own default (keyed by `self.name`) plus any extras.
177 Only ever called once `is_viable()` has returned True for the same `hass_api` -
178 callers must check that first. Most overrides trust this and skip re-checking
179 their own viability condition; the exception is a transport whose viability can
180 only be discovered by doing the very lookup this method needs anyway (see
181 `is_viable()`'s docstring) - those keep their own guard and still return an empty
182 dict, simply because there's nothing to gain by trusting the caller there.
183 """
184 return {}
186 def validate_action(self, action: str | None) -> bool:
187 """Override in subclass if transport has fixed action or doesn't require one"""
188 return action == self.delivery_defaults.action
190 def attributes(self) -> dict[str, Any]:
191 attrs: dict[str, Any] = {
192 ATTR_NAME: self.name,
193 ATTR_ENABLED: self.enabled,
194 CONF_DELIVERY_DEFAULTS: self.delivery_defaults,
195 }
196 if self.alias:
197 attrs[ATTR_FRIENDLY_NAME] = self.alias
198 if self.last_error_at:
199 attrs["last_error_at"] = self.last_error_at
200 attrs["last_error_in"] = self.last_error_in
201 attrs["last_error_message"] = self.last_error_message
202 attrs["error_count"] = self.error_count
203 attrs.update(self.extra_attributes())
204 return attrs
206 def extra_attributes(self) -> dict[str, Any]:
207 return {}
209 @abstractmethod
210 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # type: ignore # noqa: F821
211 """Delivery implementation
213 Args:
214 ----
215 envelope (Envelope): envelope to be delivered
216 debug_trace (DebugTrace): debug info collector
218 """
220 def set_action_data(self, action_data: dict[str, Any], key: str, data: Any | None) -> dict[str, Any]: # ruff: ignore[any-type]
221 if data is not None:
222 action_data[key] = data
223 return action_data
225 async def call_action(
226 self,
227 envelope: Envelope, # type: ignore # noqa: F821
228 qualified_action: str | None = None,
229 action_data: dict[str, Any] | None = None,
230 target_data: dict[str, Any] | None = None,
231 implied_target: bool = False, # True if the qualified action implies a target
232 ) -> bool:
233 action_data = action_data or {}
234 start_time = time.time()
235 domain = service = None
236 delivery: Delivery = envelope.delivery
237 try:
238 qualified_action = qualified_action or delivery.action
239 if not qualified_action:
240 _LOGGER.debug(
241 "SUPERNOTIFY Skipping %s action call with no service, targets %s",
242 envelope.delivery.name,
243 action_data.get(ATTR_TARGET),
244 )
245 envelope.skipped = 1
246 envelope.skip_reason = SuppressionReason.NO_ACTION
247 return False
248 if (
249 delivery.target_required == TargetRequired.ALWAYS
250 and not action_data.get(ATTR_TARGET)
251 and not action_data.get(ATTR_ENTITY_ID)
252 and not implied_target
253 and not target_data
254 ):
255 _LOGGER.debug(
256 "SUPERNOTIFY Skipping %s action call for service %s, missing targets",
257 envelope.delivery.name,
258 qualified_action,
259 )
260 envelope.skipped = 1
261 envelope.skip_reason = SuppressionReason.NO_TARGET
262 return False
264 domain, service = qualified_action.split(".", 1)
265 start_time = time.time()
266 timestamp: dt.datetime | None = None
267 if target_data:
268 # home-assistant messes with the service_data passed by ref
269 service_data_as_sent = dict(action_data)
270 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
271 service_response = await self.hass_api.call_service(
272 domain,
273 service,
274 service_data=action_data,
275 target=target_data,
276 debug=delivery.debug,
277 context=envelope.ha_context,
278 )
279 envelope.calls.append(
280 CallRecord(
281 timestamp,
282 time.time() - start_time,
283 domain,
284 service,
285 debug=delivery.debug,
286 action_data=service_data_as_sent,
287 target_data=target_data,
288 service_response=service_response,
289 )
290 )
291 else:
292 service_data_as_sent = dict(action_data)
293 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
294 service_response = await self.hass_api.call_service(
295 domain, service, service_data=action_data, debug=delivery.debug, context=envelope.ha_context
296 )
297 envelope.calls.append(
298 CallRecord(
299 timestamp,
300 time.time() - start_time,
301 domain,
302 service,
303 debug=delivery.debug,
304 action_data=service_data_as_sent,
305 service_response=service_response,
306 )
307 )
309 envelope.delivered = 1
310 self.log_delivery_recovered()
311 return True
312 except Exception as e:
313 self.record_error(str(e), method="call_action")
314 envelope.failed_calls.append(
315 CallRecord(
316 timestamp,
317 time.time() - start_time,
318 domain,
319 service,
320 action_data,
321 target_data,
322 exception=str(e),
323 )
324 )
325 self.log_delivery_failure(
326 e, "SUPERNOTIFY Failed to notify %s via %s.%s, id: %s", self.name, domain, service, envelope.notification_id
327 )
328 envelope.error_count += 1
329 envelope.delivery_error = format_exception(e)
330 return False
332 def record_error(self, message: str, method: str) -> None:
333 self.last_error_at = dt_util.utcnow()
334 self.last_error_message = message
335 self.last_error_in = method
336 self.error_count += 1
338 def log_delivery_failure(self, err: BaseException, message: str, *args: Any) -> None:
339 """Log a delivery failure, passing the exception caught in the caller's except block.
341 Logged at ERROR (with traceback) the first time this transport becomes unavailable,
342 then downgraded to DEBUG for consecutive failures until it recovers - avoids
343 spamming the log every notification while an external service/device stays down.
344 Call alongside record_error(), which keeps tracking the lifetime error count
345 regardless of log level.
346 """
347 if self._unavailable:
348 _LOGGER.debug(message, *args, exc_info=err)
349 else:
350 _LOGGER.error(message, *args, exc_info=err)
351 self._unavailable = True
353 def log_delivery_recovered(self) -> None:
354 """Call on a successful delivery - logs once if this transport was previously
355 flagged unavailable, then clears the flag."""
356 if self._unavailable:
357 _LOGGER.info("SUPERNOTIFY %s transport recovered after prior delivery failures", self.name)
358 self._unavailable = False
360 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None:
361 """Simplify text for delivery transports with speaking or plain text interfaces.
363 Spoken transports can be handed SSML, which the voice assistant parses itself.
364 Simplification removes angle brackets, so applying it to SSML turns the markup
365 into words the assistant reads out loud. When a spoken transport is given SSML,
366 the tags are left alone and only the text around them is simplified, so emoji,
367 URLs and symbols are still cleaned up.
368 """
369 if not text:
370 return None
371 if self.supported_features & TransportFeature.SPOKEN and self._is_ssml(text):
372 simplified = "".join(
373 fragment if index % 2 else self._simplify_around_markup(fragment, strip_urls)
374 for index, fragment in enumerate(RE_MARKUP_TAG.split(text))
375 )
376 else:
377 simplified = self._simplify_text(text, strip_urls)
378 _LOGGER.debug("SUPERNOTIFY Simplified text to: %s", simplified)
379 return simplified
381 @staticmethod
382 def _is_ssml(text: str) -> bool:
383 """Tell SSML markup apart from stray angle brackets in ordinary text."""
384 for tag in RE_MARKUP_TAG.findall(text):
385 name = RE_MARKUP_TAG_NAME.match(tag)
386 if name is not None and name.group(1).lower() in SSML_TAG_NAMES:
387 return True
388 return False
390 @staticmethod
391 def _simplify_text(text: str, strip_urls: bool = False) -> str:
392 """Remove symbols, and optionally URLs, that can trip up voice assistants."""
393 if strip_urls:
394 words = text.split()
395 text = " ".join(word for word in words if not (urlparse(word).scheme and urlparse(word).netloc))
396 text = unicodedata.normalize("NFC", text)
397 text = text.translate(str.maketrans("_", " ", "()£$<>"))
398 return "".join(c for c in text if c in SIGN_CHARS or unicodedata.category(c) not in ("So", "Sk", "Sm", "Mn", "Sc"))
400 @classmethod
401 def _simplify_around_markup(cls, fragment: str, strip_urls: bool) -> str:
402 """Simplify a fragment of text sitting between two SSML tags, keeping the
403 whitespace at either end so that words do not end up glued to the markup."""
404 if not fragment.strip():
405 return fragment
406 lead = fragment[: len(fragment) - len(fragment.lstrip())]
407 trail = fragment[len(fragment.rstrip()) :]
408 return f"{lead}{cls._simplify_text(fragment.strip(), strip_urls)}{trail}"