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

1import datetime as dt 

2import logging 

3import uuid 

4from traceback import format_exception 

5from typing import TYPE_CHECKING, Any, cast 

6 

7import voluptuous as vol 

8from homeassistant.components.notify.const import ATTR_DATA 

9from voluptuous import humanize 

10 

11from custom_components.supernotify.schema import SelectionRank 

12 

13from .archive import ArchivableObject 

14from .common import ensure_list, nullable_ensure_list, sanitize 

15from .const import ( 

16 ATTR_ACTION_GROUPS, 

17 ATTR_ACTIONS, 

18 ATTR_DEBUG, 

19 ATTR_DELIVERY, 

20 ATTR_DELIVERY_SELECTION, 

21 ATTR_MEDIA, 

22 ATTR_MEDIA_CLIP_URL, 

23 ATTR_MEDIA_SNAPSHOT_URL, 

24 ATTR_MESSAGE_HTML, 

25 ATTR_PERSON_ID, 

26 ATTR_PRIORITY, 

27 ATTR_RECIPIENTS, 

28 ATTR_SCENARIOS_APPLY, 

29 ATTR_SCENARIOS_CONSTRAIN, 

30 ATTR_SCENARIOS_REQUIRE, 

31 DELIVERY_SELECTION_EXPLICIT, 

32 DELIVERY_SELECTION_FIXED, 

33 DELIVERY_SELECTION_IMPLICIT, 

34 OPTION_UNIQUE_TARGETS, 

35 PRIORITY_MEDIUM, 

36 PRIORITY_VALUES, 

37 TARGET_USE_FIXED, 

38 TARGET_USE_MERGE_ALWAYS, 

39 TARGET_USE_MERGE_ON_DELIVERY_TARGETS, 

40 TARGET_USE_ON_NO_ACTION_TARGETS, 

41 TARGET_USE_ON_NO_DELIVERY_TARGETS, 

42) 

43from .context import Context 

44from .delivery import Delivery, DeliveryRegistry 

45from .envelope import Envelope 

46from .model import ( 

47 ConditionVariables, 

48 DebugTrace, 

49 DeliveryCustomization, 

50 SuppressionReason, 

51 Target, 

52 TargetRequired, 

53) 

54from .people import Recipient 

55from .schema import ACTION_DATA_SCHEMA, STRICT_ACTION_DATA_SCHEMA 

56 

57if TYPE_CHECKING: 

58 from .people import PeopleRegistry 

59 from .scenario import Scenario 

60 from .transport import ( 

61 Transport, 

62 ) 

63 

64_LOGGER = logging.getLogger(__name__) 

65 

66# Deliveries mapping keys for debug / archive 

67KEY_DELIVERED = "delivered" 

68KEY_SUPPRESSED = "suppressed" 

69KEY_FAILED = "failed" 

70KEY_SKIPPED = "skipped" 

71 

72type t_delivery_name = str 

73type t_outcome = str 

74 

75 

76class Notification(ArchivableObject): 

77 def __init__( 

78 self, 

79 context: Context, 

80 message: str | None = None, 

81 title: str | None = None, 

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

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

84 ) -> None: 

85 self.created: dt.datetime = dt.datetime.now(tz=dt.UTC) 

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

87 self.message: str | None = message 

88 self.context: Context = context 

89 self.people_registry: PeopleRegistry = context.people_registry 

90 self.delivery_registry: DeliveryRegistry = context.delivery_registry 

91 action_data = action_data or {} 

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

93 self._already_selected: Target = Target() 

94 self._title: str | None = title 

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

96 self.delivered: int = 0 

97 self.error_count: int = 0 

98 self.skipped: int = 0 

99 self.failed: int = 0 

100 self.suppressed: int = 0 

101 self.dupe: bool = False 

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

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

104 

105 self.validate_action_data(action_data) 

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

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

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

109 } 

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

111 

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

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

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

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

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

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

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

119 

120 delivery_data = action_data.get(ATTR_DELIVERY) 

121 if isinstance(delivery_data, list): 

122 # a bare list of deliveries implies intent to restrict 

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

124 if self.delivery_selection is None: 

125 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

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

127 elif isinstance(delivery_data, str) and delivery_data: 

128 # a bare list of deliveries implies intent to restrict 

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

130 if self.delivery_selection is None: 

131 self.delivery_selection = DELIVERY_SELECTION_EXPLICIT 

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

133 elif isinstance(delivery_data, dict): 

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

135 if self.delivery_selection is None: 

136 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

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

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

139 elif delivery_data: 

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

141 if self.delivery_selection is None: 

142 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

143 else: 

144 if self.delivery_selection is None: 

145 self.delivery_selection = DELIVERY_SELECTION_IMPLICIT 

146 

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

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

149 self.extra_data.update(action_data.get(ATTR_DATA, {})) 

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

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

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

153 

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

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

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

157 self._suppression_reason: SuppressionReason | None = None 

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

159 self.condition_variables: ConditionVariables 

160 

161 async def initialize(self) -> None: 

162 """Async post-construction initialization""" 

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

164 self.condition_variables = ConditionVariables( 

165 self.applied_scenario_names, 

166 self.required_scenario_names, 

167 self.constrain_scenario_names, 

168 self.priority, 

169 self.occupancy, 

170 self.message, 

171 self._title, 

172 ) # requires occupancy first 

173 

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

175 self.selected_scenario_names = await self.select_scenarios() 

176 enabled_scenario_names.extend(self.selected_scenario_names) 

177 if self.constrain_scenario_names: 

178 enabled_scenario_names = [ 

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

180 ] 

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

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

183 self.selected_deliveries = {} 

184 self.suppress(SuppressionReason.NO_SCENARIO) 

185 else: 

186 for s in enabled_scenario_names: 

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

188 if scenario_obj is not None: 

189 self.enabled_scenarios[s] = scenario_obj 

190 

191 self.selected_deliveries = self.select_deliveries() 

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

193 self.suppress(SuppressionReason.SNOOZED) 

194 self.apply_enabled_scenarios() 

195 

196 if not self.media: 

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

198 

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

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

201 

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

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

204 """ 

205 media_dict = {} 

206 if not data: 

207 return {} 

208 if data.get("image"): 

209 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = data.get("image") 

210 if data.get("video"): 

211 media_dict[ATTR_MEDIA_CLIP_URL] = data.get("video") 

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

213 url = data["attachment"]["url"] 

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

215 media_dict[ATTR_MEDIA_CLIP_URL] = url 

216 elif ( 

217 url 

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

219 and not media_dict.get(ATTR_MEDIA_SNAPSHOT_URL) 

220 ): 

221 media_dict[ATTR_MEDIA_SNAPSHOT_URL] = url 

222 return media_dict 

223 

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

225 if action_data.get(ATTR_PRIORITY): 

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

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

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

229 else: 

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

231 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

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

233 try: 

234 humanize.validate_with_humanized_errors(action_data, ACTION_DATA_SCHEMA) 

235 except vol.Invalid as e: 

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

237 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

238 raise 

239 except vol.error.Error as e2: 

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

241 self.suppress(SuppressionReason.INVALID_ACTION_DATA) 

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

243 

244 def apply_enabled_scenarios(self) -> None: 

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

246 action_groups: list[str] = [] 

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

248 if scenario.media: 

249 if self.media: 

250 self.media.update(scenario.media) 

251 else: 

252 self.media = scenario.media 

253 if scenario.action_groups: 

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

255 # self.action_groups only accessed from inside Envelope 

256 if self.action_groups: 

257 self.action_groups.extend(action_groups) 

258 else: 

259 self.action_groups = action_groups 

260 

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

262 scenario_enable_deliveries: list[str] = [] 

263 scenario_disable_deliveries: list[str] = [] 

264 default_enable_deliveries: list[str] = [] 

265 recipients_enable_deliveries: list[str] = [] 

266 

267 if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

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

269 scenario_enable_deliveries.extend(scenario.enabling_deliveries()) 

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

271 scenario_disable_deliveries.extend(scenario.disabling_deliveries()) 

272 

273 scenario_enable_deliveries = list(set(scenario_enable_deliveries)) 

274 scenario_disable_deliveries = list(set(scenario_disable_deliveries)) 

275 

276 for recipient in self.all_recipients(): 

277 recipients_enable_deliveries.extend(recipient.enabling_delivery_names()) 

278 if self.delivery_selection == DELIVERY_SELECTION_IMPLICIT: 

279 # all deliveries with SELECTION_DEFAULT in CONF_SELECTION 

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

281 

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

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

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

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

286 

287 override_enable_deliveries: list[str] = [] 

288 override_disable_deliveries: list[str] = [] 

289 

290 # apply the deliveries defined in the notification action call 

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

292 if ( 

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

294 and delivery in self.context.delivery_registry.enabled_deliveries 

295 ) or ( 

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

297 and delivery in self.context.delivery_registry.disabled_deliveries 

298 ): 

299 override_enable_deliveries.append(delivery) 

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

301 override_disable_deliveries.append(delivery) 

302 

303 # if self.delivery_selection != DELIVERY_SELECTION_FIXED: 

304 # scenario_disable_deliveries = [ 

305 # d.name 

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

307 # if d.selection == [SELECTION_BY_SCENARIO] 

308 # and d.name not in scenario_enable_deliveries 

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

310 # ] 

311 all_global_enabled: list[str] = list( 

312 set(scenario_enable_deliveries + default_enable_deliveries + override_enable_deliveries) 

313 ) 

314 all_enabled: list[str] = all_global_enabled + recipients_enable_deliveries 

315 all_disabled: list[str] = scenario_disable_deliveries + override_disable_deliveries 

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

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

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

319 

320 unsorted_maybe_objs: list[Delivery | None] = [ 

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

322 ] 

323 unsorted_objs: list[Delivery] = [ 

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

325 ] 

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

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

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

329 selected = first + anywhere + last 

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

331 

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

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

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

335 for personal_delivery in personal_deliveries: 

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

337 for recipient in self.all_recipients(): 

338 if personal_delivery in recipient.enabling_delivery_names(): 

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

340 return results 

341 

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

343 self._suppression_reason = reason 

344 if reason not in self._skip_reasons: 

345 self._skip_reasons.append(reason) 

346 _LOGGER.info(f"SUPERNOTIFY Suppressing notification, reason:{reason}, id:{self.id}") 

347 

348 async def deliver(self) -> bool: 

349 _LOGGER.debug( 

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

351 self.message, 

352 self.id, 

353 self.selected_deliveries, 

354 ) 

355 

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

357 self.deliveries[delivery_name] = {} 

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

359 if self._suppression_reason is not None: 

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

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

362 elif delivery: 

363 await self.call_transport(delivery, recipients=details.get("recipients")) 

364 else: 

365 _LOGGER.error(f"SUPERNOTIFY Unexpected missing delivery {delivery_name}") 

366 

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

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

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

370 if delivery.name not in self.selected_deliveries: 

371 await self.call_transport(delivery) 

372 

373 if self.failed > 0: 

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

375 if delivery.name not in self.selected_deliveries: 

376 await self.call_transport(delivery) 

377 

378 return self.delivered > 0 

379 

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

381 try: 

382 transport: Transport = delivery.transport 

383 if not transport.enabled: 

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

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

386 return 

387 

388 delivery_priorities: list[str] = delivery.priority 

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

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

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

392 return 

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

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

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

396 return 

397 

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

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

400 if not envelopes: 

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

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

403 ): 

404 reason: SuppressionReason = SuppressionReason.NO_TARGET 

405 else: 

406 reason = SuppressionReason.UNKNOWN 

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

408 

409 for envelope in envelopes: 

410 if self.context.dupe_checker.check(envelope): 

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

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

413 continue 

414 try: 

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

416 _LOGGER.debug("SUPERNOTIFY No delivery for %s", delivery.name) 

417 self.record_result(delivery, envelope) 

418 except Exception as e2: 

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

420 envelope.error_count = envelope.error_count + 1 

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

422 envelope.delivery_error = format_exception(e2) 

423 self.record_result(delivery, envelope) 

424 

425 except Exception as e: 

426 _LOGGER.exception("SUPERNOTIFY Failed to notify using %s", delivery.name) 

427 _LOGGER.debug("SUPERNOTIFY %s delivery failure", delivery, exc_info=True) 

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

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

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

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

432 

433 def record_result( 

434 self, 

435 delivery: Delivery | None, 

436 envelope: Envelope | None = None, 

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

438 suppression_reason: SuppressionReason | None = None, 

439 ) -> None: 

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

441 if delivery: 

442 if envelope: 

443 self.delivered += envelope.delivered 

444 self.error_count += envelope.error_count 

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

446 if envelope.delivered: 

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

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

449 else: 

450 if suppression_reason: 

451 envelope.skip_reason = suppression_reason 

452 if suppression_reason not in self._skip_reasons: 

453 self._skip_reasons.append(suppression_reason) 

454 if suppression_reason == SuppressionReason.DUPE: 

455 self.dupe = True 

456 if envelope.error_count: 

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

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

459 self.failed += 1 

460 else: 

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

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

463 self.suppressed += 1 

464 

465 if not envelope: 

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

467 skip_summary: dict[str, Any] = { 

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

469 "suppression_reason": str(suppression_reason), 

470 } 

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

472 if targets: 

473 skip_summary["targets"] = targets 

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

475 self.skipped += 1 

476 

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

478 """ArchiveableObject implementation""" 

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

480 keys_only = ["enabled_scenarios"] 

481 debug_only = ["debug_trace"] 

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

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

484 preferred_order = [ 

485 "id", 

486 "created", 

487 "message", 

488 "applied_scenario_names", 

489 "constrain_scenario_names", 

490 "required_scenario_names", 

491 "enabled_scenarios", 

492 "selected_scenario_names", 

493 "delivery_selection", 

494 "delivery_overrides", 

495 "delivery_selection", 

496 "selected_deliveries", 

497 "recipients_override", 

498 "delivered", 

499 "failed", 

500 "suppressed", 

501 "skipped", 

502 "error_count", 

503 "deliveries", 

504 ] 

505 # preferred fields 

506 result = { 

507 k: sanitize( 

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

509 ) 

510 for k in preferred_order 

511 } 

512 # all the rest not explicitly excluded 

513 result.update({ 

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

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

516 if k not in result 

517 and k not in exposed_if_populated 

518 and k not in object_refs 

519 and not k.startswith("_") 

520 and (not minimal or k not in keys_only) 

521 and (not minimal or k not in debug_only) 

522 }) 

523 # the exposed only if populated fields 

524 result.update({ 

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

526 for k in exposed_if_populated 

527 if self.__dict__.get(k) 

528 }) 

529 return result 

530 

531 def base_filename(self) -> str: 

532 """ArchiveableObject implementation""" 

533 return f"{self.created.isoformat()[:16]}_{self.id}" 

534 

535 def delivery_data(self, delivery_name: str) -> dict[str, Any]: 

536 delivery_override: DeliveryCustomization | None = self.delivery_overrides.get(delivery_name) 

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

538 

539 @property 

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

541 result: list[Envelope] = [] 

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

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

544 return result 

545 

546 @property 

547 def undelivered_envelopes(self) -> list[Envelope]: 

548 result: list[Envelope] = [] 

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

550 result.extend(cast("list[Envelope]", delivery_result.get(KEY_SUPPRESSED, []))) 

551 result.extend(cast("list[Envelope]", delivery_result.get(KEY_FAILED, []))) 

552 return result 

553 

554 async def select_scenarios(self) -> list[str]: 

555 return [s.name for s in self.context.scenario_registry.scenarios.values() if s.evaluate(self.condition_variables)] 

556 

557 def generate_targets(self, delivery: Delivery, recipients: list[str] | None = None) -> list[Target]: 

558 

559 if delivery.target_required == TargetRequired.NEVER: 

560 # don't waste time computing targets for deliveries that don't need them 

561 return [Target(None, target_data=delivery.data)] 

562 

563 computed_target: Target 

564 

565 if delivery.target_usage == TARGET_USE_FIXED: 

566 if delivery.target: 

567 computed_target = delivery.target.safe_copy() 

568 self.debug_trace.record_target(delivery.name, "100_delivery_default_fixed", computed_target) 

569 else: 

570 computed_target = Target(None, target_data=delivery.data) 

571 self.debug_trace.record_target(delivery.name, "101_delivery_default_fixed_empty", computed_target) 

572 elif recipients is not None: 

573 computed_target = Target(recipients) 

574 self.debug_trace.record_target(delivery.name, "102_delivery_default_fixed", computed_target) 

575 

576 elif not self._target: 

577 # Unless there are explicit targets, include everyone on the people registry 

578 computed_target = self.default_person_ids(delivery) 

579 self.debug_trace.record_target(delivery.name, "201_no_action_target", computed_target) 

580 else: 

581 computed_target = self._target.safe_copy() 

582 self.debug_trace.record_target(delivery.name, "202_action_target", computed_target) 

583 

584 # 1st round of filtering for snooze and resolving people->direct targets 

585 computed_target = self.context.snoozer.filter_recipients(computed_target, self.priority, delivery) 

586 self.debug_trace.record_target(delivery.name, "300_post_snooze", computed_target) 

587 # turn person_ids into emails and phone numbers 

588 for indirect_target in self.resolve_indirect_targets(computed_target, delivery): 

589 computed_target += indirect_target 

590 self.debug_trace.record_target(delivery.name, "310_resolve_indirect", computed_target) 

591 computed_target += self.resolve_scenario_targets(delivery) 

592 self.debug_trace.record_target(delivery.name, "320_resolved_scenario_targets", computed_target) 

593 # filter out target not required for this delivery 

594 computed_target = delivery.select_targets(computed_target) 

595 self.debug_trace.record_target(delivery.name, "330_delivery_selection", computed_target) 

596 primary_count = len(computed_target) 

597 

598 if delivery.target_usage == TARGET_USE_ON_NO_DELIVERY_TARGETS: 

599 if not computed_target.has_targets() and delivery.target: 

600 computed_target += delivery.target 

601 self.debug_trace.record_target(delivery.name, "400_delivery_default_no_delivery_targets", computed_target) 

602 elif delivery.target_usage == TARGET_USE_ON_NO_ACTION_TARGETS: 

603 if not self._target and delivery.target: 

604 computed_target += delivery.target 

605 self.debug_trace.record_target(delivery.name, "401_delivery_default_no_action_targets", computed_target) 

606 elif delivery.target_usage == TARGET_USE_MERGE_ON_DELIVERY_TARGETS: 

607 # merge in the delivery defaults if there's a target defined in action call 

608 if computed_target.has_targets() and delivery.target: 

609 computed_target += delivery.target 

610 self.debug_trace.record_target(delivery.name, "402_delivery_merge_on_delivery_targets", computed_target) 

611 elif delivery.target_usage == TARGET_USE_MERGE_ALWAYS: 

612 # merge in the delivery defaults even if there's not a target defined in action call 

613 if delivery.target: 

614 computed_target += delivery.target 

615 self.debug_trace.record_target(delivery.name, "403_delivery_merge_always_targets", computed_target) 

616 elif delivery.target_usage == TARGET_USE_FIXED: 

617 _LOGGER.debug("SUPERNOTIFY Fixed target on delivery %s", delivery.name) 

618 self.debug_trace.record_target(delivery.name, "404_fixed_target", computed_target) 

619 else: 

620 self.debug_trace.record_target(delivery.name, "405_no_target_usage_match", computed_target) 

621 _LOGGER.debug("SUPERNOTIFY No useful target definition for delivery %s", delivery.name) 

622 

623 if len(computed_target) > primary_count: 

624 _LOGGER.debug( 

625 "SUPERNOTIFY Delivery config added %s targets for %s", len(computed_target) - primary_count, delivery.name 

626 ) 

627 

628 # 2nd round of filtering for snooze and resolving people->direct targets after delivery target applied 

629 computed_target = self.context.snoozer.filter_recipients(computed_target, self.priority, delivery) 

630 self.debug_trace.record_target(delivery.name, "501_post_snooze", computed_target) 

631 for indirect_target in self.resolve_indirect_targets(computed_target, delivery): 

632 computed_target += indirect_target 

633 self.debug_trace.record_target(delivery.name, "502_resolved_indirect_targets", computed_target) 

634 computed_target += self.resolve_scenario_targets(delivery) 

635 self.debug_trace.record_target(delivery.name, "503_resolved_scenario_targets", computed_target) 

636 computed_target = delivery.select_targets(computed_target) 

637 self.debug_trace.record_target(delivery.name, "504_delivery_selection", computed_target) 

638 

639 split_targets: list[Target] = computed_target.split_by_target_data() 

640 self.debug_trace.record_target(delivery.name, "610_delivery_split_targets", split_targets) 

641 

642 direct_targets: list[Target] = [t.direct() for t in split_targets] 

643 self.debug_trace.record_target(delivery.name, "620_narrow_to_direct", direct_targets) 

644 

645 if delivery.options.get(OPTION_UNIQUE_TARGETS, False): 

646 direct_targets = [t - self._already_selected for t in direct_targets] 

647 self.debug_trace.record_target(delivery.name, "630_make_unique_across_deliveries", direct_targets) 

648 for direct_target in direct_targets: 

649 self._already_selected += direct_target 

650 self.debug_trace.record_target(delivery.name, "999_final_cut", direct_targets) 

651 return direct_targets 

652 

653 def resolve_scenario_targets(self, delivery: Delivery) -> Target: 

654 resolved: Target = Target() 

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

656 customization: DeliveryCustomization | None = scenario.delivery_customization(delivery.name)