Coverage for custom_components / supernotify / transports / email.py: 95%
174 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +0000
1from __future__ import annotations
3import logging
4import os
5import os.path
6from typing import TYPE_CHECKING, Any, TypedDict
8import aiofiles
9from anyio import Path
10from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
11from homeassistant.helpers.template import Template, TemplateError
13import custom_components.supernotify
14from custom_components.supernotify.const import (
15 ATTR_ACTION_URL,
16 ATTR_ACTION_URL_TITLE,
17 ATTR_EMAIL,
18 ATTR_MEDIA,
19 ATTR_MEDIA_SNAPSHOT_URL,
20 CONF_TEMPLATE,
21 OPTION_JPEG,
22 OPTION_MESSAGE_USAGE,
23 OPTION_PNG,
24 OPTION_SIMPLIFY_TEXT,
25 OPTION_STRICT_TEMPLATE,
26 OPTION_STRIP_URLS,
27 OPTION_TARGET_CATEGORIES,
28 TRANSPORT_EMAIL,
29)
30from custom_components.supernotify.model import DebugTrace, DeliveryConfig, MessageOnlyPolicy, TransportConfig, TransportFeature
31from custom_components.supernotify.transport import Transport
33if TYPE_CHECKING:
34 from homeassistant.helpers.typing import ConfigType
36 from custom_components.supernotify.context import Context
37 from custom_components.supernotify.envelope import Envelope
38 from custom_components.supernotify.hass_api import HomeAssistantAPI
40RE_VALID_EMAIL = (
41 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])?)+$"
42)
43OPTION_PREHEADER_BLANK = "preheader_blank"
44OPTION_PREHEADER_LENGTH = "preheader_length"
46_LOGGER = logging.getLogger(__name__)
49class AlertServer(TypedDict):
50 name: str
51 internal_url: str
52 external_url: str
53 language: str
56class AlertImage(TypedDict):
57 url: str
58 desc: str
61class Alert(TypedDict):
62 message: str | None
63 title: str | None
64 preheader: str | None
65 priority: str
66 envelope: Envelope
67 action_url: str | None
68 action_url_title: str | None
69 subheading: str
70 server: AlertServer
71 preformatted_html: str | None
72 img: AlertImage | None
75class EmailTransport(Transport):
76 name = TRANSPORT_EMAIL
78 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
79 super().__init__(context, transport_config)
80 self.default_template_path: Path = Path(os.path.join(custom_components.supernotify.__path__[0], "default_templates")) # noqa: PTH118
81 self.custom_template_path: Path | None = context.custom_template_path
82 self.custom_email_template_path: Path | None = None
83 self.template_cache: dict[str, str] = {}
85 async def initialize(self) -> None:
86 try:
87 if self.custom_template_path is not None and await self.custom_template_path.exists():
88 if await (self.custom_template_path / "email").exists():
89 _LOGGER.debug("SUPERNOTIFY Using email specific custom templates at %s", self.custom_template_path)
90 self.custom_email_template_path = Path(self.custom_template_path / "email")
91 else:
92 _LOGGER.debug("SUPERNOTIFY Email specific custom templates not configured")
93 else:
94 _LOGGER.info("SUPERNOTIFY Custom templates not configured")
95 self.custom_template_path = None
96 except Exception as e:
97 _LOGGER.error("SUPERNOTIFY Failed to verify custom template path %s: %s", self.custom_template_path, e)
99 def validate_action(self, action: str | None) -> bool:
100 """Override in subclass if transport has fixed action or doesn't require one"""
101 return action is not None
103 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None:
104 action: str | None = hass_api.find_service("notify", "homeassistant.components.smtp.notify")
105 if action:
106 delivery_config: DeliveryConfig = self.delivery_defaults
107 delivery_config.action = action
108 return delivery_config
109 return None
111 @property
112 def supported_features(self) -> TransportFeature:
113 return (
114 TransportFeature.MESSAGE
115 | TransportFeature.TITLE
116 | TransportFeature.ACTIONS
117 | TransportFeature.IMAGES
118 | TransportFeature.TEMPLATE_FILE
119 | TransportFeature.SNAPSHOT_IMAGE
120 )
122 def extra_attributes(self) -> dict[str, Any]:
123 return {
124 "cached_templates": list(self.template_cache.keys()),
125 "custom_templates": str(self.custom_template_path) if self.custom_template_path else None,
126 "custom_email_templates": str(self.custom_email_template_path) if self.custom_email_template_path else None,
127 }
129 @property
130 def default_config(self) -> TransportConfig:
131 config = TransportConfig()
132 config.delivery_defaults.options = {
133 OPTION_SIMPLIFY_TEXT: False,
134 OPTION_STRIP_URLS: False,
135 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
136 OPTION_TARGET_CATEGORIES: [ATTR_EMAIL],
137 # use sensible defaults for image attachments
138 OPTION_JPEG: {"progressive": "true", "optimize": "true"},
139 OPTION_PNG: {"optimize": "true"},
140 OPTION_STRICT_TEMPLATE: False,
141 OPTION_PREHEADER_BLANK: "͏‌ ",
142 OPTION_PREHEADER_LENGTH: 100,
143 }
144 return config
146 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
147 _LOGGER.debug("SUPERNOTIFY notify_email: %s %s", envelope.delivery_name, envelope.target.email)
149 data: dict[str, Any] = envelope.data or {}
150 html: str | None = data.get("html")
151 template_name: str | None = data.get(CONF_TEMPLATE, envelope.delivery.template)
152 strict_template: bool = envelope.delivery.options.get(OPTION_STRICT_TEMPLATE, False)
153 addresses: list[str] = envelope.target.email or []
154 snapshot_url: str | None = data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)
155 if snapshot_url is None:
156 # older location for backward compatibility
157 snapshot_url = data.get(ATTR_MEDIA_SNAPSHOT_URL)
158 # TODO: centralize in config
159 footer_template = data.get("footer")
160 footer = None
161 if footer_template:
162 try:
163 footer = footer_template.format(e=envelope)
164 except (KeyError, ValueError, AttributeError) as ex: # py3.13 compat
165 _LOGGER.warning("SUPERNOTIFY email: failed to render footer template: %s", ex)
167 action_data: dict[str, Any] = envelope.core_action_data()
168 extra_data: dict[str, Any] = {k: v for k, v in data.items() if k not in action_data}
170 if len(addresses) > 0:
171 action_data[ATTR_TARGET] = addresses
172 # default to SMTP platform default recipients if no explicit addresses
174 if data and data.get("data"):
175 action_data[ATTR_DATA] = data.get("data")
177 image_path: Path | None = await envelope.grab_image()
178 if image_path:
179 action_data.setdefault("data", {})
180 action_data["data"]["images"] = [str(image_path)]
182 if not template_name:
183 if footer and action_data.get(ATTR_MESSAGE):
184 action_data[ATTR_MESSAGE] = f"{action_data[ATTR_MESSAGE]}\n\n{footer}"
186 if envelope.message_html:
187 action_data.setdefault("data", {})
188 html = envelope.message_html
189 if image_path:
190 image_name = image_path.name
191 if html and not html.rstrip().endswith("</html>"):
192 if snapshot_url:
193 html += f'<div><p><a href="{snapshot_url}">'
194 html += f'<img src="cid:{image_name}"/></a>'
195 html += "</p></div>"
196 else:
197 html += f'<div><p><img src="cid:{image_name}"></p></div>'
199 action_data["data"]["html"] = html
200 else:
201 html = await self.render_template(
202 template_name,
203 envelope,
204 action_data,
205 debug_trace,
206 image_path=image_path,
207 snapshot_url=snapshot_url,
208 extra_data=extra_data,
209 strict_template=strict_template,
210 )
211 if html:
212 action_data.setdefault("data", {})
213 action_data["data"]["html"] = html
214 return await self.call_action(envelope, action_data=action_data)
216 async def load_template(self, template_name: str) -> str | None:
217 if template_name in self.template_cache:
218 return self.template_cache[template_name]
220 for root_path in (
221 self.custom_email_template_path,
222 self.custom_template_path,
223 self.default_template_path / "email",
224 self.default_template_path,
225 ):
226 if root_path is not None:
227 template_path: Path = root_path / template_name
228 if await template_path.exists():
229 template: str
230 async with aiofiles.open(template_path) as file:
231 template = os.linesep.join(await file.readlines())
232 self.template_cache[template_name] = template
233 return template
234 return None
236 async def render_template(
237 self,
238 template_name: str,
239 envelope: Envelope,
240 action_data: dict[str, Any],
241 debug_trace: DebugTrace | None = None,
242 image_path: Path | None = None,
243 snapshot_url: str | None = None,
244 extra_data: dict[str, Any] | None = None,
245 strict_template: bool = False,
246 ) -> str | None:
247 extra_data = extra_data or {}
248 alert: Alert
250 try:
251 title: str | None = action_data.get(ATTR_TITLE)
252 message: str | None = action_data.get(ATTR_MESSAGE)
253 preheader: str = f"{title or ''}{' ' if title else ''}{message}"
254 preheader = preheader or "Home Assistant Notification"
255 alert = Alert(
256 message=message,
257 title=title,
258 preheader=self.pack_preheader(preheader, envelope.delivery.options),
259 priority=envelope.priority,
260 action_url=extra_data.get(ATTR_ACTION_URL),
261 action_url_title=extra_data.get(ATTR_ACTION_URL_TITLE),
262 envelope=envelope,
263 subheading="Home Assistant Notification",
264 server=AlertServer(
265 name=self.hass_api.hass_name,
266 internal_url=self.hass_api.internal_url,
267 external_url=self.hass_api.external_url,
268 language=self.hass_api.language,
269 ),
270 preformatted_html=envelope.message_html,
271 img=None,
272 )
274 if snapshot_url:
275 alert["img"] = AlertImage(url=snapshot_url, desc="Snapshot Image")
276 elif image_path:
277 alert["img"] = AlertImage(url=f"cid:{image_path.name}", desc=image_path.name)
279 template_content: str | None = await self.load_template(template_name)
281 if template_content is None:
282 _LOGGER.error("SUPERNOTIFY No template found for %s", template_name)
283 return None
285 template_obj: Template = self.context.hass_api.template(template_content)
286 template_obj.ensure_valid()
288 if debug_trace:
289 debug_trace.record_delivery_artefact(envelope.delivery.name, "alert", alert)
291 html: str = template_obj.async_render(variables={"alert": alert}, parse_result=False, strict=strict_template)
292 if not html:
293 _LOGGER.error("SUPERNOTIFY Empty result from template %s", template_name)
294 else:
295 return html
296 except TemplateError as te:
297 _LOGGER.exception("SUPERNOTIFY Failed to render template html mail: %s", te)
298 if debug_trace:
299 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", te)
300 except Exception as e:
301 _LOGGER.exception("SUPERNOTIFY Failed to generate html mail: %s", e)
302 if debug_trace:
303 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", e)
304 return None
306 def pack_preheader(self, preheader: str, options: dict[str, Any]) -> str:
307 preheader = preheader or ""
308 phchars: str = options.get(OPTION_PREHEADER_BLANK, "")
309 phlength: int = options.get(OPTION_PREHEADER_LENGTH, 0)
310 if phlength and phchars:
311 return f"{preheader}{phchars * (phlength - len(preheader))}"
312 return preheader