Coverage for custom_components/supernotify/transports/matrix.py: 100%
83 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1"""Matrix transport for SuperNotify.
3Sends messages to Matrix rooms using Home Assistant's `matrix` integration,
4calling the native `matrix.send_message` service (not the thin legacy notify
5wrapper), for granular control over format, images and threads.
7Supported data keys (all optional):
8 matrix_format str "text" | "html" (default: "html" when a
9 title is present, otherwise "text")
10 matrix_thread_id str Send the message into a Matrix thread
11 matrix_attach_image bool Attach camera snapshot (default: False)
12 matrix_priority_prefix bool Prefix message with an emoji derived from
13 the SuperNotify priority (default: False):
14 critical=siren, high=warning,
15 low/minimum=small diamond, medium=none
17Notes on the HA `matrix.send_message` service schema:
18- The `data` sub-dict is STRICT (no ALLOW_EXTRA): only `format`, `images`
19 and `thread_id` are accepted, any other key makes the whole call fail
20 with `vol.Invalid`. Residual generic data keys are therefore NOT merged
21 into the payload (they are dropped with a debug log), unlike the standard
22 transport pattern.
23- `target` is required and every entry must match the room regex
24 `^[!|#][^:]*:.*` (room ID `!abc:server` or alias `#name:server`). A single
25 invalid entry rejects the whole call, so targets are pre-filtered here and
26 invalid ones are dropped with a debug log. Prefer room IDs, or aliases
27 listed in the `rooms:` config of the matrix integration.
28- There is no `title` field: the title is composed into the message body
29 (bold + line break for html, plain line break for text).
30- With `format: html` the core sets both `formatted_body` and plain `body`
31 to the same string (no markup strip), so clients without HTML support
32 will show raw tags. This mirrors core behaviour and is accepted.
33- `images` is a list of LOCAL paths and the core checks
34 `hass.config.is_allowed_path()`: the SuperNotify media path must be listed
35 in `homeassistant.allowlist_external_dirs`, otherwise the image is dropped
36 by the integration (the text message is still sent first).
37- Matrix has no native message priority: the only priority mapping offered
38 is the opt-in emoji prefix above.
39"""
41from __future__ import annotations
43import html
44import logging
45import re
46from typing import TYPE_CHECKING, Any
48from custom_components.supernotify.common import boolify
49from custom_components.supernotify.const import ATTR_DATA, TRANSPORT_MATRIX
50from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature
51from custom_components.supernotify.transport import Transport
53if TYPE_CHECKING:
54 from custom_components.supernotify.envelope import Envelope
56_LOGGER = logging.getLogger(__name__)
58# Slightly stricter subset of the core service regex `^[!|#][^:]*:.*`:
59# require a room ID (!) or alias (#) sigil and a non-empty server part.
60_MATRIX_ROOM_RE = re.compile(r"^[!#][^:]*:.+")
62_VALID_FORMATS = ("text", "html")
64# Opt-in emoji prefix per SuperNotify priority (medium: no prefix)
65_PRIORITY_PREFIX = {
66 "critical": "\U0001f6a8 ", # police car light
67 "high": "⚠️ ", # warning sign
68 "low": "\U0001f539 ", # small blue diamond
69 "minimum": "\U0001f539 ", # small blue diamond
70}
73class MatrixTransport(Transport):
74 """Notify via Matrix rooms using Home Assistant matrix integration."""
76 def __init__(self, *args: Any, **kwargs: Any) -> None:
77 super().__init__(*args, **kwargs)
79 name = TRANSPORT_MATRIX
81 @property
82 def supported_features(self) -> TransportFeature:
83 return TransportFeature.MESSAGE | TransportFeature.TITLE | TransportFeature.IMAGES | TransportFeature.SNAPSHOT_IMAGE
85 @property
86 def default_config(self) -> TransportConfig:
87 config = TransportConfig()
88 config.delivery_defaults.action = "matrix.send_message"
89 config.delivery_defaults.target_required = TargetRequired.ALWAYS
90 return config
92 def validate_action(self, action: str | None) -> bool:
93 """Validate that action is the matrix send_message service."""
94 return action == "matrix.send_message"
96 def select_rooms(self, envelope: Envelope) -> list[str]:
97 """Filter envelope targets down to valid Matrix room IDs or aliases.
99 The service rejects the whole call if any target fails the room
100 regex, so invalid entries are dropped here (with a debug log) instead
101 of being forwarded. Duplicates are removed preserving order.
102 """
103 raw_targets: list[str] = envelope.target.resolved_targets() if envelope.target else []
104 rooms: list[str] = []
105 for target in raw_targets:
106 if isinstance(target, str) and _MATRIX_ROOM_RE.match(target):
107 if target not in rooms:
108 rooms.append(target)
109 else:
110 _LOGGER.debug("SUPERNOTIFY matrix: skipping invalid room target %r", target)
111 return rooms
113 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
114 _LOGGER.debug("SUPERNOTIFY matrix %s", envelope.message)
116 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}
118 # Pop Matrix-specific data keys
119 format_override = raw_data.pop("matrix_format", None)
120 thread_id = raw_data.pop("matrix_thread_id", None)
121 attach_image = boolify(raw_data.pop("matrix_attach_image", False), default=False)
122 priority_prefix = boolify(raw_data.pop("matrix_priority_prefix", False), default=False)
124 # Resolve and pre-validate room targets
125 rooms = self.select_rooms(envelope)
126 if not rooms:
127 _LOGGER.warning("SUPERNOTIFY matrix: no valid room targets (expected !room:server or #alias:server)")
128 self.record_error("no valid Matrix room targets", "deliver")
129 return False
131 # Resolve format: explicit override, or html when a title must be
132 # composed in bold, plain text otherwise
133 default_format = "html" if envelope.title else "text"
134 if format_override:
135 fmt = str(format_override).lower()
136 if fmt not in _VALID_FORMATS:
137 _LOGGER.warning("SUPERNOTIFY matrix: invalid matrix_format '%s', using '%s'", format_override, default_format)
138 fmt = default_format
139 else:
140 fmt = default_format
142 # Compose title into the message body (the service has no title field).
143 # Only the title is escaped in html mode: the body may already contain
144 # HTML the user wrote intentionally (consistent with core behaviour).
145 message_text = envelope.message or ""
146 if envelope.title:
147 if fmt == "html":
148 message_text = f"<b>{html.escape(envelope.title)}</b><br>{message_text}"
149 else:
150 message_text = f"{envelope.title}\n{message_text}"
152 # Opt-in emoji prefix mapped from SuperNotify priority
153 if priority_prefix:
154 message_text = _PRIORITY_PREFIX.get(envelope.priority or "medium", "") + message_text
156 # Grab camera snapshot if requested; images are local paths and the
157 # matrix integration checks them against allowlist_external_dirs
158 images: list[str] = []
159 if attach_image:
160 image_path = None
161 try:
162 image_path = await envelope.grab_image()
163 except Exception as e:
164 _LOGGER.warning("SUPERNOTIFY matrix: failed to grab image: %s", e)
165 if image_path:
166 images.append(str(image_path))
167 else:
168 _LOGGER.debug("SUPERNOTIFY matrix: no image available, sending text only")
170 # Build the payload. The service data sub-dict is whitelist-only
171 # (format / images / thread_id): residual generic data keys are NOT
172 # merged, they would fail the whole call as extra keys.
173 action_data: dict[str, Any] = {
174 "message": message_text,
175 "target": rooms,
176 }
177 service_data: dict[str, Any] = {"format": fmt}
178 if thread_id:
179 service_data["thread_id"] = str(thread_id)
180 if images:
181 service_data["images"] = images
182 action_data[ATTR_DATA] = service_data
184 if raw_data:
185 _LOGGER.debug(
186 "SUPERNOTIFY matrix: dropping data keys not supported by the strict service schema: %s",
187 sorted(raw_data),
188 )
190 return await self.call_action(envelope, action_data=action_data)