Coverage for custom_components/supernotify/transports/email.py: 95%

1from __future__ import annotations 

2 

3import datetime as dt 

4import email.utils 

5import logging 

6import mimetypes 

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 

17 

18import aiofiles 

19import aiofiles.os 

20from anyio import Path 

21from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE 

22from homeassistant.const import ( 

23 ATTR_ENTITY_ID, 

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 

36 

37import custom_components.supernotify 

38from custom_components.supernotify import const 

39from custom_components.supernotify.common import CallRecord, ensure_list 

40from custom_components.supernotify.compat import vol 

41from custom_components.supernotify.const import ( 

42 ATTR_ACTION_URL, 

43 ATTR_ACTION_URL_TITLE, 

44 ATTR_EMAIL, 

45 ATTR_MEDIA, 

46 ATTR_MEDIA_CAMERA_ENTITY_ID, 

47 ATTR_MEDIA_SNAPSHOT_URL, 

48 CONF_CONNECTION, 

49 CONF_DELIVERY_DEFAULTS, 

50 CONF_ENCRYPTION, 

51 CONF_OPTIONS, 

52 CONF_TEMPLATE, 

53 TRANSPORT_EMAIL, 

54) 

55from custom_components.supernotify.model import ( 

56 DebugTrace, 

57 MessageOnlyPolicy, 

58 SuppressionReason, 

59 TransportConfig, 

60 TransportFeature, 

61) 

62from custom_components.supernotify.options import ( 

63 DATA_KEYS_SELECT_OPTION, 

64 MEDIA_OPTIONS, 

65 OPTION_DATA_KEYS_SELECT, 

66 OPTION_JPEG, 

67 OPTION_MESSAGE_USAGE, 

68 OPTION_PNG, 

69 OPTION_SIMPLIFY_TEXT, 

70 OPTION_STRIP_URLS, 

71 DeliveryOption, 

72) 

73from custom_components.supernotify.target import TargetEntityCategory 

74from custom_components.supernotify.transport import Transport 

75 

76if TYPE_CHECKING: 

77 from ssl import SSLContext 

78 

79 from homeassistant.helpers.typing import ConfigType 

80 

81 from custom_components.supernotify.context import Context 

82 from custom_components.supernotify.envelope import Envelope 

83 from custom_components.supernotify.hass_api import HomeAssistantAPI 

84 

85# RE_VALID_EMAIL = ( 

86# 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])?)+$" 

87# ) re-consider if needed or remove 

88OPTION_PREHEADER_BLANK = "preheader_blank" 

89OPTION_PREHEADER_LENGTH = "preheader_length" 

90OPTION_STRICT_TEMPLATE = "strict_template" 

91OPTION_SENDER = "sender" 

92OPTION_SENDER_NAME = "sender_name" 

93OPTION_DEFAULT_TITLE = "default_title" 

94OPTION_MODE = "mode" 

95EMAIL_OPTION_MODE_DIRECT = "direct" 

96EMAIL_OPTION_MODE_HA_SMTP = "ha_smtp" 

97 

98DEFAULT_SMTP_PORT = 587 

99DEFAULT_SMTP_ENCRYPTION = "starttls" 

100DEFAULT_SMTP_TIMEOUT = 5 

101NULL_RETURN_PATH = "<>" 

102 

103# Keys used in the HA core smtp integration's config entry data, for reuse when no 

104# connection is configured here. "server" is smtp-specific; the rest match generic 

105# homeassistant.const keys already imported above. 

106HA_SMTP_DOMAIN = "smtp" 

107SMTP_SEND_MESSAGE = "send_message" 

108# Duplicates of the constants in `homeassistant.components.smtp.const`, to be kept aligned with it, 

109# since importing from there loads the smtp integration and so needs its `aiosmtplib` requirement, 

110# which is only installed where the smtp integration is in use 

111CONF_SERVER = "server" 

112CONF_SENDER_NAME = "sender_name" 

113ATTR_ATTACHMENTS = "attachments" 

114ATTR_MEDIA_SOURCE = "media_source" 

115ATTR_FILENAME = "filename" 

116ATTR_CONTENT_ID = "content_id" 

117ATTR_PRIORITY = "priority" 

118 

119# smtp.send_message's priority, which it sends only as X-Priority 

120SMTP_PRIORITY_MAP: dict[str, str] = { 

121 const.PRIORITY_CRITICAL: "highest", 

122 const.PRIORITY_HIGH: "high", 

123 const.PRIORITY_MEDIUM: "normal", 

124 const.PRIORITY_LOW: "low", 

125 const.PRIORITY_MINIMUM: "lowest", 

126} 

127 

128IMPORTANCE_HEADER_MAP: dict[str, str] = { 

129 const.PRIORITY_CRITICAL: "high", 

130 const.PRIORITY_HIGH: "high", 

131 const.PRIORITY_MEDIUM: "normal", 

132 const.PRIORITY_LOW: "low", 

133 const.PRIORITY_MINIMUM: "low", 

134} 

135PRIORITY_HEADER_MAP: dict[str, str] = { 

136 const.PRIORITY_CRITICAL: "urgent", 

137 const.PRIORITY_HIGH: "urgent", 

138 const.PRIORITY_MEDIUM: "normal", 

139 const.PRIORITY_LOW: "non-urgent", 

140 const.PRIORITY_MINIMUM: "non-urgent", 

141} 

142X_MSMAIL_PRIORITY_HEADER_MAP: dict[str, str] = { 

143 const.PRIORITY_CRITICAL: "High", 

144 const.PRIORITY_HIGH: "High", 

145 const.PRIORITY_MEDIUM: "Normal", 

146 const.PRIORITY_LOW: "Low", 

147 const.PRIORITY_MINIMUM: "Low", 

148} 

149X_PRIORITY_HEADER_MAP: dict[str, str] = { 

150 const.PRIORITY_CRITICAL: "1", 

151 const.PRIORITY_HIGH: "2", 

152 const.PRIORITY_MEDIUM: "3", 

153 const.PRIORITY_LOW: "4", 

154 const.PRIORITY_MINIMUM: "5", 

155} 

156 

157_LOGGER = logging.getLogger(__name__) 

158 

159 

160class AlertServer(TypedDict): 

161 name: str 

162 internal_url: str 

163 external_url: str 

164 language: str 

165 

166 

167class AlertImage(TypedDict): 

168 url: str 

169 desc: str 

170 

171 

172class Alert(TypedDict): 

173 message: str | None 

174 title: str | None 

175 preheader: str | None 

176 priority: str 

177 envelope: Envelope 

178 action_url: str | None 

179 action_url_title: str | None 

180 subheading: str 

181 server: AlertServer 

182 preformatted_html: str | None 

183 img: AlertImage | None 

184 

185 

186class EmailTransport(Transport): 

187 name = TRANSPORT_EMAIL 

188 uses_connection = True 

189 declared_options: ClassVar[list[DeliveryOption]] = [ 

190 *MEDIA_OPTIONS, 

191 DeliveryOption( 

192 OPTION_STRICT_TEMPLATE, 

193 "Fail template if Jinja2 issues found when true, render anyway if false", 

194 value_type=cv.boolean, 

195 ), 

196 DeliveryOption(OPTION_PREHEADER_BLANK, "HTML code used to pack the pre-header with blanks for HTML email"), 

197 DeliveryOption( 

198 OPTION_PREHEADER_LENGTH, 

199 "Minimum size to pack the pre-header with blanks for HTML email", 

200 value_type=int, 

201 ), 

202 DeliveryOption( 

203 OPTION_MODE, 

204 "Set to direct to send over a direct SMTP connection instead of an action call", 

205 value_type=vol.In({ 

206 EMAIL_OPTION_MODE_DIRECT: "Use the native SMTP transport", 

207 EMAIL_OPTION_MODE_HA_SMTP: "Use the Home Assistant SMTP integration", 

208 }), 

209 ), 

210 DeliveryOption(OPTION_SENDER, "Sender address used in direct SMTP mode"), 

211 DeliveryOption(OPTION_SENDER_NAME, "Sender display name used in direct SMTP mode"), 

212 DeliveryOption(OPTION_DEFAULT_TITLE, "Default email subject if none supplied"), 

213 DATA_KEYS_SELECT_OPTION, 

214 ] 

215 

216 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None: 

217 super().__init__(context, transport_config) 

218 self.default_template_path: Path = Path(os.path.join(custom_components.supernotify.__path__[0], "default_templates")) 

219 self.custom_template_path: Path | None = context.custom_template_path 

220 self.custom_email_template_path: Path | None = None 

221 self.template_cache: dict[str, str] = {} 

222 

223 # Connection details for sending via a direct SMTP connection - only used for 

224 # deliveries with the OPTION_MODE option set to direct, rather than the default of 

225 # calling an HA notify action, but always read here since a delivery can request 

226 # direct sending independently of how this transport itself was configured. 

227 connection: ConfigType = (transport_config or {}).get(CONF_CONNECTION, {}) 

228 self.host: str | None = connection.get(CONF_HOST) 

229 self.port: int = connection.get(CONF_PORT, DEFAULT_SMTP_PORT) 

230 self.encryption: str = connection.get(CONF_ENCRYPTION, DEFAULT_SMTP_ENCRYPTION) 

231 self.username: str | None = connection.get(CONF_USERNAME) 

232 self.password: str | None = connection.get(CONF_PASSWORD) 

233 self.timeout: int = connection.get(CONF_TIMEOUT, DEFAULT_SMTP_TIMEOUT) 

234 self.verify_ssl: bool = connection.get(CONF_VERIFY_SSL, True) 

235 options: dict[str, Any] = (transport_config or {}).get(CONF_DELIVERY_DEFAULTS, {}).get(CONF_OPTIONS, {}) 

236 self.sender: str | None = options.get(OPTION_SENDER) 

237 self.sender_name: str | None = options.get(OPTION_SENDER_NAME) 

238 self.default_title: str | None = options.get(OPTION_DEFAULT_TITLE) 

239 self.ha_action: str | None = self.hass_api.find_service("notify", "homeassistant.components.smtp.notify") 

240 if not self.host: 

241 self._reuse_ha_smtp_connection() 

242 if self.host and self.port: 

243 self.local_smtp = True 

244 else: 

245 self.local_smtp = False 

246 

247 def _reuse_ha_smtp_connection(self) -> None: 

248 """No direct SMTP connection configured here; fall back to a configured HA smtp 

249 integration entry, if any.""" 

250 entry_data = self.hass_api.find_config_entry_data(HA_SMTP_DOMAIN) 

251 if not entry_data: 

252 _LOGGER.debug("SUPERNOTIFY No home assistant official smtp configuration to reuse") 

253 return 

254 _LOGGER.info("SUPERNOTIFY Email transport reusing connection from HA smtp integration for direct SMTP sends") 

255 self.host = entry_data.get(CONF_SERVER) 

256 self.port = entry_data.get(CONF_PORT, self.port) 

257 self.encryption = entry_data.get(CONF_ENCRYPTION, self.encryption) 

258 self.username = entry_data.get(CONF_USERNAME, self.username) 

259 self.password = entry_data.get(CONF_PASSWORD, self.password) 

260 self.verify_ssl = entry_data.get(CONF_VERIFY_SSL, self.verify_ssl) 

261 if not self.sender: 

262 self.sender = entry_data.get(CONF_SENDER) 

263 if not self.sender_name: 

264 self.sender_name = entry_data.get(CONF_SENDER_NAME) 

265 

266 async def initialize(self) -> None: 

267 if self.local_smtp and not self.sender: 

268 # as for Home Assistant's own SMTP integration, there's no sending without a sender 

269 _LOGGER.warning( 

270 "SUPERNOTIFY Email connection to %s has no sender, only deliveries with their own will send", self.host 

271 ) 

272 try: 

273 if self.custom_template_path is not None: 

274 if await self.custom_template_path.exists(): 

275 if await (self.custom_template_path / "email").exists(): 

276 _LOGGER.debug("SUPERNOTIFY Using email specific custom templates at %s", self.custom_template_path) 

277 self.custom_email_template_path = Path(self.custom_template_path / "email") 

278 else: 

279 _LOGGER.debug("SUPERNOTIFY Email specific custom templates not configured") 

280 else: 

281 _LOGGER.info("SUPERNOTIFY Custom email template directory not present at %s", self.custom_template_path) 

282 self.custom_template_path = None 

283 else: 

284 _LOGGER.info("SUPERNOTIFY Custom email templates not configured") 

285 except Exception as e: 

286 _LOGGER.error("SUPERNOTIFY Failed to verify custom template path %s: %s", self.custom_template_path, e) 

287 

288 def validate_action(self, action: str | None) -> bool: 

289 """Valid either with an HA notify action, or a usable direct SMTP connection for 

290 deliveries that set OPTION_MODE to 'direct'.""" 

291 return ( 

292 action is not None 

293 or self.ha_action is not None 

294 or self.local_smtp 

295 or self.hass_api.has_service(HA_SMTP_DOMAIN, SMTP_SEND_MESSAGE) 

296 ) 

297 

298 def is_viable(self, hass_api: HomeAssistantAPI) -> bool: 

299 # like validate_action() above, an explicit delivery can supply its own action 

300 # (or connection details) regardless of whether the native smtp integration or a 

301 # transport-level host/sender is discoverable - is_viable() can't see delivery-level 

302 # config, so it can't rule that out; DeliveryRegistry prunes this transport entirely 

303 # once it's confirmed no delivery (explicit or auto) actually uses it 

304 return True 

305 

306 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]: 

307 if ( 

308 self.delivery_defaults.options.get(OPTION_MODE, EMAIL_OPTION_MODE_DIRECT) == EMAIL_OPTION_MODE_DIRECT 

309 and self.local_smtp 

310 and self.sender 

311 ) or self.ha_action: 

312 return {self.name: {}} 

313 if hass_api.has_service(HA_SMTP_DOMAIN, SMTP_SEND_MESSAGE) and hass_api.smtp_recipients(): 

314 return {self.name: {}} 

315 return {} 

316 

317 @property 

318 def supported_features(self) -> TransportFeature: 

319 return ( 

320 TransportFeature.MESSAGE 

321 | TransportFeature.TITLE 

322 | TransportFeature.ACTIONS 

323 | TransportFeature.IMAGES 

324 | TransportFeature.TEMPLATE_FILE 

325 | TransportFeature.SNAPSHOT_IMAGE 

326 ) 

327 

328 def extra_attributes(self) -> dict[str, Any]: 

329 return { 

330 "cached_templates": list(self.template_cache.keys()), 

331 "custom_templates": str(self.custom_template_path) if self.custom_template_path else None, 

332 "custom_email_templates": str(self.custom_email_template_path) if self.custom_email_template_path else None, 

333 } 

334 

335 @property 

336 def default_config(self) -> TransportConfig: 

337 config = super().default_config 

338 config.delivery_defaults.options = { 

339 OPTION_SIMPLIFY_TEXT: False, 

340 OPTION_STRIP_URLS: False, 

341 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

342 # use sensible defaults for image attachments 

343 OPTION_JPEG: {"progressive": "true", "optimize": "true"}, 

344 OPTION_PNG: {"optimize": "true"}, 

345 OPTION_STRICT_TEMPLATE: False, 

346 OPTION_PREHEADER_BLANK: "&#847;&zwnj;&nbsp;", 

347 OPTION_PREHEADER_LENGTH: 100, 

348 OPTION_MODE: EMAIL_OPTION_MODE_DIRECT, # default to avoiding the e-mail integration, since it will get locked down to notify entities 

349 OPTION_DATA_KEYS_SELECT: None, 

350 } 

351 return config 

352 

353 @property 

354 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

355 return [ATTR_EMAIL] 

356 

357 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: 

358 # resolved_targets(), not the typed `.email` getter: envelope.target is already 

359 # scoped to this delivery by Delivery.select_targets(), so this also picks up a 

360 # `email:`/`{email: ...}`-qualified address that isn't shaped like a validated one 

361 addresses: list[str] = envelope.target.resolved_targets() if envelope.target else [] 

362 _LOGGER.debug("SUPERNOTIFY notify_email: %s %s", envelope.delivery_name, addresses) 

363 

364 # what's left of `data` once Supernotify's own keys are taken out is the target integration's own, 

365 # such as smtp's `html` and `images`, or Google Mail's `cc` 

366 data: dict[str, Any] = dict(envelope.data or {}) 

367 nested: dict[str, Any] | None = data.pop(ATTR_DATA, None) 

368 if nested: 

369 # notify.supernotify needed a `data` inside `data` for the integration's own keys 

370 _LOGGER.warning( 

371 "SUPERNOTIFY email %s: give %s directly in `data` or `extra_data`, rather than in a nested `data`", 

372 envelope.delivery_name, 

373 ", ".join(nested), 

374 ) 

375 data = nested | data 

376 template_name: str | None = data.pop(CONF_TEMPLATE, envelope.delivery.template) 

377 strict_template: bool = envelope.delivery.options.get(OPTION_STRICT_TEMPLATE, False) 

378 # `media` is a notification field, also once looked for in `data` 

379 legacy_media: dict[str, Any] = data.pop(ATTR_MEDIA, None) or {} 

380 legacy_snapshot_url: str | None = data.pop(ATTR_MEDIA_SNAPSHOT_URL, None) 

381 snapshot_url: str | None = ( 

382 (envelope.media or {}).get(ATTR_MEDIA_SNAPSHOT_URL) 

383 or legacy_media.get(ATTR_MEDIA_SNAPSHOT_URL) 

384 or legacy_snapshot_url 

385 ) 

386 template_data: dict[str, Any] = {k: data.pop(k) for k in (ATTR_ACTION_URL, ATTR_ACTION_URL_TITLE) if k in data} 

387 # TODO: centralize in config 

388 footer_template = data.pop("footer", None) 

389 footer = None 

390 if footer_template: 

391 try: 

392 footer = footer_template.format(e=envelope) 

393 except (KeyError, ValueError, AttributeError) as ex: # py3.13 compat 

394 _LOGGER.warning("SUPERNOTIFY email: failed to render footer template: %s", ex) 

395 

396 action_data: dict[str, Any] = envelope.core_action_data() 

397 

398 if len(addresses) > 0: 

399 action_data[ATTR_TARGET] = addresses 

400 # default to SMTP platform default recipients if no explicit addresses 

401 

402 image_path: Path | None = await envelope.grab_image() 

403 if image_path: 

404 data["images"] = [*ensure_list(data.get("images")), str(image_path)] 

405 

406 if not template_name and footer and action_data.get(ATTR_MESSAGE): 

407 action_data[ATTR_MESSAGE] = f"{action_data[ATTR_MESSAGE]}\n\n{footer}" 

408 

409 # message_html is Supernotify's own for smtp's `html`, which wins when given itself 

410 if "html" not in data: 

411 if not template_name: 

412 if envelope.message_html: 

413 html: str = envelope.message_html 

414 if image_path: 

415 image_name = image_path.name 

416 if not html.rstrip().endswith("</html>"): 

417 if snapshot_url: 

418 html += f'<div><p><a href="{snapshot_url}">' 

419 html += f'<img src="cid:{image_name}"/></a>' 

420 html += "</p></div>" 

421 else: 

422 html += f'<div><p><img src="cid:{image_name}"></p></div>' 

423 data["html"] = html 

424 else: 

425 rendered: str | None = await self.render_template( 

426 template_name, 

427 envelope, 

428 action_data, 

429 debug_trace, 

430 image_path=image_path, 

431 snapshot_url=snapshot_url, 

432 extra_data=template_data, 

433 strict_template=strict_template, 

434 ) 

435 if rendered: 

436 data["html"] = rendered 

437 data = envelope.customize_data(data) 

438 if data: 

439 action_data[ATTR_DATA] = data 

440 return await self._send(envelope, action_data) 

441 

442 async def _send(self, envelope: Envelope, action_data: dict[str, Any]) -> bool: 

443 """Send the built action_data, either via an HA notify action, or by owning the SMTP 

444 connection directly - for deliveries with the OPTION_MODE option set, so 

445 email can be sent to arbitrary addresses without every recipient needing to be 

446 pre-registered as a notify entity, and isn't limited to whatever a given HA notify 

447 action exposes.""" 

448 if envelope.delivery.action: 

449 # explicit action, use that 

450 return await self.call_action(envelope, action_data=await self._without_missing_images(action_data)) 

451 if envelope.delivery.options.get(OPTION_MODE, EMAIL_OPTION_MODE_DIRECT) == EMAIL_OPTION_MODE_DIRECT and self.local_smtp: 

452 return await self._send_direct_smtp(envelope, action_data) 

453 return await self._send_ha_smtp(envelope, await self._without_missing_images(action_data)) 

454 

455 async def _without_missing_images(self, action_data: dict[str, Any]) -> dict[str, Any]: 

456 """Leave out image files that can't be found, as a direct send does, so at least the text gets 

457 through - Home Assistant's SMTP integration fails the whole email for one from 2026.11""" 

458 data: dict[str, Any] = action_data.get(ATTR_DATA) or {} 

459 if not data.get("images"): 

460 return action_data 

461 images: list[str] = [] 

462 for image in ensure_list(data["images"]): 

463 if await aiofiles.os.path.exists(image): 

464 images.append(image) 

465 else: 

466 _LOGGER.warning("SUPERNOTIFY SMTP attachment %s not found, skipping", image) 

467 data = {k: v for k, v in data.items() if k != "images"} 

468 if images: 

469 data["images"] = images 

470 return {**action_data, ATTR_DATA: data} 

471 

472 async def _send_ha_smtp(self, envelope: Envelope, action_data: dict[str, Any]) -> bool: 

473 """Through Home Assistant's SMTP integration - smtp.send_message to the addresses set up there as 

474 recipients, and its legacy notify action, where there is one, for any others""" 

475 addresses: list[str] = action_data.get(ATTR_TARGET) or [] 

476 recipients: dict[str, str] = ( 

477 self.hass_api.smtp_recipients() if self.hass_api.has_service(HA_SMTP_DOMAIN, SMTP_SEND_MESSAGE) else {} 

478 ) 

479 entity_ids: list[str] = [recipients[address] for address in addresses if address in recipients] 

480 if not entity_ids: 

481 return await self.call_action(envelope, action_data=action_data, qualified_action=self.ha_action) 

482 

483 sent: bool = await self.call_action( 

484 envelope, 

485 qualified_action=f"{HA_SMTP_DOMAIN}.{SMTP_SEND_MESSAGE}", 

486 action_data=await self._send_message_data(envelope, action_data), 

487 target_data={ATTR_ENTITY_ID: entity_ids}, 

488 ) 

489 others: list[str] = [address for address in addresses if address not in recipients] 

490 if others: 

491 if self.ha_action: 

492 sent = ( 

493 await self.call_action( 

494 envelope, action_data={**action_data, ATTR_TARGET: others}, qualified_action=self.ha_action 

495 ) 

496 or sent 

497 ) 

498 else: 

499 _LOGGER.warning( 

500 "SUPERNOTIFY Not emailing %s for %s, not set up as recipients in the SMTP integration", 

501 ", ".join(others), 

502 envelope.delivery_name, 

503 ) 

504 return sent 

505 

506 async def _send_message_data(self, envelope: Envelope, action_data: dict[str, Any]) -> dict[str, Any]: 

507 """The fields of smtp.send_message, which takes images as media sources rather than files""" 

508 call_data: dict[str, Any] = {k: v for k, v in action_data.items() if k in (ATTR_MESSAGE, ATTR_TITLE)} 

509 call_data.update(action_data.get(ATTR_DATA) or {}) 

510 attachments: list[dict[str, Any]] = ensure_list(call_data.get(ATTR_ATTACHMENTS)) 

511 for image in ensure_list(call_data.pop("images", None)): 

512 if attachment := self._media_source_attachment(envelope, image): 

513 attachments.append(attachment) 

514 if attachments: 

515 call_data[ATTR_ATTACHMENTS] = attachments 

516 if envelope.priority in SMTP_PRIORITY_MAP: 

517 call_data[ATTR_PRIORITY] = SMTP_PRIORITY_MAP[envelope.priority] 

518 # anything smtp.send_message doesn't take, such as another integration's keys, is left out 

519 return self.hass_api.coerce_schema(HA_SMTP_DOMAIN, SMTP_SEND_MESSAGE, call_data) 

520 

521 def _media_source_attachment(self, envelope: Envelope, image: str) -> dict[str, Any] | None: 

522 """An image file as an attachment from one of Home Assistant's media folders, or else as a 

523 fresh snapshot of the notification's camera, named as the html refers to it with `cid:`""" 

524 name: str = os.path.basename(image) 

525 image_path: str = os.path.abspath(image) 

526 media_content_id: str | None = None 

527 for media_dir, root in self.hass_api.media_dirs.items(): 

528 root_path: str = os.path.abspath(root) 

529 if os.path.commonpath([image_path, root_path]) == root_path: 

530 media_content_id = f"media-source://media_source/{media_dir}/{os.path.relpath(image_path, root_path)}" 

531 break 

532 camera_entity_id: str | None = (envelope.media or {}).get(ATTR_MEDIA_CAMERA_ENTITY_ID) 

533 if media_content_id is None and camera_entity_id: 

534 media_content_id = f"media-source://camera/{camera_entity_id}" 

535 if media_content_id is None: 

536 _LOGGER.warning("SUPERNOTIFY Leaving %s out of the email, as it's not in one of Home Assistant's media_dirs", image) 

537 return None 

538 return { 

539 ATTR_MEDIA_SOURCE: { 

540 "media_content_id": media_content_id, 

541 "media_content_type": mimetypes.guess_type(name)[0] or "image/jpeg", 

542 }, 

543 ATTR_FILENAME: name, 

544 ATTR_CONTENT_ID: name, 

545 } 

546 

547 async def _send_direct_smtp(self, envelope: Envelope, action_data: dict[str, Any]) -> bool: 

548 addresses: list[str] = action_data.get(ATTR_TARGET) or [] 

549 start_time = time.time() 

550 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone()) 

551 sender: str | None = envelope.delivery.options.get(OPTION_SENDER) or self.sender 

552 if not self.host or not sender: 

553 _LOGGER.warning("SUPERNOTIFY Skipping direct smtp delivery %s, no sender configured", envelope.delivery.name) 

554 envelope.skipped = 1 

555 envelope.skip_reason = SuppressionReason.NO_ACTION 

556 return False 

557 if not addresses: 

558 _LOGGER.debug("SUPERNOTIFY Skipping direct smtp delivery %s, no target addresses", envelope.delivery.name) 

559 envelope.skipped = 1 

560 envelope.skip_reason = SuppressionReason.NO_TARGET 

561 return False 

562 

563 try: 

564 msg = await self._build_message(action_data, addresses, envelope.priority, envelope.id, envelope.delivery.options) 

565 await self.hass_api.create_job(self._send_smtp, msg, addresses, sender) 

566 envelope.calls.append( 

567 CallRecord( 

568 timestamp, 

569 time.time() - start_time, 

570 domain="smtp", 

571 action="send_message", 

572 debug=envelope.delivery.debug, 

573 action_data=dict(action_data), 

574 target_data={ATTR_TARGET: addresses}, 

575 ) 

576 ) 

577 envelope.delivered = 1 

578 self.log_delivery_recovered() 

579 return True 

580 except Exception as e: 

581 self.record_error(str(e), method="_send_direct_smtp") 

582 envelope.failed_calls.append( 

583 CallRecord( 

584 timestamp, 

585 time.time() - start_time, 

586 domain="smtp", 

587 action="send_message", 

588 action_data=dict(action_data), 

589 target_data={ATTR_TARGET: addresses}, 

590 exception=str(e), 

591 ) 

592 ) 

593 self.log_delivery_failure(e, "SUPERNOTIFY Failed to send smtp email for %s", envelope.delivery.name) 

594 envelope.error_count += 1 

595 envelope.delivery_error = format_exception(e) 

596 return False 

597 

598 async def _build_message( 

599 self, 

600 action_data: dict[str, Any], 

601 addresses: list[str], 

602 priority: str | None, 

603 id: str | None, 

604 options: dict[str, Any] | None = None, 

605 ) -> MIMEMultipart | MIMEText: 

606 """`options` are the delivery's own, falling back to the transport's, then those of HA's smtp integration""" 

607 options = options or {} 

608 sender: str | None = options.get(OPTION_SENDER) or self.sender 

609 # as Home Assistant's own SMTP has no default sender name, the home's name is used, as more use than none 

610 sender_name: str | None = options.get(OPTION_SENDER_NAME) or self.sender_name or self.hass_api.hass_name 

611 title: str | None = action_data.get(ATTR_TITLE) 

612 message: str = action_data.get(ATTR_MESSAGE) or "" 

613 data: dict[str, Any] = action_data.get(ATTR_DATA) or {} 

614 html: str | None = data.get("html") 

615 images: list[str] = data.get("images") or [] 

616 

617 msg: MIMEMultipart | MIMEText 

618 if html or images: 

619 msg = MIMEMultipart("related") 

620 alternative = MIMEMultipart("alternative") 

621 alternative.attach(MIMEText(message, _charset="utf-8")) 

622 if html: 

623 alternative.attach(MIMEText(html, "html", _charset="utf-8")) 

624 msg.attach(alternative) 

625 for image_path in images: 

626 attachment = await self._attach_file(image_path) 

627 if attachment: 

628 msg.attach(attachment) 

629 else: 

630 msg = MIMEText(message) 

631 

632 msg["Subject"] = ( 

633 title or options.get(OPTION_DEFAULT_TITLE) or self.default_title or await self.common_string("email_heading") 

634 ) 

635 msg["To"] = ", ".join(addresses) 

636 if sender_name or sender: 

637 msg["From"] = email.utils.formataddr((sender_name or "", sender or "")) 

638 else: 

639 msg["From"] = NULL_RETURN_PATH 

640 msg["X-Mailer"] = "Home Assistant Supernotify" 

641 msg["Date"] = email.utils.format_datetime(dt_util.now()) 

642 msg["Message-Id"] = email.utils.make_msgid(idstring=id) 

643 if priority: 

644 msg["Importance"] = IMPORTANCE_HEADER_MAP.get(priority, "Normal") 

645 msg["Priority"] = PRIORITY_HEADER_MAP.get(priority, "normal") 

646 msg["X-Priority"] = X_PRIORITY_HEADER_MAP.get(priority, "3") 

647 msg["X-MSMail-Priority"] = X_MSMAIL_PRIORITY_HEADER_MAP.get(priority, "Normal") 

648 return msg 

649 

650 async def _attach_file(self, image_path: str) -> MIMEImage | MIMEApplication | None: 

651 try: 

652 async with aiofiles.open(image_path, "rb") as attachment_file: 

653 file_bytes = await attachment_file.read() 

654 except OSError: 

655 _LOGGER.warning("SUPERNOTIFY SMTP attachment %s not found, skipping", image_path) 

656 return None 

657 

658 content_id: str = os.path.basename(image_path) 

659 attachment: MIMEImage | MIMEApplication 

660 try: 

661 attachment = MIMEImage(file_bytes) 

662 except TypeError: 

663 attachment = MIMEApplication(file_bytes, Name=content_id) 

664 attachment["Content-Disposition"] = f'attachment; filename="{content_id}"' 

665 attachment.add_header("Content-ID", f"<{content_id}>") 

666 return attachment 

667 

668 def _send_smtp(self, msg: MIMEMultipart | MIMEText, addresses: list[str], sender: str | None = None) -> None: 

669 if not self.local_smtp or not self.host or not self.port: # redundant but quietens mypy 

670 _LOGGER.warning("SUPERNOTIFY Direct SMTP connection not configured") 

671 return 

672 

673 ssl_context: SSLContext | None = create_client_context() if self.verify_ssl else None 

674 client: smtplib.SMTP | smtplib.SMTP_SSL 

675 if self.encryption == "tls": 

676 client = smtplib.SMTP_SSL(self.host, self.port, timeout=self.timeout, context=ssl_context) 

677 else: 

678 client = smtplib.SMTP(self.host, self.port, timeout=self.timeout) 

679 try: 

680 client.ehlo_or_helo_if_needed() 

681 if self.encryption == "starttls": 

682 client.starttls(context=ssl_context) 

683 client.ehlo() 

684 if self.username and self.password: 

685 client.login(self.username, self.password) 

686 client.sendmail(sender or self.sender or NULL_RETURN_PATH, addresses, msg.as_string()) 

687 finally: 

688 with suppress(smtplib.SMTPException): 

689 client.quit() 

690 

691 async def load_template(self, template_name: str) -> str | None: 

692 if template_name in self.template_cache: 

693 return self.template_cache[template_name] 

694 

695 for root_path in ( 

696 self.custom_email_template_path, 

697 self.custom_template_path, 

698 self.default_template_path / "email", 

699 self.default_template_path, 

700 ): 

701 if root_path is not None: 

702 template_path: Path = root_path / template_name 

703 if await template_path.exists(): 

704 template: str 

705 async with aiofiles.open(template_path) as file: 

706 template = await file.read() 

707 self.template_cache[template_name] = template 

708 return template 

709 return None 

710 

711 async def render_template( 

712 self, 

713 template_name: str, 

714 envelope: Envelope, 

715 action_data: dict[str, Any], 

716 debug_trace: DebugTrace | None = None, 

717 image_path: Path | None = None,