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

1from __future__ import annotations 

2 

3import asyncio 

4import datetime as dt 

5import logging 

6import uuid 

7from dataclasses import dataclass 

8from traceback import format_exception 

9from typing import TYPE_CHECKING, Any, cast 

10 

11import homeassistant.util.dt as dt_util 

12import voluptuous as vol 

13from homeassistant.components.notify.const import ATTR_DATA 

14from voluptuous import humanize 

15 

16from custom_components.supernotify.schema import SelectionRank 

17 

18from .archive import ArchivableObject 

19from .common import ensure_list, nullable_ensure_list, sanitize 

20from .const import ( 

21 ATTR_ACTION_GROUPS, 

22 ATTR_ACTIONS, 

23 ATTR_DEBUG, 

24 ATTR_DELIVERY, 

25 ATTR_DELIVERY_SELECTION, 

26 ATTR_FORCE_RESEND, 

27 ATTR_IMAGE, 

28 ATTR_MEDIA, 

29 ATTR_MEDIA_CAMERA_ENTITY_ID, 

30 ATTR_MEDIA_CLIP_URL, 

31 ATTR_MEDIA_SNAPSHOT_URL, 

32 ATTR_MESSAGE_HTML, 

33 ATTR_PERSON_ID, 

34 ATTR_PRIORITY, 

35 ATTR_RECIPIENTS, 

36 ATTR_SCENARIOS_APPLY, 

37 ATTR_SCENARIOS_CONSTRAIN, 

38 ATTR_SCENARIOS_REQUIRE, 

39 ATTR_SPOKEN_MESSAGE, 

40 ATTR_VIDEO, 

41 DELIVERY_SELECTION_EXPLICIT, 

42 DELIVERY_SELECTION_FIXED, 

43 DELIVERY_SELECTION_IMPLICIT, 

44 OPTION_UNIQUE_TARGETS, 

45 PRIORITY_MEDIUM, 

46 PRIORITY_VALUES, 

47 TARGET_USE_FIXED, 

48 TARGET_USE_MERGE_ALWAYS, 

49 TARGET_USE_MERGE_ON_DELIVERY_TARGETS, 

50 TARGET_USE_ON_NO_ACTION_TARGETS, 

51 TARGET_USE_ON_NO_DELIVERY_TARGETS, 

52) 

53from .envelope import Envelope 

54from .media_grab import snap_notification_image as _snap_notification_image 

55from .model import ( 

56 ConditionVariables, 

57 DebugTrace, 

58 DeliveryCustomization, 

59 SuppressionReason, 

60 Target, 

61 TargetRequired, 

62 TransportFeature, 

63) 

64from .schema import ACTION_DATA_SCHEMA, STRICT_ACTION_DATA_SCHEMA, DeliveryOutcome, EnvelopeOutcome 

65 

66if TYPE_CHECKING: 

67 from .context import Context 

68 from .delivery import Delivery, DeliveryRegistry 

69 from .people import PeopleRegistry, Recipient 

70 from .scenario import Scenario 

71 from .transport import ( 

72 Transport, 

73 ) 

74 

75_LOGGER = logging.getLogger(__name__) 

76 

77_VERSION: str = "unknown" 

78 

79 

80def set_version(version: str) -> None: 

81 """Called once from async_setup_entry with the integration's manifest version. 

82 

83 Avoids a blocking manifest.json read on the event loop at import time. 

84 """ 

85 global _VERSION 

86 _VERSION = version 

87 

88 

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

90INTERNAL_DATA_KEYS = (ATTR_FORCE_RESEND, ATTR_SPOKEN_MESSAGE) 

91 

92type DeliveryName = str 

93 

94 

95@dataclass 

96class DeliveryTargetOverride: 

97 fixed: list[str] | None = None 

98 include: list[str] | None = None 

99 exclude: list[str] | None = None 

100 

101 def as_dict(self) -> dict[str, Any]: 

102 return {"fixed": self.fixed, "include": self.include, "exclude": self.exclude} 

103 

104 

105class Notification(ArchivableObject): 

106 def __init__( 

107 self, 

108 context: Context, 

109 message: str | None = None, 

110 title: str | None = None, 

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

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

113 ) -> None: 

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

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

116 self.message: str | None = message 

117 self.context: Context = context 

118 self.people_registry: PeopleRegistry = context.people_registry 

119 self.delivery_registry: DeliveryRegistry = context.delivery_registry 

120 action_data = action_data or {} 

121 

122 if action_data.get(ATTR_RECIPIENTS): 

123 _LOGGER.warning("SUPERNOTIFY recipients key in notify action data deprecated, list recipients in target instead") 

124 target = ensure_list(target) + ensure_list(action_data.get(ATTR_RECIPIENTS)) 

125 

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

127 self._already_selected: Target = Target() 

128 self._title: str | None = title 

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

130 self.delivered: int = 0 

131 self.error_count: int = 0 

132 self.skipped: int = 0 

133 self.failed: int = 0 

134 self.suppressed: int = 0 

135 self.fallback: int = 0 

136 self.dupe: bool = False 

137 self.deliveries: dict[DeliveryName, dict[EnvelopeOutcome, list[str] | list[Envelope] | dict[str, Any]]] = {} 

138 self.delivery_exceptions: dict[DeliveryName, list[str]] 

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

140 

141 self.validate_action_data(action_data) 

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

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

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

145 } 

146 

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

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

149 

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

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

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

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

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

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

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

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

158 

159 delivery_data = action_data.get(ATTR_DELIVERY) 

160 if isinstance(delivery_data, list): 

161 # a bare list of deliveries implies intent to restrict 

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

163 if self.delivery_selection is None: 

164 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

165 self.delivery_overrides = { 

166 k: DeliveryCustomization(config=None, default_enabled=True) for k in action_data.get(ATTR_DELIVERY, []) 

167 } 

168 elif isinstance(delivery_data, str) and delivery_data: 

169 # a bare list of deliveries implies intent to restrict 

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

171 if self.delivery_selection is None: 

172 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

173 self.delivery_overrides = {delivery_data: DeliveryCustomization(config=None, default_enabled=True)} 

174 elif isinstance(delivery_data, dict): 

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

176 if self.delivery_selection is None: 

177 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

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

179 self.delivery_overrides = { 

180 k: DeliveryCustomization(config={**(v or {})}, default_enabled=True) 

181 for k, v in action_data.get(ATTR_DELIVERY, {}).items() 

182 } 

183 elif delivery_data: 

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

185 if self.delivery_selection is None: 

186 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

187 else: 

188 if self.delivery_selection is None: 

189 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

190 

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

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

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

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

195 

196 self.selected_deliveries: dict[str, DeliveryTargetOverride | None] = {} 

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

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

199 self._suppression_reason: SuppressionReason | None = None 

200 self._raw_image_path: Any = None 

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

202 self.condition_variables: ConditionVariables 

203 

204 async def initialize(self) -> None: 

205 """Async post-construction initialization""" 

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

207 self.condition_variables = ConditionVariables( 

208 self.applied_scenario_names, 

209 self.required_scenario_names, 

210 self.constrain_scenario_names, 

211 self.priority, 

212 self.occupancy, 

213 self.message, 

214 self._title, 

215 self.extra_data, 

216 ) # requires occupancy first 

217 

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

219 self.selected_scenario_names = await self.select_scenarios() 

220 enabled_scenario_names.extend(self.selected_scenario_names) 

221 if self.constrain_scenario_names: 

222 enabled_scenario_names = [ 

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

224 ] 

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

226 _LOGGER.info("SUPERNOTIFY Suppressing notification, no required scenarios enabled") 

227 self.selected_deliveries = {} 

228 self.suppress(SuppressionReason.NO_SCENARIO) 

229 else: 

230 for s in enabled_scenario_names: 

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

232 if scenario_obj is not None: 

233 self.enabled_scenarios[s] = scenario_obj 

234 

235 self.selected_deliveries = self.select_deliveries() 

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

237 self.suppress(SuppressionReason.SNOOZED) 

238 self.apply_enabled_scenarios() 

239 

240 if not self.media: 

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

242 

243 def convert_notify_entities(self, target: list[str] | str | None = None) -> list[str] | str | None: 

244 """Short circuit supernotify notify entities so they're handled directly so not 

245 going round in circles via calls to notify.send_message 

246 

247 Defined here rather than in models/Target since requires access to registries 

248 

249 """ 

250 if not target: 

251 return target 

252 updated: list[str] = [] 

253 changed: bool = False 

254 known_entities: dict[str, Recipient] = self.people_registry.notify_entities() 

255 for t in ensure_list(target): 

256 if t in known_entities: 

257 # e.g. switch to a person id if a notify entity for recipient 

258 updated.append(known_entities[t].entity_id) 

259 changed = True 

260 else: 

261 updated.append(t) 

262 

263 if changed: 

264 return updated 

265 return target 

266 

267 def outcome(self) -> DeliveryOutcome: 

268 if self.error_count > 0: 

269 return DeliveryOutcome.ERROR 

270 if self.dupe: 

271 return DeliveryOutcome.DUPE 

272 if not self.delivered: 

273 return DeliveryOutcome.NO_DELIVERY 

274 if self.fallback: 

275 return DeliveryOutcome.FALLBACK_DELIVERY 

276 if self.skipped: 

277 return DeliveryOutcome.PARTIAL_DELIVERY 

278 return DeliveryOutcome.SUCCESS 

279 

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

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

282 

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

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

285 """ 

286 media_dict = {} 

287 if not data: 

288 return {} 

289 if data.get(ATTR_IMAGE): 

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

291 if data.get(ATTR_VIDEO): 

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

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

294 url = data["attachment"]["url"] 

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

296 media_dict[ATTR_MEDIA_CLIP_URL] = url 

297 elif url and (url.endswith((".jpg", ".jpeg", ".png"))) and not media_dict.get(ATTR_MEDIA_SNAPSHOT_URL): 

298 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = url 

299 return media_dict 

300 

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

302 if action_data.get(ATTR_PRIORITY): 

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

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

305 _LOGGER.info("SUPERNOTIFY Custom priority %s", action_data.get(ATTR_PRIORITY)) 

306 else: 

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

308 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

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

310 try: 

311 humanize.validate_with_humanized_errors(action_data, ACTION_DATA_SCHEMA) # type: ignore[arg-type] 

312 except vol.Invalid as e: 

313 _LOGGER.warning("SUPERNOTIFY Invalid action data %s: %s", action_data, e) 

314 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

315 raise 

316 except vol.error.Error as e2: 

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

318 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

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

320 

321 def apply_enabled_scenarios(self) -> None: 

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

323 action_groups: list[str] = [] 

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

325 if scenario.media: 

326 if self.media: 

327 self.media.update(scenario.media) 

328 else: 

329 self.media = scenario.media 

330 if scenario.action_groups: 

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

332 # self.action_groups only accessed from inside Envelope 

333 if self.action_groups: 

334 self.action_groups.extend(action_groups) 

335 else: 

336 self.action_groups = action_groups 

337 

338 def select_deliveries(self) -> dict[str, DeliveryTargetOverride | None]: 

339 scenario_enable_deliveries: list[str] = [] 

340 scenario_disable_deliveries: list[str] = [] 

341 default_enable_deliveries: list[str] = [] 

342 recipients_enable_deliveries: list[str] = [] 

343 recipients_disable_deliveries: list[str] = [] 

344 

345 if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

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

347 scenario_enable_deliveries.extend(scenario.enabling_deliveries()) 

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

349 scenario_disable_deliveries.extend(scenario.disabling_deliveries()) 

350 

351 scenario_enable_deliveries = list(set(scenario_enable_deliveries)) 

352 scenario_disable_deliveries = list(set(scenario_disable_deliveries)) 

353 

354 for recipient in self.all_recipients(): 

355 recipients_enable_deliveries.extend(recipient.enabling_delivery_names()) 

356 # See also Recipient.target() for the disabled cases 

357 recipients_disable_deliveries.extend(recipient.disabling_delivery_names()) 

358 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT: 

359 # all deliveries with SELECTION_DEFAULT in CONF_SELECTION 

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

361 

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

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

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

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

366 

367 override_enable_deliveries: list[str] = [] 

368 override_disable_deliveries: list[str] = [] 

369 

370 # apply the deliveries defined in the notification action call 

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

372 if ( 

373 (delivery_override is None or delivery_override.enabled is not False) 

374 and delivery in self.context.delivery_registry.enabled_deliveries 

375 ) or ( 

376 (delivery_override is not None and delivery_override.enabled is not False) 

377 and delivery in self.context.delivery_registry.disabled_deliveries 

378 ): 

379 override_enable_deliveries.append(delivery) 

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

381 override_disable_deliveries.append(delivery) 

382 

383 # if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

384 # scenario_disable_deliveries = [ 

385 # d.name 

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

387 # if d.selection == [SELECTION_BY_SCENARIO] 

388 # and d.name not in scenario_enable_deliveries 

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

390 # ] 

391 all_global_enabled: list[str] = list( 

392 set(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries) 

393 ) 

394 all_enabled: list[str] = all_global_enabled + recipients_enable_deliveries 

395 # override_enable_deliveries takes precedence: if the action call explicitly 

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

397 all_disabled: list[str] = [ 

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

399 ] 

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

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

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

403 

404 unsorted_maybe_objs: list[Delivery | None] = [ 

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

406 ] 

407 unsorted_objs: list[Delivery] = [ 

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

409 ] 

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

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

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

413 selected = first + anywhere + last 

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

415 

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

417 selected_deliveries: dict[str, DeliveryTargetOverride | None] = {d: None for d in selected} 

418 personal_deliveries = [d for d in selected if d in recipients_enable_deliveries] 

419 for personal_delivery in personal_deliveries: 

420 fixed_targets: list[str] = [] 

421 include_targets: list[str] = [] 

422 exclude_targets: list[str] = [] 

423 for recipient in self.all_recipients(): 

424 if personal_delivery in recipient.enabling_delivery_names(): 

425 if personal_delivery in all_global_enabled: 

426 # delivery already selected, make sure this recipient also included 

427 include_targets.append(recipient.entity_id) 

428 else: 

429 # this delivery has been explicitly enabled for individual recipients 

430 fixed_targets.append(recipient.entity_id) 

431 elif personal_delivery in recipient.disabling_delivery_names(): 

432 exclude_targets.append(recipient.entity_id) 

433 selected_deliveries[personal_delivery] = DeliveryTargetOverride( 

434 fixed=fixed_targets, exclude=exclude_targets, include=include_targets 

435 ) 

436 return selected_deliveries 

437 

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

439 self._suppression_reason = reason 

440 if reason not in self._skip_reasons: 

441 self._skip_reasons.append(reason) 

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

443 

444 async def deliver(self) -> bool: 

445 _LOGGER.debug( 

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

447 self.message, 

448 self.id, 

449 self.selected_deliveries, 

450 ) 

451 

452 for delivery_name in self.selected_deliveries: 

453 self.deliveries[delivery_name] = {} 

454 

455 if self._suppression_reason is not None: 

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

457 for delivery_name in self.selected_deliveries: 

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

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

460 else: 

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

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

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

464 immediate_deliveries: dict[str, DeliveryTargetOverride | None] = {} 

465 deferred_deliveries: dict[str, DeliveryTargetOverride | None] = {} 

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

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

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

469 deferred_deliveries[delivery_name] = details 

470 else: 

471 immediate_deliveries[delivery_name] = details 

472 

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

474 image_task: asyncio.Task | None = None 

475 if deferred_deliveries: 

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

477 

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

479 await self._schedule_deliveries(immediate_deliveries) 

480 

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

482 if image_task is not None: 

483 wait_timeout: int = 30 

484 try: 

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

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

487 await image_task 

488 except Exception: 

489 _LOGGER.exception("SUPERNOTIFY Failed to pre-grab image") 

490 

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

492 await self._schedule_deliveries(deferred_deliveries) 

493 

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

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

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

497 _LOGGER.info( 

498 "SUPERNOTIFY No delivery succeeded, activating fallback_by_default: %s", 

499 delivery.name, 

500 ) 

501 if delivery.name not in self.selected_deliveries: 

502 await self.call_transport(delivery) 

503 self.fallback += 1 

504 

505 if self.failed > 0: 

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

507 _LOGGER.warning( 

508 "SUPERNOTIFY Delivery failed, activating fallback_on_error: %s", 

509 delivery.name, 

510 ) 

511 if delivery.name not in self.selected_deliveries: 

512 await self.call_transport(delivery) 

513 self.fallback += 1 

514 

515 return self.delivered > 0 

516 

517 async def _schedule_deliveries(self, deliveries: dict[str, DeliveryTargetOverride | None]) -> None: 

518 delivery_coros = [] 

519 for delivery_name, target_overrides in deliveries.items(): 

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

521 if delivery: 

522 delivery_coros.append(self.call_transport(delivery, target_overrides)) 

523 else: 

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

525 if delivery_coros: 

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

527 for result in results: 

528 if isinstance(result, BaseException): 

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

530 

531 async def call_transport(self, delivery: Delivery, target_override: DeliveryTargetOverride | None = None) -> None: 

532 try: 

533 transport: Transport = delivery.transport 

534 if not transport.enabled: 

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

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

537 return 

538 

539 delivery_priorities: list[str] = delivery.priority 

540 if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

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

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

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

544 return 

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

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

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

548 return 

549 

550 targets: list[Target] = self.generate_targets(delivery, target_override) 

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

552 if not envelopes: 

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

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

555 ): 

556 reason: SuppressionReason = SuppressionReason.NO_TARGET 

557 else: 

558 reason = SuppressionReason.UNKNOWN 

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

560 

561 for envelope in envelopes: 

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

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

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

565 continue 

566 try: 

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

568 _LOGGER.info( 

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

570 delivery.name, 

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

572 ) 

573 self.record_result(delivery, envelope) 

574 except Exception as e2: 

575 envelope.error_count = envelope.error_count + 1 

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

577 transport.log_delivery_failure(e2, "SUPERNOTIFY Failed to deliver %s", delivery.name) 

578 envelope.delivery_error = format_exception(e2) 

579 self.record_result(delivery, envelope) 

580 

581 except Exception as e: 

582 _LOGGER.exception( 

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

584 delivery.name, 

585 type(delivery.transport).__name__, 

586 ) 

587 self.delivery_exceptions.setdefault(delivery.name, []) 

588 self.delivery_exceptions[delivery.name].append("\n".join(format_exception(e))) 

589 

590 def record_result( 

591 self, 

592 delivery: Delivery | None, 

593 envelope: Envelope | None = None, 

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

595 suppression_reason: SuppressionReason | None = None, 

596 ) -> None: 

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

598 if delivery: 

599 if envelope: 

600 self.delivered += envelope.delivered 

601 self.error_count += envelope.error_count 

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

603 if envelope.delivered: 

604 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.SUCCESS, []) 

605 self.deliveries[delivery.name][EnvelopeOutcome.SUCCESS].append(envelope) # type: ignore 

606 else: 

607 if suppression_reason: 

608 envelope.skip_reason = suppression_reason 

609 if suppression_reason not in self._skip_reasons: 

610 self._skip_reasons.append(suppression_reason) 

611 if suppression_reason == SuppressionReason.DUPE: 

612 self.dupe = True 

613 if envelope.error_count: 

614 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.ERROR, []) 

615 self.deliveries[delivery.name][EnvelopeOutcome.ERROR].append(envelope) # type: ignore 

616 self.failed += 1 

617 else: 

618 self.deliveries[delivery.name].setdefault(EnvelopeOutcome.SUPPRESSED, []) 

619 self.deliveries[delivery.name][EnvelopeOutcome.SUPPRESSED].append(envelope) # type: ignore 

620 self.suppressed += 1 

621 

622 if not envelope: 

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

624 skip_summary: dict[str, Any] = { 

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

626 "suppression_reason": str(suppression_reason), 

627 } 

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

629 if targets: 

630 skip_summary["targets"] = targets 

631 self.deliveries[delivery_name][EnvelopeOutcome.SKIPPED] = skip_summary 

632 self.skipped += 1 

633 

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

635 """ArchiveableObject implementation""" 

636 minimal = not diagnostics 

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

638 keys_only = ["enabled_scenarios"] 

639 debug_only = ["debug_trace"] 

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

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

642 preferred_order = [ 

643 "id", 

644 "created", 

645 "message", 

646 "applied_scenario_names", 

647 "constrain_scenario_names", 

648 "required_scenario_names", 

649 "enabled_scenarios", 

650 "selected_scenario_names", 

651 "delivery_selection", 

652 "delivery_overrides", 

653 "delivery_selection", 

654 "selected_deliveries", 

655 "delivered", 

656 "failed", 

657 "suppressed", 

658 "skipped", 

659 "error_count", 

660 "delivery_exceptions", 

661 "deliveries", 

662 ] 

663 # preferred fields 

664 result: dict[str, Any] = {"version": _VERSION, "outcome": self.outcome()} 

665 result.update({ 

666 k: sanitize( 

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

668 ) 

669 for k in preferred_order 

670 if k in self.__dict__ 

671 }) 

672 # all the rest not explicitly excluded 

673 result.update({ 

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

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

676 if k not in result 

677 and k not in exposed_if_populated 

678 and k not in object_refs 

679 and not k.startswith("_") 

680 and (not minimal or k not in keys_only) 

681 and (not minimal or k not in debug_only) 

682 }) 

683 # the exposed only if populated fields 

684 result.update({ 

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

686 for k in exposed_if_populated 

687 if self.__dict__.get(k) 

688 }) 

689 # delivery_stats: aggregate delivery metrics 

690 try: 

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

692 total_ok = 0 

693 total_all = 0 

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