Coverage for custom_components/supernotify/transports/email.py: 92%
353 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +0000
1from __future__ import annotations
3import datetime as dt
4import email.utils
5import logging
6import os
7import os.path
8import smtplib
9import time
10from contextlib import suppress
11from email.mime.application import MIMEApplication
12from email.mime.image import MIMEImage
13from email.mime.multipart import MIMEMultipart
14from email.mime.text import MIMEText
15from traceback import format_exception
16from typing import TYPE_CHECKING, Any, ClassVar, TypedDict
18import aiofiles
19import voluptuous as vol
20from anyio import Path
21from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
22from homeassistant.components.smtp.const import CONF_SENDER_NAME, CONF_SERVER
23from homeassistant.const import (
24 CONF_HOST,
25 CONF_PASSWORD,
26 CONF_PORT,
27 CONF_SENDER,
28 CONF_TIMEOUT,
29 CONF_USERNAME,
30 CONF_VERIFY_SSL,
31)
32from homeassistant.helpers import config_validation as cv
33from homeassistant.helpers.template import Template, TemplateError
34from homeassistant.util import dt as dt_util
35from homeassistant.util.ssl import create_client_context
37import custom_components.supernotify
38from custom_components.supernotify import const
39from custom_components.supernotify.common import CallRecord
40from custom_components.supernotify.const import (
41 ATTR_ACTION_URL,
42 ATTR_ACTION_URL_TITLE,
43 ATTR_EMAIL,
44 ATTR_MEDIA,
45 ATTR_MEDIA_SNAPSHOT_URL,
46 CONF_CONNECTION,
47 CONF_DELIVERY_DEFAULTS,
48 CONF_ENCRYPTION,
49 CONF_OPTIONS,
50 CONF_TEMPLATE,
51 INCLUSION_DEFAULT,
52 TRANSPORT_EMAIL,
53)
54from custom_components.supernotify.model import (
55 DebugTrace,
56 MessageOnlyPolicy,
57 SuppressionReason,
58 TransportConfig,
59 TransportFeature,
60)
61from custom_components.supernotify.options import (
62 MEDIA_OPTIONS,
63 OPTION_JPEG,
64 OPTION_MESSAGE_USAGE,
65 OPTION_PNG,
66 OPTION_SIMPLIFY_TEXT,
67 OPTION_STRIP_URLS,
68 OPTION_UNIQUE_TARGETS,
69 DeliveryOption,
70)
71from custom_components.supernotify.target import TargetEntityCategory
72from custom_components.supernotify.transport import Transport
74if TYPE_CHECKING:
75 from ssl import SSLContext
77 from homeassistant.helpers.typing import ConfigType
79 from custom_components.supernotify.context import Context
80 from custom_components.supernotify.envelope import Envelope
81 from custom_components.supernotify.hass_api import HomeAssistantAPI
83RE_VALID_EMAIL = (
84 r"^[a-zA-Z0-9.+/=?^_-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+$"
85)
86OPTION_PREHEADER_BLANK = "preheader_blank"
87OPTION_PREHEADER_LENGTH = "preheader_length"
88OPTION_STRICT_TEMPLATE = "strict_template"
89OPTION_SENDER = "sender"
90OPTION_SENDER_NAME = "sender_name"
91OPTION_DEFAULT_TITLE = "default_title"
92OPTION_MODE = "mode"
93EMAIL_OPTION_MODE_DIRECT = "direct"
94EMAIL_OPTION_MODE_HA_SMTP = "ha_smtp"
96DEFAULT_SMTP_PORT = 587
97DEFAULT_SMTP_ENCRYPTION = "starttls"
98DEFAULT_SMTP_TIMEOUT = 5
99NULL_RETURN_PATH = "<>"
101# Keys used in the HA core smtp integration's config entry data, for reuse when no
102# connection is configured here. "server" is smtp-specific; the rest match generic
103# homeassistant.const keys already imported above.
104HA_SMTP_DOMAIN = "smtp"
106IMPORTANCE_HEADER_MAP: dict[str, str] = {
107 const.PRIORITY_CRITICAL: "high",
108 const.PRIORITY_HIGH: "high",
109 const.PRIORITY_MEDIUM: "normal",
110 const.PRIORITY_LOW: "low",
111 const.PRIORITY_MINIMUM: "low",
112}
113PRIORITY_HEADER_MAP: dict[str, str] = {
114 const.PRIORITY_CRITICAL: "urgent",
115 const.PRIORITY_HIGH: "urgent",
116 const.PRIORITY_MEDIUM: "normal",
117 const.PRIORITY_LOW: "non-urgent",
118 const.PRIORITY_MINIMUM: "non-urgent",
119}
120X_MSMAIL_PRIORITY_HEADER_MAP: dict[str, str] = {
121 const.PRIORITY_CRITICAL: "High",
122 const.PRIORITY_HIGH: "High",
123 const.PRIORITY_MEDIUM: "Normal",
124 const.PRIORITY_LOW: "Low",
125 const.PRIORITY_MINIMUM: "Low",
126}
127X_PRIORITY_HEADER_MAP: dict[str, str] = {
128 const.PRIORITY_CRITICAL: "1",
129 const.PRIORITY_HIGH: "2",
130 const.PRIORITY_MEDIUM: "3",
131 const.PRIORITY_LOW: "4",
132 const.PRIORITY_MINIMUM: "5",
133}
135_LOGGER = logging.getLogger(__name__)
138class AlertServer(TypedDict):
139 name: str
140 internal_url: str
141 external_url: str
142 language: str
145class AlertImage(TypedDict):
146 url: str
147 desc: str
150class Alert(TypedDict):
151 message: str | None
152 title: str | None
153 preheader: str | None
154 priority: str
155 envelope: Envelope
156 action_url: str | None
157 action_url_title: str | None
158 subheading: str
159 server: AlertServer
160 preformatted_html: str | None
161 img: AlertImage | None
164class EmailTransport(Transport):
165 name = TRANSPORT_EMAIL
166 declared_options: ClassVar[list[DeliveryOption]] = [
167 *MEDIA_OPTIONS,
168 DeliveryOption(
169 OPTION_STRICT_TEMPLATE,
170 "Fail template if Jinja2 issues found when true, render anyway if false",
171 value_type=cv.boolean,
172 ),
173 DeliveryOption(OPTION_PREHEADER_BLANK, "HTML code used to pack the pre-header with blanks for HTML email"),
174 DeliveryOption(
175 OPTION_PREHEADER_LENGTH,
176 "Minimum size to pack the pre-header with blanks for HTML email",
177 value_type=int,
178 ),
179 DeliveryOption(
180 OPTION_MODE,
181 "Set to direct to send over a direct SMTP connection instead of an action call",
182 value_type=vol.In({
183 EMAIL_OPTION_MODE_DIRECT: "Use the native SMTP transport",
184 EMAIL_OPTION_MODE_HA_SMTP: "Use the Home Assistant SMTP integration",
185 }),
186 ),
187 DeliveryOption(OPTION_SENDER, "Sender address used in direct SMTP mode"),
188 DeliveryOption(OPTION_SENDER_NAME, "Sender display name used in direct SMTP mode"),
189 DeliveryOption(OPTION_DEFAULT_TITLE, "Default email subject if none supplied"),
190 ]
192 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
193 super().__init__(context, transport_config)
194 self.default_template_path: Path = Path(os.path.join(custom_components.supernotify.__path__[0], "default_templates"))
195 self.custom_template_path: Path | None = context.custom_template_path
196 self.custom_email_template_path: Path | None = None
197 self.template_cache: dict[str, str] = {}
199 # Connection details for sending via a direct SMTP connection - only used for
200 # deliveries with the OPTION_MODE option set to direct, rather than the default of
201 # calling an HA notify action, but always read here since a delivery can request
202 # direct sending independently of how this transport itself was configured.
203 connection: ConfigType = (transport_config or {}).get(CONF_CONNECTION, {})
204 self.host: str | None = connection.get(CONF_HOST)
205 self.port: int = connection.get(CONF_PORT, DEFAULT_SMTP_PORT)
206 self.encryption: str = connection.get(CONF_ENCRYPTION, DEFAULT_SMTP_ENCRYPTION)
207 self.username: str | None = connection.get(CONF_USERNAME)
208 self.password: str | None = connection.get(CONF_PASSWORD)
209 self.timeout: int = connection.get(CONF_TIMEOUT, DEFAULT_SMTP_TIMEOUT)
210 self.verify_ssl: bool = connection.get(CONF_VERIFY_SSL, True)
211 options: dict[str, Any] = (transport_config or {}).get(CONF_DELIVERY_DEFAULTS, {}).get(CONF_OPTIONS, {})
212 self.sender: str | None = options.get(OPTION_SENDER)
213 self.sender_name: str | None = options.get(OPTION_SENDER_NAME)
214 self.default_title: str | None = options.get(OPTION_DEFAULT_TITLE)
215 self.ha_action: str | None = self.hass_api.find_service("notify", "homeassistant.components.smtp.notify")
216 if not self.host:
217 self._reuse_ha_smtp_connection()
218 if self.host and self.port:
219 self.local_smtp = True
220 else:
221 self.local_smtp = False
223 def _reuse_ha_smtp_connection(self) -> None:
224 """No direct SMTP connection configured here; fall back to a configured HA smtp
225 integration entry, if any."""
226 entry_data = self.hass_api.find_config_entry_data(HA_SMTP_DOMAIN)
227 if not entry_data:
228 _LOGGER.debug("SUPERNOTIFY No home assistant official smtp configuration to reuse")
229 return
230 _LOGGER.info("SUPERNOTIFY Email transport reusing connection from HA smtp integration for direct SMTP sends")
231 self.host = entry_data.get(CONF_SERVER)
232 self.port = entry_data.get(CONF_PORT, self.port)
233 self.encryption = entry_data.get(CONF_ENCRYPTION, self.encryption)
234 self.username = entry_data.get(CONF_USERNAME, self.username)
235 self.password = entry_data.get(CONF_PASSWORD, self.password)
236 self.verify_ssl = entry_data.get(CONF_VERIFY_SSL, self.verify_ssl)
237 if not self.sender:
238 self.sender = entry_data.get(CONF_SENDER)
239 if not self.sender_name:
240 self.sender_name = entry_data.get(CONF_SENDER_NAME)
242 async def initialize(self) -> None:
243 try:
244 if self.custom_template_path is not None:
245 if await self.custom_template_path.exists():
246 if await (self.custom_template_path / "email").exists():
247 _LOGGER.debug("SUPERNOTIFY Using email specific custom templates at %s", self.custom_template_path)
248 self.custom_email_template_path = Path(self.custom_template_path / "email")
249 else:
250 _LOGGER.debug("SUPERNOTIFY Email specific custom templates not configured")
251 else:
252 _LOGGER.info("SUPERNOTIFY Custom email template directory not present at %s", self.custom_template_path)
253 self.custom_template_path = None
254 else:
255 _LOGGER.info("SUPERNOTIFY Custom email templates not configured")
256 except Exception as e:
257 _LOGGER.error("SUPERNOTIFY Failed to verify custom template path %s: %s", self.custom_template_path, e)
259 def validate_action(self, action: str | None) -> bool:
260 """Valid either with an HA notify action, or a usable direct SMTP connection for
261 deliveries that set OPTION_MODE to 'direct'."""
262 return action is not None or self.ha_action is not None or self.local_smtp
264 def is_viable(self, hass_api: HomeAssistantAPI) -> bool:
265 # like validate_action() above, an explicit delivery can supply its own action
266 # (or connection details) regardless of whether the native smtp integration or a
267 # transport-level host/sender is discoverable - is_viable() can't see delivery-level
268 # config, so it can't rule that out; DeliveryRegistry prunes this transport entirely
269 # once it's confirmed no delivery (explicit or auto) actually uses it
270 return True
272 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]:
273 if (
274 self.delivery_defaults.options.get(OPTION_MODE, EMAIL_OPTION_MODE_DIRECT) == EMAIL_OPTION_MODE_DIRECT
275 and self.local_smtp
276 ) or self.ha_action:
277 return {self.name: {}}
278 return {}
280 @property
281 def inclusion_mode(self) -> list[str]:
282 # email addresses map cleanly to recipients, so it's reasonable to fire on
283 # every notification by default
284 return [INCLUSION_DEFAULT]
286 @property
287 def supported_features(self) -> TransportFeature:
288 return (
289 TransportFeature.MESSAGE
290 | TransportFeature.TITLE
291 | TransportFeature.ACTIONS
292 | TransportFeature.IMAGES
293 | TransportFeature.TEMPLATE_FILE
294 | TransportFeature.SNAPSHOT_IMAGE
295 )
297 def extra_attributes(self) -> dict[str, Any]:
298 return {
299 "cached_templates": list(self.template_cache.keys()),
300 "custom_templates": str(self.custom_template_path) if self.custom_template_path else None,
301 "custom_email_templates": str(self.custom_email_template_path) if self.custom_email_template_path else None,
302 }
304 @property
305 def default_config(self) -> TransportConfig:
306 config = TransportConfig()
307 config.delivery_defaults.inclusion = self.inclusion_mode
308 config.delivery_defaults.options = {
309 OPTION_SIMPLIFY_TEXT: False,
310 OPTION_STRIP_URLS: False,
311 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
312 # use sensible defaults for image attachments
313 OPTION_JPEG: {"progressive": "true", "optimize": "true"},
314 OPTION_PNG: {"optimize": "true"},
315 OPTION_STRICT_TEMPLATE: False,
316 OPTION_PREHEADER_BLANK: "͏‌ ",
317 OPTION_PREHEADER_LENGTH: 100,
318 OPTION_MODE: EMAIL_OPTION_MODE_DIRECT, # default to avoiding the e-mail integration, since it will get locked down to notify entities
319 OPTION_UNIQUE_TARGETS: True, # disable if people get multiple deliveries on same address
320 # only used for deliveries with OPTION_MODE set to 'direct'
321 OPTION_SENDER_NAME: "Home Assistant",
322 OPTION_DEFAULT_TITLE: "Home Assistant Notification",
323 }
324 return config
326 @property
327 def target_categories(self) -> list[str | TargetEntityCategory]:
328 return [ATTR_EMAIL]
330 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
331 # resolved_targets(), not the typed `.email` getter: envelope.target is already
332 # scoped to this delivery by Delivery.select_targets(), so this also picks up a
333 # `email:`/`{email: ...}`-qualified address that isn't shaped like a validated one
334 addresses: list[str] = envelope.target.resolved_targets() if envelope.target else []
335 _LOGGER.debug("SUPERNOTIFY notify_email: %s %s", envelope.delivery_name, addresses)
337 data: dict[str, Any] = envelope.data or {}
338 html: str | None = data.get("html")
339 template_name: str | None = data.get(CONF_TEMPLATE, envelope.delivery.template)
340 strict_template: bool = envelope.delivery.options.get(OPTION_STRICT_TEMPLATE, False)
341 snapshot_url: str | None = data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)
342 if snapshot_url is None:
343 # older location for backward compatibility
344 snapshot_url = data.get(ATTR_MEDIA_SNAPSHOT_URL)
345 # TODO: centralize in config
346 footer_template = data.get("footer")
347 footer = None
348 if footer_template:
349 try:
350 footer = footer_template.format(e=envelope)
351 except (KeyError, ValueError, AttributeError) as ex: # py3.13 compat
352 _LOGGER.warning("SUPERNOTIFY email: failed to render footer template: %s", ex)
354 action_data: dict[str, Any] = envelope.core_action_data()
355 extra_data: dict[str, Any] = {k: v for k, v in data.items() if k not in action_data}
357 if len(addresses) > 0:
358 action_data[ATTR_TARGET] = addresses
359 # default to SMTP platform default recipients if no explicit addresses
361 if data and data.get("data"):
362 action_data[ATTR_DATA] = data.get("data")
364 image_path: Path | None = await envelope.grab_image()
365 if image_path:
366 action_data.setdefault("data", {})
367 action_data["data"]["images"] = [str(image_path)]
369 if not template_name:
370 if footer and action_data.get(ATTR_MESSAGE):
371 action_data[ATTR_MESSAGE] = f"{action_data[ATTR_MESSAGE]}\n\n{footer}"
373 if envelope.message_html:
374 action_data.setdefault("data", {})
375 html = envelope.message_html
376 if image_path:
377 image_name = image_path.name
378 if html and not html.rstrip().endswith("</html>"):
379 if snapshot_url:
380 html += f'<div><p><a href="{snapshot_url}">'
381 html += f'<img src="cid:{image_name}"/></a>'
382 html += "</p></div>"
383 else:
384 html += f'<div><p><img src="cid:{image_name}"></p></div>'
386 action_data["data"]["html"] = html
387 else:
388 html = await self.render_template(
389 template_name,
390 envelope,
391 action_data,
392 debug_trace,
393 image_path=image_path,
394 snapshot_url=snapshot_url,
395 extra_data=extra_data,
396 strict_template=strict_template,
397 )
398 if html:
399 action_data.setdefault("data", {})
400 action_data["data"]["html"] = html
401 return await self._send(envelope, action_data)
403 async def _send(self, envelope: Envelope, action_data: dict[str, Any]) -> bool:
404 """Send the built action_data, either via an HA notify action, or by owning the SMTP
405 connection directly - for deliveries with the OPTION_MODE option set, so
406 email can be sent to arbitrary addresses without every recipient needing to be
407 pre-registered as a notify entity, and isn't limited to whatever a given HA notify
408 action exposes."""
409 if envelope.delivery.action:
410 # explicit action, use that
411 return await self.call_action(envelope, action_data=action_data)
412 if envelope.delivery.options.get(OPTION_MODE, EMAIL_OPTION_MODE_DIRECT) == EMAIL_OPTION_MODE_DIRECT and self.local_smtp:
413 return await self._send_direct_smtp(envelope, action_data)
414 return await self.call_action(envelope, action_data=action_data, qualified_action=self.ha_action)
416 async def _send_direct_smtp(self, envelope: Envelope, action_data: dict[str, Any]) -> bool:
417 addresses: list[str] = action_data.get(ATTR_TARGET) or []
418 start_time = time.time()
419 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
420 if not self.host or not self.sender:
421 _LOGGER.debug("SUPERNOTIFY Skipping direct smtp delivery %s, no connection configured", envelope.delivery.name)
422 envelope.skipped = 1
423 envelope.skip_reason = SuppressionReason.NO_ACTION
424 return False
425 if not addresses:
426 _LOGGER.debug("SUPERNOTIFY Skipping direct smtp delivery %s, no target addresses", envelope.delivery.name)
427 envelope.skipped = 1
428 envelope.skip_reason = SuppressionReason.NO_TARGET
429 return False
431 try:
432 msg = await self._build_message(action_data, addresses, envelope.priority, envelope.id)
433 await self.hass_api.create_job(self._send_smtp, msg, addresses)
434 envelope.calls.append(
435 CallRecord(
436 timestamp,
437 time.time() - start_time,
438 domain="smtp",
439 action="send_message",
440 debug=envelope.delivery.debug,
441 action_data=dict(action_data),
442 target_data={ATTR_TARGET: addresses},
443 )
444 )
445 envelope.delivered = 1
446 self.log_delivery_recovered()
447 return True
448 except Exception as e:
449 self.record_error(str(e), method="_send_direct_smtp")
450 envelope.failed_calls.append(
451 CallRecord(
452 timestamp,
453 time.time() - start_time,
454 domain="smtp",
455 action="send_message",
456 action_data=dict(action_data),
457 target_data={ATTR_TARGET: addresses},
458 exception=str(e),
459 )
460 )
461 self.log_delivery_failure(e, "SUPERNOTIFY Failed to send smtp email for %s", envelope.delivery.name)
462 envelope.error_count += 1
463 envelope.delivery_error = format_exception(e)
464 return False
466 async def _build_message(
467 self, action_data: dict[str, Any], addresses: list[str], priority: str | None, id: str | None
468 ) -> MIMEMultipart | MIMEText:
469 title: str | None = action_data.get(ATTR_TITLE)
470 message: str = action_data.get(ATTR_MESSAGE) or ""
471 data: dict[str, Any] = action_data.get(ATTR_DATA) or {}
472 html: str | None = data.get("html")
473 images: list[str] = data.get("images") or []
475 msg: MIMEMultipart | MIMEText
476 if html or images:
477 msg = MIMEMultipart("related")
478 alternative = MIMEMultipart("alternative")
479 alternative.attach(MIMEText(message, _charset="utf-8"))
480 if html:
481 alternative.attach(MIMEText(html, "html", _charset="utf-8"))
482 msg.attach(alternative)
483 for image_path in images:
484 attachment = await self._attach_file(image_path)
485 if attachment:
486 msg.attach(attachment)
487 else:
488 msg = MIMEText(message)
490 msg["Subject"] = title or self.default_title or ""
491 msg["To"] = ", ".join(addresses)
492 if self.sender_name or self.sender:
493 sender: str = email.utils.formataddr((self.sender_name or "", self.sender or ""))
494 else:
495 sender = NULL_RETURN_PATH
497 msg["From"] = sender
498 msg["X-Mailer"] = "Home Assistant Supernotify"
499 msg["Date"] = email.utils.format_datetime(dt_util.now())
500 msg["Message-Id"] = email.utils.make_msgid(idstring=id)
501 if priority:
502 msg["Importance"] = IMPORTANCE_HEADER_MAP.get(priority, "Normal")
503 msg["Priority"] = PRIORITY_HEADER_MAP.get(priority, "normal")
504 msg["X-Priority"] = X_PRIORITY_HEADER_MAP.get(priority, "3")
505 msg["X-MSMail-Priority"] = X_MSMAIL_PRIORITY_HEADER_MAP.get(priority, "Normal")
506 return msg
508 async def _attach_file(self, image_path: str) -> MIMEImage | MIMEApplication | None:
509 try:
510 async with aiofiles.open(image_path, "rb") as attachment_file:
511 file_bytes = await attachment_file.read()
512 except OSError:
513 _LOGGER.warning("SUPERNOTIFY SMTP attachment %s not found, skipping", image_path)
514 return None
516 content_id: str = os.path.basename(image_path)
517 attachment: MIMEImage | MIMEApplication
518 try:
519 attachment = MIMEImage(file_bytes)
520 except TypeError:
521 attachment = MIMEApplication(file_bytes, Name=content_id)
522 attachment["Content-Disposition"] = f'attachment; filename="{content_id}"'
523 attachment.add_header("Content-ID", f"<{content_id}>")
524 return attachment
526 def _send_smtp(self, msg: MIMEMultipart | MIMEText, addresses: list[str]) -> None:
527 if not self.local_smtp or not self.host or not self.port: # redundant but quietens mypy
528 _LOGGER.warning("SUPERNOTIFY Direct SMTP connection not configured")
529 return
531 ssl_context: SSLContext | None = create_client_context() if self.verify_ssl else None
532 client: smtplib.SMTP | smtplib.SMTP_SSL
533 if self.encryption == "tls":
534 client = smtplib.SMTP_SSL(self.host, self.port, timeout=self.timeout, context=ssl_context)
535 else:
536 client = smtplib.SMTP(self.host, self.port, timeout=self.timeout)
537 try:
538 client.ehlo_or_helo_if_needed()
539 if self.encryption == "starttls":
540 client.starttls(context=ssl_context)
541 client.ehlo()
542 if self.username and self.password:
543 client.login(self.username, self.password)
544 client.sendmail(self.sender or NULL_RETURN_PATH, addresses, msg.as_string())
545 finally:
546 with suppress(smtplib.SMTPException):
547 client.quit()
549 async def load_template(self, template_name: str) -> str | None:
550 if template_name in self.template_cache:
551 return self.template_cache[template_name]
553 for root_path in (
554 self.custom_email_template_path,
555 self.custom_template_path,
556 self.default_template_path / "email",
557 self.default_template_path,
558 ):
559 if root_path is not None:
560 template_path: Path = root_path / template_name
561 if await template_path.exists():
562 template: str
563 async with aiofiles.open(template_path) as file:
564 template = os.linesep.join(await file.readlines())
565 self.template_cache[template_name] = template
566 return template
567 return None
569 async def render_template(
570 self,
571 template_name: str,
572 envelope: Envelope,
573 action_data: dict[str, Any],
574 debug_trace: DebugTrace | None = None,
575 image_path: Path | None = None,
576 snapshot_url: str | None = None,
577 extra_data: dict[str, Any] | None = None,
578 strict_template: bool = False,
579 ) -> str | None:
580 extra_data = extra_data or {}
581 alert: Alert
583 try:
584 title: str | None = action_data.get(ATTR_TITLE)
585 message: str | None = action_data.get(ATTR_MESSAGE)
586 preheader: str = f"{title or ''}{' ' if title else ''}{message}"
587 preheader = preheader or "Home Assistant Notification"
588 alert = Alert(
589 message=message,
590 title=title,
591 preheader=self.pack_preheader(preheader, envelope.delivery.options),
592 priority=envelope.priority,
593 action_url=extra_data.get(ATTR_ACTION_URL),
594 action_url_title=extra_data.get(ATTR_ACTION_URL_TITLE),
595 envelope=envelope,
596 subheading="Home Assistant Notification",
597 server=AlertServer(
598 name=self.hass_api.hass_name,
599 internal_url=self.hass_api.internal_url,
600 external_url=self.hass_api.external_url,
601 language=self.hass_api.language,
602 ),
603 preformatted_html=envelope.message_html,
604 img=None,
605 )
607 if snapshot_url:
608 alert["img"] = AlertImage(url=snapshot_url, desc="Snapshot Image")
609 elif image_path:
610 alert["img"] = AlertImage(url=f"cid:{image_path.name}", desc=image_path.name)
612 template_content: str | None = await self.load_template(template_name)
614 if template_content is None:
615 _LOGGER.error("SUPERNOTIFY No template found for %s", template_name)
616 return None
618 template_obj: Template = self.context.hass_api.template(template_content)
619 template_obj.ensure_valid()
621 if debug_trace:
622 debug_trace.record_delivery_artefact(envelope.delivery.name, "alert", alert)
624 html: str = template_obj.async_render(variables={"alert": alert}, parse_result=False, strict=strict_template)
625 if not html:
626 _LOGGER.error("SUPERNOTIFY Empty result from template %s", template_name)
627 else:
628 return html
629 except TemplateError as te:
630 _LOGGER.exception("SUPERNOTIFY Failed to render template html mail")
631 if debug_trace:
632 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", te)
633 except Exception as e:
634 _LOGGER.exception("SUPERNOTIFY Failed to generate html mail")
635 if debug_trace:
636 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", e)
637 return None
639 def pack_preheader(self, preheader: str, options: dict[str, Any]) -> str:
640 preheader = preheader or ""
641 phchars: str = options.get(OPTION_PREHEADER_BLANK, "")
642 phlength: int = options.get(OPTION_PREHEADER_LENGTH, 0)
643 if phlength and phchars:
644 return f"{preheader}{phchars * (phlength - len(preheader))}"
645 return preheader