Coverage for custom_components/supernotify/transport.py: 99%
241 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +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, ServiceNotFound
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.schema import SelectionRank
29from custom_components.supernotify.target import Target, TargetEntityCategory
31from . import DOMAIN
32from .common import CallRecord
33from .const import (
34 ATTR_ENABLED,
35 ATTR_MEDIA_SNAPSHOT_URL,
36 CONF_DELIVERY_DEFAULTS,
37 INCLUSION_DEFAULT,
38 INCLUSION_EXPLICIT,
39)
40from .model import DeliveryConfig, SuppressionReason
41from .options import DeliveryOption
44def default_target_required(
45 unique: list[str | TargetEntityCategory],
46 fallback: list[str | TargetEntityCategory],
47 other: list[str | TargetEntityCategory],
48) -> TargetRequired:
49 """A transport that declares no target categories at all never has anything to resolve
50 a target from, so it can't need one; a transport that declares any does, by default."""
51 return TargetRequired.NEVER if not (unique or fallback or other) else TargetRequired.ALWAYS
54def default_selection_rank(
55 unique: list[str | TargetEntityCategory], fallback: list[str | TargetEntityCategory]
56) -> SelectionRank:
57 """A transport that can definitively claim a unique target category should be tried before
58 a fallback one; which is presumed to be either more broadly defined in terms of platforms
59 supported, or not as capable as the unique handler. Designed for Notify Entity."""
60 if unique:
61 return SelectionRank.FIRST
62 if fallback:
63 return SelectionRank.LAST
64 return SelectionRank.ANY
67def default_inclusion(unique: list[str | TargetEntityCategory], fallback: list[str | TargetEntityCategory]) -> list[str]:
68 """A transport that can never definitively claim anything (no unique/fallback category)
69 has no way to prove a given notification is relevant to it, so it requires explicit
70 enablement by config / scenario / action to be considered for delivery"""
71 return [INCLUSION_DEFAULT] if (unique or fallback) else [INCLUSION_EXPLICIT]
74if TYPE_CHECKING:
75 from anyio import Path
76 from homeassistant.helpers.typing import ConfigType
78 from .context import Context
79 from .delivery import Delivery, DeliveryRegistry
80 from .envelope import Envelope
81 from .hass_api import HomeAssistantAPI
82 from .people import PeopleRegistry
84# Markup that spoken transports hand straight to the voice assistant. Simplification
85# strips angle brackets, so SSML has to be passed through untouched or the assistant
86# ends up speaking the tag names out loud.
87RE_MARKUP_TAG = re.compile(r"(</?[A-Za-z][\w.:-]*(?:\s[^<>]*?)?/?>)")
88RE_MARKUP_TAG_NAME = re.compile(r"</?([A-Za-z][\w.:-]*)")
89SSML_TAG_NAMES = frozenset({
90 "alexa:name",
91 "amazon:domain",
92 "amazon:effect",
93 "amazon:emotion",
94 "audio",
95 "break",
96 "emphasis",
97 "lang",
98 "mark",
99 "phoneme",
100 "prosody",
101 "say-as",
102 "speak",
103 "sub",
104 "voice",
105})
107# Emoji prefix per priority, for a transport that offers one (medium: no prefix)
108PRIORITY_PREFIX: dict[str, str] = {
109 "critical": "\U0001f6a8 ", # police car light
110 "high": "⚠️ ", # warning sign
111 "low": "\U0001f539 ", # small blue diamond
112 "minimum": "\U0001f539 ", # small blue diamond
113}
115# Sign characters kept even though their Unicode category (Sm) would otherwise be stripped,
116# so numeric values like "+3" or "-3" aren't left indistinguishable from "3".
117SIGN_CHARS = frozenset("+-=%")
119_LOGGER = logging.getLogger(__name__)
122class Transport:
123 """Base class for delivery transports.
125 Sub classes integrste with Home Assistant notification services
126 or alternative notification mechanisms.
127 """
129 name: str
130 # the Home Assistant integration this transport depends on, if any - only used to judge
131 # how loudly to log when the transport isn't viable
132 integration_domain: ClassVar[str | None] = None
133 declared_options: ClassVar[list[DeliveryOption]] = []
134 # a transport sending only through an action the configuration names, with no way of finding one itself,
135 # whose standard delivery is the transport's own when its `delivery_defaults` has an action
136 action_configured: ClassVar[bool] = False
137 # whether it makes its own network connection, set up by its `connection` configuration
138 uses_connection: ClassVar[bool] = False
140 @abstractmethod
141 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
142 self.hass_api: HomeAssistantAPI = context.hass_api
143 self.people_registry: PeopleRegistry = context.people_registry
144 self.delivery_registry: DeliveryRegistry = context.delivery_registry
145 self.context: Context = context
146 transport_config = transport_config or {}
147 self.transport_config = TransportConfig(transport_config, class_config=self.default_config)
149 self.delivery_defaults: DeliveryConfig = self.transport_config.delivery_defaults
150 self.config_enabled = self.transport_config.enabled
151 self.enabled = self.config_enabled
152 self.alias = self.transport_config.alias
153 self.last_error_at: dt.datetime | None = None
154 self.last_error_in: str | None = None
155 self.last_error_message: str | None = None
156 self.error_count: int = 0
157 self._unavailable: bool = False
159 async def initialize(self) -> None:
160 """Async post-construction initialization"""
161 if self.name is None:
162 raise IntegrationError("Invalid nameless transport adaptor subclass")
164 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]:
165 return {}
167 @property
168 def supported_features(self) -> TransportFeature:
169 return TransportFeature.MESSAGE | TransportFeature.TITLE
171 @property
172 def targets(self) -> Target:
173 return self.delivery_defaults.target if self.delivery_defaults.target is not None else Target()
175 @property
176 def unique_target_categories(self) -> list[str | TargetEntityCategory]:
177 """Target categories nothing else could claim - a value shape/prefix unique to this
178 transport (e.g. `ATTR_EMAIL`), or an entity domain+platform combination unique to it.
180 This is what drives implicit selection (`Notification.select_deliveries()`): a
181 delivery whose transport declares one of these, and whose category is actually
182 present in the notification's target or an available recipient's own capabilities,
183 becomes an auto-selection candidate - the same way giving an email address has always
184 implied the `email` delivery. Empty by default; a transport with nothing here is
185 never implicitly selected on its own, only by an explicit call/scenario/recipient ask.
186 """
187 return []
189 @property
190 def fallback_target_categories(self) -> list[str | TargetEntityCategory]:
191 """Target categories this transport will claim once nothing more specific did -
192 e.g. `notify_entity`'s bare `notify.*` domain match, after html5/alexa_devices' own
193 platform-scoped (and therefore `unique`) matches on the same domain get first refusal.
195 Also drives implicit selection, like `unique_target_categories`, but ranked to lose
196 selection-order ties to a delivery selected via a `unique` match (see
197 `default_selection_rank()`).
198 """
199 return []
201 @property
202 def other_target_categories(self) -> list[str | TargetEntityCategory]:
203 """Target categories this transport understands well enough to build a full envelope
204 from, but that are too ambiguous to drive auto-selection on their own - several
205 transports competing for the same bare entity domain (`media_player`, matched by
206 kodi/tts/the `media` transport/alexa_media_player/chime), where an entity alone
207 doesn't say which one of them was actually meant.
208 """
209 return []
211 @property
212 def target_categories(self) -> list[str | TargetEntityCategory]:
213 """The target categories this transport understands, independent of any delivery -
214 the union of `unique_target_categories`, `fallback_target_categories` and
215 `other_target_categories`, for envelope-building (`Delivery.target_categories`/
216 `Delivery.select_targets()`). Selection itself only ever looks at the first two.
217 """
218 return [*self.unique_target_categories, *self.fallback_target_categories, *self.other_target_categories]
220 @property
221 def default_config(self) -> TransportConfig:
222 config = TransportConfig()
223 config.delivery_defaults.target_required = default_target_required(
224 self.unique_target_categories, self.fallback_target_categories, self.other_target_categories
225 )
226 config.delivery_defaults.selection_rank = default_selection_rank(
227 self.unique_target_categories, self.fallback_target_categories
228 )
229 config.delivery_defaults.inclusion = default_inclusion(self.unique_target_categories, self.fallback_target_categories)
230 return config
232 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
233 """Whether this transport currently has what it needs to auto-configure a delivery.
235 Always True for an `action_configured` transport: an explicit delivery can supply its own action
236 regardless of whether the service is discoverable here, and is_viable() can't see delivery-level
237 config, so it can't rule that out - DeliveryRegistry prunes the transport entirely once it's
238 confirmed no delivery (explicit or auto) uses it.
240 Default implementation just defers to `build_standard_deliveries()` and checks for
241 a non-empty result - correct for any transport, but builds (and discards) the
242 `DeliveryConfig`s to answer what's otherwise a yes/no question. Override with a
243 standalone check (matching `build_standard_deliveries()`'s own condition) in a
244 transport where that's cheap and doesn't require mutating `self.delivery_defaults`
245 to find out - most transports that gate purely on hass_api state (a config entry, a
246 registered service, discovered entities) can. Skip the override where viability can
247 only be discovered by doing the same service/entity lookup
248 `build_standard_deliveries()` itself needs to build the config (e.g. `discord`,
249 `pushover`, `sms` - discovering *which* service is available - or `email`, which
250 also decides *how* to send based on what's found).
251 """
252 return self.action_configured or bool(self.build_standard_deliveries(hass_api))
254 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
255 """Build every 'standard' (auto-generatable) delivery this transport contributes,
256 keyed by name: its own default (keyed by `self.name`) plus any extras.
258 Only ever called once `is_viable()` has returned True for the same `hass_api` -
259 callers must check that first. Most overrides trust this and skip re-checking
260 their own viability condition; the exception is a transport whose viability can
261 only be discovered by doing the very lookup this method needs anyway (see
262 `is_viable()`'s docstring) - those keep their own guard and still return an empty
263 dict, simply because there's nothing to gain by trusting the caller there.
264 """
265 if self.action_configured and self.delivery_defaults.action:
266 return {self.name: {}}
267 return {}
269 def validate_action(self, action: str | None) -> bool:
270 """Override in subclass if transport has fixed action or doesn't require one"""
271 return action == self.delivery_defaults.action
273 def priority_prefix(self, envelope: Envelope) -> str:
274 """Emoji to start a message with for its priority, or nothing for medium"""
275 return PRIORITY_PREFIX.get(envelope.priority or "medium", "")
277 def validate_notify_action(self, action: str | None) -> bool:
278 """For a transport sending through an integration's `notify.<name>` action, named as the user chose"""
279 if action and action.startswith("notify.") and action.split(".", 1)[1]:
280 return True
281 _LOGGER.warning("SUPERNOTIFY %s: action must be a notify.* service, got: %r", self.name, action)
282 return False
284 async def safe_grab_image(self, envelope: Envelope) -> Path | None:
285 """The notification's image, or None, a failure to get one not stopping the delivery"""
286 try:
287 return await envelope.grab_image()
288 except Exception as e:
289 _LOGGER.warning("SUPERNOTIFY %s: failed to grab image: %s", self.name, e)
290 return None
292 def absolute_url(self, url: str) -> str | None:
293 """A URL the recipient's device can fetch, for one that may be relative to Home Assistant"""
294 return self.hass_api.abs_url(url)
296 async def image_url(self, envelope: Envelope, grab: bool = True) -> str | None:
297 """A URL the recipient's device can fetch the notification's image from - its `snapshot_url`,
298 or else, with `grab`, an image grabbed from its camera and served from the media folder"""
299 snapshot_url: str | None = (envelope.media or {}).get(ATTR_MEDIA_SNAPSHOT_URL)
300 if snapshot_url:
301 return self.absolute_url(str(snapshot_url))
302 if not grab:
303 return None
304 image_path: Path | None = await self.safe_grab_image(envelope)
305 if image_path is None:
306 return None
307 try:
308 return await self.context.media_storage.object_url(image_path)
309 except Exception as e:
310 _LOGGER.debug("SUPERNOTIFY %s: no url for %s: %s", self.name, image_path, e)
311 return None
313 def attributes(self) -> dict[str, Any]:
314 attrs: dict[str, Any] = {
315 ATTR_NAME: self.name,
316 ATTR_ENABLED: self.enabled,
317 CONF_DELIVERY_DEFAULTS: self.delivery_defaults,
318 }
319 if self.alias:
320 attrs[ATTR_FRIENDLY_NAME] = self.alias
321 if self.last_error_at:
322 attrs["last_error_at"] = self.last_error_at
323 attrs["last_error_in"] = self.last_error_in
324 attrs["last_error_message"] = self.last_error_message
325 attrs["error_count"] = self.error_count
326 attrs.update(self.extra_attributes())
327 return attrs
329 def extra_attributes(self) -> dict[str, Any]:
330 return {}
332 @abstractmethod
333 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
334 """Delivery implementation
336 Args:
337 ----
338 envelope (Envelope): envelope to be delivered
339 debug_trace (DebugTrace): debug info collector
341 """
343 async def common_string(self, key: str) -> str:
344 """Text for the recipient from the `common` translations, in the Home Assistant language"""
345 strings: dict[str, str] = await self.hass_api.translations("common")
346 return strings[f"component.{DOMAIN}.common.{key}"]
348 def set_action_data(self, action_data: dict[str, Any], key: str, data: Any | None) -> dict[str, Any]: # ruff: ignore[any-type]
349 if data is not None:
350 action_data[key] = data
351 return action_data
353 async def call_action(
354 self,
355 envelope: Envelope,
356 qualified_action: str | None = None,
357 action_data: dict[str, Any] | None = None,
358 target_data: dict[str, Any] | None = None,
359 implied_target: bool = False, # True if the qualified action implies a target
360 single_target: bool = False, # True if the action reaches only one target, so not finding it isn't an error
361 ) -> bool:
362 action_data = action_data or {}
363 start_time = time.time()
364 domain = service = None
365 delivery: Delivery = envelope.delivery
366 try:
367 qualified_action = qualified_action or delivery.action
368 if not qualified_action:
369 _LOGGER.debug(
370 "SUPERNOTIFY Skipping %s action call with no service, targets %s",
371 envelope.delivery.name,
372 action_data.get(ATTR_TARGET),
373 )
374 envelope.skipped = 1
375 envelope.skip_reason = SuppressionReason.NO_ACTION
376 return False
377 if (
378 delivery.target_required == TargetRequired.ALWAYS
379 and not action_data.get(ATTR_TARGET)
380 and not action_data.get(ATTR_ENTITY_ID)
381 and not implied_target
382 and not target_data
383 ):
384 _LOGGER.debug(
385 "SUPERNOTIFY Skipping %s action call for service %s, missing targets",
386 envelope.delivery.name,
387 qualified_action,
388 )
389 envelope.skipped = 1
390 envelope.skip_reason = SuppressionReason.NO_TARGET
391 return False
393 domain, service = qualified_action.split(".", 1)
394 # a delivery with a fallback, or used as one, waits for its call, so that an error in the
395 # service is seen and the next one can be tried - otherwise only a missing service or bad
396 # data would be
397 wait: dict[str, Any] = {"blocking": True} if (delivery.fallback or delivery.is_fallback) else {}
398 start_time = time.time()
399 timestamp: dt.datetime | None = None
400 if target_data:
401 # home-assistant messes with the service_data passed by ref
402 service_data_as_sent = dict(action_data)
403 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
404 service_response = await self.hass_api.call_service(
405 domain,
406 service,
407 service_data=action_data,
408 target=target_data,
409 debug=delivery.debug,
410 context=envelope.ha_context,
411 **wait,
412 )
413 envelope.calls.append(
414 CallRecord(
415 timestamp,
416 time.time() - start_time,
417 domain,
418 service,
419 debug=delivery.debug,
420 action_data=service_data_as_sent,
421 target_data=target_data,
422 service_response=service_response,
423 )
424 )
425 else:
426 service_data_as_sent = dict(action_data)
427 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
428 service_response = await self.hass_api.call_service(
429 domain, service, service_data=action_data, debug=delivery.debug, context=envelope.ha_context, **wait
430 )
431 envelope.calls.append(
432 CallRecord(
433 timestamp,
434 time.time() - start_time,
435 domain,
436 service,
437 debug=delivery.debug,
438 action_data=service_data_as_sent,
439 service_response=service_response,
440 )
441 )
443 envelope.delivered = 1
444 self.log_delivery_recovered()
445 return True
446 except Exception as e:
447 not_found: bool = isinstance(e, ServiceNotFound)
448 envelope.failed_calls.append(
449 CallRecord(
450 timestamp,
451 time.time() - start_time,
452 domain,
453 service,
454 action_data,
455 target_data,
456 exception=str(e),
457 )
458 )
459 if not_found and single_target:
460 # like a rejected e-mail address, one target that couldn't be reached rather than
461 # the transport failing
462 _LOGGER.warning("SUPERNOTIFY Missed %s target, %s, id: %s", self.name, e, envelope.notification_id)
463 return False
464 self.record_error(str(e), method="call_action")
465 self.log_delivery_failure(
466 e, "SUPERNOTIFY Failed to notify %s via %s.%s, id: %s", self.name, domain, service, envelope.notification_id
467 )
468 envelope.error_count += 1
469 # a missing action explains itself, only an unexpected error needs its stack trace
470 envelope.delivery_error = [str(e)] if not_found else format_exception(e)
471 return False
473 def record_error(self, message: str, method: str) -> None:
474 self.last_error_at = dt_util.utcnow()
475 self.last_error_message = message
476 self.last_error_in = method
477 self.error_count += 1
479 def log_delivery_failure(self, err: BaseException, message: str, *args: Any) -> None:
480 """Log a delivery failure, passing the exception caught in the caller's except block.
482 Logged at ERROR (with traceback, unless it's only a missing action) the first time this transport becomes unavailable,
483 then downgraded to DEBUG for consecutive failures until it recovers - avoids
484 spamming the log every notification while an external service/device stays down.
485 Call alongside record_error(), which keeps tracking the lifetime error count
486 regardless of log level.
487 """
488 exc_info: BaseException | None = None if isinstance(err, ServiceNotFound) else err
489 if self._unavailable:
490 _LOGGER.debug(message, *args, exc_info=exc_info)
491 else:
492 _LOGGER.error(message, *args, exc_info=exc_info)
493 self._unavailable = True
495 def log_delivery_recovered(self) -> None:
496 """Call on a successful delivery - logs once if this transport was previously
497 flagged unavailable, then clears the flag."""
498 if self._unavailable:
499 _LOGGER.info("SUPERNOTIFY %s transport recovered after prior delivery failures", self.name)
500 self._unavailable = False
502 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None:
503 """Simplify text for delivery transports with speaking or plain text interfaces.
505 Spoken transports can be handed SSML, which the voice assistant parses itself.
506 Simplification removes angle brackets, so applying it to SSML turns the markup
507 into words the assistant reads out loud. When a spoken transport is given SSML,
508 the tags are left alone and only the text around them is simplified, so emoji,
509 URLs and symbols are still cleaned up.
510 """
511 if not text:
512 return None
513 if self.supported_features & TransportFeature.SPOKEN and self._is_ssml(text):
514 simplified = "".join(
515 fragment if index % 2 else self._simplify_around_markup(fragment, strip_urls)
516 for index, fragment in enumerate(RE_MARKUP_TAG.split(text))
517 )
518 else:
519 simplified = self._simplify_text(text, strip_urls)
520 _LOGGER.debug("SUPERNOTIFY Simplified text to: %s", simplified)
521 return simplified
523 @staticmethod
524 def _is_ssml(text: str) -> bool:
525 """Tell SSML markup apart from stray angle brackets in ordinary text."""
526 for tag in RE_MARKUP_TAG.findall(text):
527 name = RE_MARKUP_TAG_NAME.match(tag)
528 if name is not None and name.group(1).lower() in SSML_TAG_NAMES:
529 return True
530 return False
532 @staticmethod
533 def _strip_urls(text: str) -> str:
534 return " ".join(word for word in text.split() if not (urlparse(word).scheme and urlparse(word).netloc))
536 @classmethod
537 def _simplify_text(cls, text: str, strip_urls: bool = False) -> str:
538 """Remove symbols, and optionally URLs, that can trip up voice assistants."""
539 if strip_urls:
540 text = cls._strip_urls(text)
541 text = unicodedata.normalize("NFC", text)
542 text = text.translate(str.maketrans("_", " ", "()£$<>"))
543 return "".join(c for c in text if c in SIGN_CHARS or unicodedata.category(c) not in ("So", "Sk", "Sm", "Mn", "Sc"))
545 @classmethod
546 def _simplify_around_markup(cls, fragment: str, strip_urls: bool) -> str:
547 """Simplify a fragment of text sitting between two SSML tags, keeping the
548 whitespace at either end so that words do not end up glued to the markup."""
549 if not fragment.strip():
550 return fragment
551 lead = fragment[: len(fragment) - len(fragment.lstrip())]
552 trail = fragment[len(fragment.rstrip()) :]
553 return f"{lead}{cls._simplify_text(fragment.strip(), strip_urls)}{trail}"