Coverage for custom_components / supernotify / notification.py: 90%

1from __future__ import annotations 

2 

3import asyncio 

4import datetime as dt 

5import json 

6import logging 

7import uuid 

8from pathlib import Path as _Path 

9from traceback import format_exception 

10from typing import TYPE_CHECKING, Any, cast 

11 

12import homeassistant.util.dt as dt_util 

13import voluptuous as vol 

14from homeassistant.components.notify.const import ATTR_DATA 

15from voluptuous import humanize 

16 

17from custom_components.supernotify.schema import SelectionRank 

18 

19from .archive import ArchivableObject 

20from .common import ensure_list, nullable_ensure_list, sanitize 

21from .const import ( 

22 ATTR_ACTION_GROUPS, 

23 ATTR_ACTIONS, 

24 ATTR_DEBUG, 

25 ATTR_DELIVERY, 

26 ATTR_DELIVERY_SELECTION, 

27 ATTR_FORCE_RESEND, 

28 ATTR_IMAGE, 

29 ATTR_MEDIA, 

30 ATTR_MEDIA_CAMERA_ENTITY_ID, 

31 ATTR_MEDIA_CLIP_URL, 

32 ATTR_MEDIA_SNAPSHOT_URL, 

33 ATTR_MESSAGE_HTML, 

34 ATTR_PERSON_ID, 

35 ATTR_PRIORITY, 

36 ATTR_RECIPIENTS, 

37 ATTR_SCENARIOS_APPLY, 

38 ATTR_SCENARIOS_CONSTRAIN, 

39 ATTR_SCENARIOS_REQUIRE, 

40 ATTR_SPOKEN_MESSAGE, 

41 ATTR_VIDEO, 

42 DELIVERY_SELECTION_EXPLICIT, 

43 DELIVERY_SELECTION_FIXED, 

44 DELIVERY_SELECTION_IMPLICIT, 

45 OPTION_UNIQUE_TARGETS, 

46 PRIORITY_MEDIUM, 

47 PRIORITY_VALUES, 

48 TARGET_USE_FIXED, 

49 TARGET_USE_MERGE_ALWAYS, 

50 TARGET_USE_MERGE_ON_DELIVERY_TARGETS, 

51 TARGET_USE_ON_NO_ACTION_TARGETS, 

52 TARGET_USE_ON_NO_DELIVERY_TARGETS, 

53) 

54from .envelope import Envelope 

55from .media_grab import snap_notification_image as _snap_notification_image 

56from .model import ( 

57 ConditionVariables, 

58 DebugTrace, 

59 DeliveryCustomization, 

60 SuppressionReason, 

61 Target, 

62 TargetRequired, 

63 TransportFeature, 

64) 

65from .schema import ACTION_DATA_SCHEMA, STRICT_ACTION_DATA_SCHEMA, Outcome 

66 

67if TYPE_CHECKING: 

68 from .context import Context 

69 from .delivery import Delivery, DeliveryRegistry 

70 from .people import PeopleRegistry, Recipient 

71 from .scenario import Scenario 

72 from .transport import ( 

73 Transport, 

74 ) 

75 

76_LOGGER = logging.getLogger(__name__) 

77 

78_VERSION: str = json.loads((_Path(__file__).parent / "manifest.json").read_text()).get("version", "unknown") 

79 

80# Deliveries mapping keys for debug / archive 

81KEY_DELIVERED = "delivered" 

82KEY_SUPPRESSED = "suppressed" 

83KEY_FAILED = "failed" 

84KEY_SKIPPED = "skipped" 

85 

86# supernotify specific data items not to be passed to transports in data 

87INTERNAL_DATA_KEYS = (ATTR_FORCE_RESEND, ATTR_SPOKEN_MESSAGE) 

88 

89type t_delivery_name = str 

90type t_outcome = str 

91 

92 

93class Notification(ArchivableObject): 

94 def __init__( 

95 self, 

96 context: Context, 

97 message: str | None = None, 

98 title: str | None = None, 

99 target: list[str] | str | None = None, 

100 action_data: dict[str, Any] | None = None, 

101 ) -> None: 

102 self.created: dt.datetime = dt.datetime.now(tz=dt_util.get_default_time_zone()) 

103 self.debug_trace: DebugTrace = DebugTrace(message=message, title=title, data=action_data, target=target) 

104 self.message: str | None = message 

105 self.context: Context = context 

106 self.people_registry: PeopleRegistry = context.people_registry 

107 self.delivery_registry: DeliveryRegistry = context.delivery_registry 

108 action_data = action_data or {} 

109 self._target: Target | None = Target(target) if target else None 

110 self._already_selected: Target = Target() 

111 self._title: str | None = title 

112 self.id = str(uuid.uuid1()) 

113 self.delivered: int = 0 

114 self.error_count: int = 0 

115 self.skipped: int = 0 

116 self.failed: int = 0 

117 self.suppressed: int = 0 

118 self.fallback: int = 0 

119 self.dupe: bool = False 

120 self.deliveries: dict[t_delivery_name, dict[t_outcome, list[str] | list[Envelope] | dict[str, Any]]] = {} 

121 self._skip_reasons: list[SuppressionReason] = [] 

122 

123 self.validate_action_data(action_data) 

124 # for compatibility with other notify calls, pass thru surplus data to underlying delivery transports 

125 self.extra_data: dict[str, Any] = { 

126 k: v for k, v in action_data.items() if k not in STRICT_ACTION_DATA_SCHEMA(action_data) 

127 } 

128 

129 action_data = {k: v for k, v in action_data.items() if k not in self.extra_data} 

130 self.extra_data.update(action_data.get(ATTR_DATA, {})) # nested `data` could be supernotify or target service 

131 

132 self.priority: str = action_data.get(ATTR_PRIORITY, PRIORITY_MEDIUM) 

133 self.message_html: str | None = action_data.get(ATTR_MESSAGE_HTML) 

134 self.force_resend: bool = action_data.get(ATTR_FORCE_RESEND, False) 

135 self.required_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_REQUIRE)) 

136 self.applied_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_APPLY)) 

137 self.constrain_scenario_names: list[str] = ensure_list(action_data.get(ATTR_SCENARIOS_CONSTRAIN)) 

138 self.delivery_selection: str | None = action_data.get(ATTR_DELIVERY_SELECTION) 

139 self.delivery_overrides: dict[str, DeliveryCustomization] = {} 

140 

141 delivery_data = action_data.get(ATTR_DELIVERY) 

142 if isinstance(delivery_data, list): 

143 # a bare list of deliveries implies intent to restrict 

144 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as explicit for list %s", delivery_data) 

145 if self.delivery_selection is None: 

146 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

147 self.delivery_overrides = {k: DeliveryCustomization({}) for k in action_data.get(ATTR_DELIVERY, [])} 

148 elif isinstance(delivery_data, str) and delivery_data: 

149 # a bare list of deliveries implies intent to restrict 

150 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as explicit for single %s", delivery_data) 

151 if self.delivery_selection is None: 

152 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

153 self.delivery_overrides = {delivery_data: DeliveryCustomization({})} 

154 elif isinstance(delivery_data, dict): 

155 # whereas a dict may be used to tune or restrict 

156 if self.delivery_selection is None: 

157 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

158 _LOGGER.debug("SUPERNOTIFY defaulting delivery selection as implicit for mapping %s", delivery_data) 

159 self.delivery_overrides = {k: DeliveryCustomization(v) for k, v in action_data.get(ATTR_DELIVERY, {}).items()} 

160 elif delivery_data: 

161 _LOGGER.warning("SUPERNOTIFY Unable to interpret delivery data %s", delivery_data) 

162 if self.delivery_selection is None: 

163 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

164 else: 

165 if self.delivery_selection is None: 

166 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

167 

168 self.action_groups: list[str] | None = nullable_ensure_list(action_data.get(ATTR_ACTION_GROUPS)) 

169 self.recipients_override: list[str] | None = nullable_ensure_list(action_data.get(ATTR_RECIPIENTS)) 

170 self.media: dict[str, Any] = action_data.get(ATTR_MEDIA) or {} 

171 self.debug: bool = action_data.get(ATTR_DEBUG, False) 

172 self.actions: list[dict[str, Any]] = ensure_list(action_data.get(ATTR_ACTIONS)) 

173 

174 self.selected_deliveries: dict[str, dict[str, Any]] = {} 

175 self.enabled_scenarios: dict[str, Scenario] = {} 

176 self.selected_scenario_names: list[str] = [] 

177 self._suppression_reason: SuppressionReason | None = None 

178 self._raw_image_path: Any = None 

179 self._delivery_error: list[str] | None = None 

180 self.condition_variables: ConditionVariables 

181 

182 async def initialize(self) -> None: 

183 """Async post-construction initialization""" 

184 self.occupancy: dict[str, list[Recipient]] = self.people_registry.determine_occupancy() 

185 self.condition_variables = ConditionVariables( 

186 self.applied_scenario_names, 

187 self.required_scenario_names, 

188 self.constrain_scenario_names, 

189 self.priority, 

190 self.occupancy, 

191 self.message, 

192 self._title, 

193 self.extra_data, 

194 ) # requires occupancy first 

195 

196 enabled_scenario_names: list[str] = list(self.applied_scenario_names) or [] 

197 self.selected_scenario_names = await self.select_scenarios() 

198 enabled_scenario_names.extend(self.selected_scenario_names) 

199 if self.constrain_scenario_names: 

200 enabled_scenario_names = [ 

201 s for s in enabled_scenario_names if (s in self.constrain_scenario_names or s in self.applied_scenario_names) 

202 ] 

203 if self.required_scenario_names and not any(s in enabled_scenario_names for s in self.required_scenario_names): 

204 _LOGGER.info("SUPERNOTIFY suppressing notification, no required scenarios enabled") 

205 self.selected_deliveries = {} 

206 self.suppress(SuppressionReason.NO_SCENARIO) 

207 else: 

208 for s in enabled_scenario_names: 

209 scenario_obj = self.context.scenario_registry.scenarios.get(s) 

210 if scenario_obj is not None: 

211 self.enabled_scenarios[s] = scenario_obj 

212 

213 self.selected_deliveries = self.select_deliveries() 

214 if self.context.snoozer.is_global_snooze(self.priority): 

215 self.suppress(SuppressionReason.SNOOZED) 

216 self.apply_enabled_scenarios() 

217 

218 if not self.media: 

219 self.media = self.media_requirements(self.extra_data) 

220 

221 def outcome(self) -> Outcome: 

222 if self.error_count > 0: 

223 return Outcome.ERROR 

224 if self.dupe: 

225 return Outcome.DUPE 

226 if not self.delivered: 

227 return Outcome.NO_DELIVERY 

228 if self.fallback: 

229 return Outcome.FALLBACK_DELIVERY 

230 if self.skipped: 

231 return Outcome.PARTIAL_DELIVERY 

232 return Outcome.SUCCESS 

233 

234 def media_requirements(self, data: dict[str, Any]) -> dict[str, Any]: 

235 """If no media defined, look for iOS / Android actions that have media defined 

236 

237 Example is the Frigate blueprint, which generates `image`, `video` etc 

238 in the `data` section, that can also be used for email attachments 

239 """ 

240 media_dict = {} 

241 if not data: 

242 return {} 

243 if data.get(ATTR_IMAGE): 

244 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = data.get(ATTR_IMAGE) 

245 if data.get(ATTR_VIDEO): 

246 media_dict[ATTR_MEDIA_CLIP_URL] = data.get(ATTR_VIDEO) 

247 if data.get("attachment", {}).get("url"): 

248 url = data["attachment"]["url"] 

249 if url and url.endswith(".mp4") and not media_dict.get(ATTR_MEDIA_CLIP_URL): 

250 media_dict[ATTR_MEDIA_CLIP_URL] = url 

251 elif ( 

252 url 

253 and (url.endswith(".jpg") or url.endswith(".jpeg") or url.endswith(".png")) 

254 and not media_dict.get(ATTR_MEDIA_SNAPSHOT_URL) 

255 ): 

256 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = url 

257 return media_dict 

258 

259 def validate_action_data(self, action_data: dict[str, Any]) -> None: 

260 if action_data.get(ATTR_PRIORITY): 

261 if isinstance(action_data.get(ATTR_PRIORITY), (str, int, float)): 

262 if action_data.get(ATTR_PRIORITY) not in PRIORITY_VALUES: 

263 _LOGGER.info("SUPERNOTIFY custom priority %s", action_data.get(ATTR_PRIORITY)) 

264 else: 

265 _LOGGER.info("SUPERNOTIFY Invalid priority %s", action_data.get(ATTR_PRIORITY)) 

266 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

267 raise vol.Invalid("Priority value must be a simple value") 

268 try: 

269 humanize.validate_with_humanized_errors(action_data, ACTION_DATA_SCHEMA) 

270 except vol.Invalid as e: 

271 _LOGGER.warning("SUPERNOTIFY invalid action data %s: %s", action_data, e) 

272 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

273 raise 

274 except vol.error.Error as e2: 

275 _LOGGER.warning("SUPERNOTIFY failed to validate action data %s: %s", action_data, e2) 

276 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

277 raise vol.Invalid(f"Unable to validate action data - {e2}") from e2 

278 

279 def apply_enabled_scenarios(self) -> None: 

280 """Set media and action_groups from scenario if defined, first come first applied""" 

281 action_groups: list[str] = [] 

282 for scenario in self.enabled_scenarios.values(): 

283 if scenario.media: 

284 if self.media: 

285 self.media.update(scenario.media) 

286 else: 

287 self.media = scenario.media 

288 if scenario.action_groups: 

289 action_groups.extend(ag for ag in scenario.action_groups if ag not in action_groups) 

290 # self.action_groups only accessed from inside Envelope 

291 if self.action_groups: 

292 self.action_groups.extend(action_groups) 

293 else: 

294 self.action_groups = action_groups 

295 

296 def select_deliveries(self) -> dict[str, dict[str, Any]]: 

297 scenario_enable_deliveries: list[str] = [] 

298 scenario_disable_deliveries: list[str] = [] 

299 default_enable_deliveries: list[str] = [] 

300 recipients_enable_deliveries: list[str] = [] 

301 

302 if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

303 for scenario in self.enabled_scenarios.values(): 

304 scenario_enable_deliveries.extend(scenario.enabling_deliveries()) 

305 for scenario in self.enabled_scenarios.values(): 

306 scenario_disable_deliveries.extend(scenario.disabling_deliveries()) 

307 

308 scenario_enable_deliveries = list(set(scenario_enable_deliveries)) 

309 scenario_disable_deliveries = list(set(scenario_disable_deliveries)) 

310 

311 for recipient in self.all_recipients(): 

312 recipients_enable_deliveries.extend(recipient.enabling_delivery_names()) 

313 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT: 

314 # all deliveries with SELECTION_DEFAULT in CONF_SELECTION 

315 default_enable_deliveries = [d.name for d in self.context.delivery_registry.implicit_deliveries] 

316 

317 self.debug_trace.record_delivery_selection("scenario_enable_deliveries", scenario_enable_deliveries) 

318 self.debug_trace.record_delivery_selection("scenario_disable_deliveries", scenario_disable_deliveries) 

319 self.debug_trace.record_delivery_selection("default_enable_deliveries", default_enable_deliveries) 

320 self.debug_trace.record_delivery_selection("recipient_enable_deliveries", recipients_enable_deliveries) 

321 

322 override_enable_deliveries: list[str] = [] 

323 override_disable_deliveries: list[str] = [] 

324 

325 # apply the deliveries defined in the notification action call 

326 for delivery, delivery_override in self.delivery_overrides.items(): 

327 if ( 

328 (delivery_override is None or delivery_override.enabled is True) 

329 and delivery in self.context.delivery_registry.enabled_deliveries 

330 ) or ( 

331 (delivery_override is not None and delivery_override.enabled is True) 

332 and delivery in self.context.delivery_registry.disabled_deliveries 

333 ): 

334 override_enable_deliveries.append(delivery) 

335 elif delivery_override is not None and delivery_override.enabled is False: 

336 override_disable_deliveries.append(delivery) 

337 

338 # if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

339 # scenario_disable_deliveries = [ 

340 # d.name 

341 # for d in self.context.delivery_registry.deliveries.values() 

342 # if d.selection == [SELECTION_BY_SCENARIO] 

343 # and d.name not in scenario_enable_deliveries 

344 # and (d.name not in override_enable_deliveries or self.delivery_selection != DELIVERY_SELECTION_EXPLICIT) 

345 # ] 

346 all_global_enabled: list[str] = list( 

347 set(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries) 

348 ) 

349 all_enabled: list[str] = all_global_enabled + recipients_enable_deliveries 

350 # override_enable_deliveries takes precedence: if the action call explicitly 

351 # re-enables a delivery that a scenario disabled, remove it from all_disabled. 

352 all_disabled: list[str] = [ 

353 d for d in scenario_disable_deliveries + override_disable_deliveries if d not in override_enable_deliveries 

354 ] 

355 override_enabled: list[str] = list(set(scenario_enable_deliveries + override_enable_deliveries)) 

356 self.debug_trace.record_delivery_selection("override_disable_deliveries", override_disable_deliveries) 

357 self.debug_trace.record_delivery_selection("override_enable_deliveries", override_enable_deliveries) 

358 

359 unsorted_maybe_objs: list[Delivery | None] = [ 

360 self.delivery_registry.deliveries.get(d) for d in all_enabled if d not in all_disabled 

361 ] 

362 unsorted_objs: list[Delivery] = [ 

363 d for d in unsorted_maybe_objs if d is not None and (d.enabled or d.name in override_enabled) 

364 ] 

365 first: list[str] = [d.name for d in unsorted_objs if d.selection_rank == SelectionRank.FIRST] 

366 anywhere: list[str] = [d.name for d in unsorted_objs if d.selection_rank == SelectionRank.ANY] 

367 last: list[str] = [d.name for d in unsorted_objs if d.selection_rank == SelectionRank.LAST] 

368 selected = first + anywhere + last 

369 self.debug_trace.record_delivery_selection("ranked", selected) 

370 

371 # TODO: clean up this ugly logic, reorganize delivery around people 

372 results: dict[str, dict[str, Any]] = {d: {} for d in selected} 

373 personal_deliveries = [d for d in selected if d not in all_global_enabled and d in recipients_enable_deliveries] 

374 for personal_delivery in personal_deliveries: 

375 results[personal_delivery].setdefault("recipients", []) 

376 for recipient in self.all_recipients(): 

377 if personal_delivery in recipient.enabling_delivery_names(): 

378 results[personal_delivery]["recipients"].append(recipient.entity_id) 

379 return results 

380 

381 def suppress(self, reason: SuppressionReason) -> None: 

382 self._suppression_reason = reason 

383 if reason not in self._skip_reasons: 

384 self._skip_reasons.append(reason) 

385 _LOGGER.info("SUPERNOTIFY Suppressing notification, reason:%s, id:%s", reason, self.id) 

386 

387 async def deliver(self) -> bool: 

388 _LOGGER.debug( 

389 "Message: %s, notification: %s, deliveries: %s", 

390 self.message, 

391 self.id, 

392 self.selected_deliveries, 

393 ) 

394 

395 for delivery_name in self.selected_deliveries: 

396 self.deliveries[delivery_name] = {} 

397 

398 if self._suppression_reason is not None: 

399 _LOGGER.info("SUPERNOTIFY Suppressing globally silenced/snoozed notification (%s)", self.id) 

400 for delivery_name in self.selected_deliveries: 

401 delivery = self.context.delivery_registry.deliveries.get(delivery_name) 

402 self.record_result(delivery, suppression_reason=SuppressionReason.SNOOZED) 

403 else: 

404 # Deliveries for transports that call grab_image() are deferred so that 

405 # PTZ movement runs concurrently with non-image deliveries (chime, TTS, etc.) 

406 camera_configured = bool(self.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID) or self.media.get(ATTR_MEDIA_SNAPSHOT_URL)) 

407 immediate_deliveries: dict[str, dict[str, Any]] = {} 

408 deferred_deliveries: dict[str, dict[str, Any]] = {} 

409 for delivery_name, details in self.selected_deliveries.items(): 

410 d = self.context.delivery_registry.deliveries.get(delivery_name) 

411 if d and camera_configured and d.transport.supported_features & TransportFeature.SNAPSHOT_IMAGE: 

412 deferred_deliveries[delivery_name] = details 

413 else: 

414 immediate_deliveries[delivery_name] = details 

415 

416 # Start image grab immediately so PTZ runs while immediate deliveries execute 

417 image_task: asyncio.Task | None = None 

418 if deferred_deliveries: 

419 image_task = asyncio.create_task(_snap_notification_image(self, self.context)) 

420 

421 _LOGGER.debug("SUPERNOTIFY Scheduling %s immediate deliveries", len(deferred_deliveries)) 

422 await self._schedule_deliveries(immediate_deliveries) 

423 

424 # Ensure image is ready before running image-requiring deliveries 

425 if image_task is not None: 

426 wait_timeout: int = 30 

427 try: 

428 _LOGGER.debug("SUPERNOTIFY Waiting up to %s for image grab to complete", wait_timeout) 

429 async with asyncio.timeout(wait_timeout): # TODO: configurable time-out 

430 await image_task 

431 except Exception as e: 

432 _LOGGER.exception("SUPERNOTIFY Failed to pre-grab image: %s", e) 

433 

434 _LOGGER.debug("SUPERNOTIFY Scheduling %s deferred deliveries", len(deferred_deliveries)) 

435 await self._schedule_deliveries(deferred_deliveries) 

436 

437 if self.delivered == 0 and not self._suppression_reason: 

438 if self.failed == 0 and not self.dupe: 

439 for delivery in self.context.delivery_registry.fallback_by_default_deliveries: 

440 _LOGGER.info( 

441 "SUPERNOTIFY no delivery succeeded, activating fallback_by_default: %s", 

442 delivery.name, 

443 ) 

444 if delivery.name not in self.selected_deliveries: 

445 await self.call_transport(delivery) 

446 self.fallback += 1 

447 

448 if self.failed > 0: 

449 for delivery in self.context.delivery_registry.fallback_on_error_deliveries: 

450 _LOGGER.warning( 

451 "SUPERNOTIFY delivery failed, activating fallback_on_error: %s", 

452 delivery.name, 

453 ) 

454 if delivery.name not in self.selected_deliveries: 

455 await self.call_transport(delivery) 

456 self.fallback += 1 

457 

458 return self.delivered > 0 

459 

460 async def _schedule_deliveries(self, deliveries: dict[str, dict[str, Any]]) -> None: 

461 delivery_coros = [] 

462 for delivery_name, details in deliveries.items(): 

463 delivery = self.context.delivery_registry.deliveries.get(delivery_name) 

464 if delivery: 

465 delivery_coros.append(self.call_transport(delivery, recipients=details.get("recipients"))) 

466 else: 

467 _LOGGER.error("SUPERNOTIFY Unexpected missing delivery %s", delivery_name) 

468 if delivery_coros: 

469 results = await asyncio.gather(*delivery_coros, return_exceptions=True) 

470 for result in results: 

471 if isinstance(result, BaseException): 

472 _LOGGER.error("SUPERNOTIFY Unexpected error in parallel delivery: %s", result) 

473 

474 async def call_transport(self, delivery: Delivery, recipients: list[str] | None = None) -> None: 

475 try: 

476 transport: Transport = delivery.transport 

477 if not transport.enabled: 

478 self.record_result(delivery, suppression_reason=SuppressionReason.TRANSPORT_DISABLED) 

479 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on transport disabled", delivery) 

480 return 

481 

482 delivery_priorities: list[str] = delivery.priority 

483 if self.priority and delivery_priorities and self.priority not in delivery_priorities: 

484 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on priority (%s)", delivery, self.priority) 

485 self.record_result(delivery, suppression_reason=SuppressionReason.PRIORITY) 

486 return 

487 if not delivery.evaluate_conditions(self.condition_variables): 

488 _LOGGER.debug("SUPERNOTIFY Skipping delivery %s based on conditions", delivery) 

489 self.record_result(delivery, suppression_reason=SuppressionReason.DELIVERY_CONDITION) 

490 return 

491 

492 targets: list[Target] = self.generate_targets(delivery, recipients=recipients) 

493 envelopes: list[Envelope] = self.generate_envelopes(delivery, targets) 

494 if not envelopes: 

495 if delivery.target_required == TargetRequired.ALWAYS and ( 

496 not targets or not any(t.has_resolved_target() for t in targets) 

497 ): 

498 reason: SuppressionReason = SuppressionReason.NO_TARGET 

499 else: 

500 reason = SuppressionReason.UNKNOWN 

501 self.record_result(delivery, targets=targets, suppression_reason=reason) 

502 

503 for envelope in envelopes: 

504 if not self.force_resend and self.context.dupe_checker.check(envelope): 

505 _LOGGER.debug("SUPERNOTIFY Suppressing dupe envelope, %s", self.message) 

506 self.record_result(delivery, envelope, suppression_reason=SuppressionReason.DUPE) 

507 continue 

508 try: 

509 if not await transport.deliver(envelope, debug_trace=self.debug_trace): 

510 _LOGGER.info( 

511 "SUPERNOTIFY No delivery for %s (targets: %s)", 

512 delivery.name, 

513 envelope.target.as_dict() if envelope.target else "NONE", 

514 ) 

515 self.record_result(delivery, envelope) 

516 except Exception as e2: 

517 _LOGGER.exception("SUPERNOTIFY Failed to deliver %s: %s", delivery.name, e2) 

518 envelope.error_count = envelope.error_count + 1 

519 transport.record_error(str(e2), method="deliver") 

520 envelope.delivery_error = format_exception(e2) 

521 self.record_result(delivery, envelope) 

522 

523 except Exception as e: 

524 _LOGGER.exception( 

525 "SUPERNOTIFY Failed to notify using delivery %s via %s", 

526 delivery.name, 

527 type(delivery.transport).__name__, 

528 ) 

529 self.deliveries.setdefault(delivery.name, {}) 

530 self.deliveries[delivery.name].setdefault("errors", []) 

531 errors: list[str] = cast("list[str]", self.deliveries[delivery.name]["errors"]) 

532 errors.append("\n".join(format_exception(e))) 

533 

534 def record_result( 

535 self, 

536 delivery: Delivery | None, 

537 envelope: Envelope | None = None, 

538 targets: list[Target] | None = None, 

539 suppression_reason: SuppressionReason | None = None, 

540 ) -> None: 

541 """Debugging (and unit test) support for notifications that failed or were skipped""" 

542 if delivery: 

543 if envelope: 

544 self.delivered += envelope.delivered 

545 self.error_count += envelope.error_count 

546 self.deliveries.setdefault(delivery.name, {}) 

547 if envelope.delivered: 

548 self.deliveries[delivery.name].setdefault(KEY_DELIVERED, []) 

549 self.deliveries[delivery.name][KEY_DELIVERED].append(envelope) # type: ignore 

550 else: 

551 if suppression_reason: 

552 envelope.skip_reason = suppression_reason 

553 if suppression_reason not in self._skip_reasons: 

554 self._skip_reasons.append(suppression_reason) 

555 if suppression_reason == SuppressionReason.DUPE: 

556 self.dupe = True 

557 if envelope.error_count: 

558 self.deliveries[delivery.name].setdefault(KEY_FAILED, []) 

559 self.deliveries[delivery.name][KEY_FAILED].append(envelope) # type: ignore 

560 self.failed += 1 

561 else: 

562 self.deliveries[delivery.name].setdefault(KEY_SUPPRESSED, []) 

563 self.deliveries[delivery.name][KEY_SUPPRESSED].append(envelope) # type: ignore 

564 self.suppressed += 1 

565 

566 if not envelope: 

567 delivery_name: str = delivery.name if delivery else "!UNKNOWN!" 

568 skip_summary: dict[str, Any] = { 

569 "target_required": delivery.target_required if delivery else "!UNKNOWN!", 

570 "suppression_reason": str(suppression_reason), 

571 } 

572 self.deliveries.setdefault(delivery_name, {}) 

573 if targets: 

574 skip_summary["targets"] = targets 

575 self.deliveries[delivery_name][KEY_SKIPPED] = skip_summary 

576 self.skipped += 1 

577 

578 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> dict[str, Any]: 

579 """ArchiveableObject implementation""" 

580 minimal = not diagnostics 

581 object_refs = ["context", "people_registry", "delivery_registry"] 

582 keys_only = ["enabled_scenarios"] 

583 debug_only = ["debug_trace"] 

584 exposed_if_populated = ["_delivery_error", "message_html", "extra_data", "actions", "_suppression_reason"] 

585 # fine tune dict order to ease the eye-burden when reviewing archived notifications 

586 preferred_order = [ 

587 "id", 

588 "created", 

589 "message", 

590 "applied_scenario_names", 

591 "constrain_scenario_names", 

592 "required_scenario_names", 

593 "enabled_scenarios", 

594 "selected_scenario_names", 

595 "delivery_selection", 

596 "delivery_overrides", 

597 "delivery_selection", 

598 "selected_deliveries", 

599 "recipients_override", 

600 "delivered", 

601 "failed", 

602 "suppressed", 

603 "skipped", 

604 "error_count", 

605 "deliveries", 

606 ] 

607 # preferred fields 

608 result: dict[str, Any] = {"version": _VERSION} 

609 result.update({ 

610 k: sanitize( 

611 self.__dict__[k], minimal=minimal, occupancy_only=True, top_level_keys_only=(minimal and k in keys_only) 

612 ) 

613 for k in preferred_order 

614 }) 

615 # all the rest not explicitly excluded 

616 result.update({ 

617 k: sanitize(v, minimal=minimal, occupancy_only=True) 

618 for k, v in self.__dict__.items() 

619 if k not in result 

620 and k not in exposed_if_populated 

621 and k not in object_refs 

622 and not k.startswith("_") 

623 and (not minimal or k not in keys_only) 

624 and (not minimal or k not in debug_only) 

625 }) 

626 # the exposed only if populated fields 

627 result.update({ 

628 k: sanitize(self.__dict__[k], minimal=minimal, occupancy_only=True) 

629 for k in exposed_if_populated 

630 if self.__dict__.get(k) 

631 }) 

632 # delivery_stats: aggregate delivery metrics 

633 try: 

634 all_durations: dict[str, float] = {} 

635 total_ok = 0 

636 total_all = 0 

637 for d_name, outcomes in self.deliveries.items(): 

638 for envelope in outcomes.get(KEY_DELIVERED, []): 

639 dur = sum(c.contents().get("elapsed", 0) for c in getattr(envelope, "calls", [])) * 1000 

640 all_durations[d_name] = dur 

641 total_ok += 1 

642 total_all += 1 

643 for _envelope in outcomes.get(KEY_FAILED, []): 

644 all_durations.setdefault(d_name, 0) 

645 total_all += 1 

646 if outcomes.get(KEY_SKIPPED): 

647 total_all += 1 

648 if all_durations: 

649 result["stats"] = { 

650 "total_duration_ms": round(sum(all_durations.values()), 1), 

651 "slowest_delivery": max(all_durations, key=lambda k: all_durations[k]), 

652 "fastest_delivery": min(all_durations, key=lambda k: all_durations[k]), 

653 "delivery_success_rate": round(total_ok / total_all, 2) if total_all else 1.0, 

654 } 

655 except Exception as e: 

656 _LOGGER.warning("SUPERNOTIFY delivery_stats computation failed: %s", e) 

657 return result 

658 

659 def base_filename(self) -> str: 

660 """ArchiveableObject implementation""" 

661 return f"{self.created.isoformat()[:16].replace(':', '-')}_{self.id}" 

662 

663 def delivery_data(self, delivery: Delivery) -> dict[str, Any]: 

664 if delivery is None: 

665 return {} 

666 delivery_override: DeliveryCustomization | None = self.delivery_overrides.get(delivery.name) 

667 if delivery_override is None: 

668 delivery_override = self.delivery_overrides.get(delivery.transport.name) 

669 return delivery_override.data if delivery_override and delivery_override.data else {} 

670 

671 @property 

672 def delivered_envelopes(self) -> list[Envelope]: 

673 result: list[Envelope] = [] 

674 for delivery_result in self.deliveries.values(): 

675 result.extend(cast("list[Envelope]", delivery_result.get(KEY_DELIVERED, [])))