Coverage for custom_components/supernotify/archive.py: 96%
335 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import datetime as dt
4import json
5import logging
6import math
7import os
8from abc import abstractmethod
9from typing import TYPE_CHECKING, Any
11import aiofiles.os
12import anyio
13import homeassistant.util.dt as dt_util
14from anyio import Path
15from homeassistant.const import (
16 CONF_ENABLED,
17)
19from .const import (
20 CONF_ARCHIVE_DAYS,
21 CONF_ARCHIVE_DIAGNOSTICS,
22 CONF_ARCHIVE_EVENT_NAME,
23 CONF_ARCHIVE_EVENT_SELECTION,
24 CONF_ARCHIVE_MQTT_QOS,
25 CONF_ARCHIVE_MQTT_RETAIN,
26 CONF_ARCHIVE_MQTT_TOPIC,
27 CONF_ARCHIVE_PATH,
28)
29from .schema import DeliveryOutcome, OutcomeSelection
31if TYPE_CHECKING:
32 from homeassistant.core import Context as HAContext
33 from homeassistant.helpers.typing import ConfigType
35 from custom_components.supernotify.hass_api import HomeAssistantAPI
37 from .engine import SupernotifyEngine
38 from .people import Recipient
40_LOGGER = logging.getLogger(__name__)
42ARCHIVE_DEFAULT_DAYS = 1
43WRITE_TEST = ".startup"
44ARCHIVE_PATH_ISSUE = "archive_path"
47def _filename_period(filename: str) -> tuple[float, float] | None:
48 """The span of time an archive file could have been created in, going by the local time
49 to the minute at the start of its name, or ``None`` if it isn't named that way.
51 Both sides of a daylight saving change are allowed for, since the name has no UTC offset.
52 """
53 time_zone = dt_util.get_default_time_zone()
54 try:
55 named: dt.datetime = dt.datetime.strptime(filename[:16], "%Y-%m-%dT%H-%M").replace(tzinfo=time_zone)
56 except ValueError:
57 return None
58 stamps: list[float] = [named.replace(fold=fold).timestamp() for fold in (0, 1)]
59 return min(stamps), max(stamps) + 60
62def _list_entries_sync(
63 archive_path: str,
64 limit: int,
65 after: dt.datetime | None,
66 before: dt.datetime | None,
67 outcome: str | None,
68 priority: list[str] | None = None,
69) -> list[dict[str, Any]]:
70 """The body of ``ArchiveDirectory.list_entries``, run in the executor as one job."""
71 entries: list[dict[str, Any]] = []
72 priorities: set[str] | None = {str(p).lower() for p in priority} if priority else None
73 try:
74 with os.scandir(archive_path) as scan:
75 files = [e for e in scan if e.name.endswith(".json") and e.name != WRITE_TEST]
76 files.sort(key=lambda e: e.stat().st_ctime, reverse=True)
77 earliest: float = after.timestamp() if after else -math.inf
78 latest: float = before.timestamp() if before else math.inf
79 for entry in files:
80 if len(entries) >= limit:
81 break
82 if (after or before) and (named := _filename_period(entry.name)) and (named[1] < earliest or named[0] > latest):
83 continue
84 try:
85 with open(entry.path, encoding="utf-8") as fh:
86 data: dict[str, Any] = json.load(fh)
87 except Exception as exc:
88 _LOGGER.debug("SUPERNOTIFY Skipping unreadable archive file %s: %s", entry.name, exc)
89 continue
90 created: dt.datetime | None = dt_util.parse_datetime(data.get("created") or "")
91 if after is not None and created is not None and created < after:
92 continue
93 if before is not None and created is not None and created > before:
94 continue
95 if outcome is not None and str(data.get("outcome", "")).lower() != outcome.lower():
96 continue
97 if priorities is not None and str(data.get("priority", "")).lower() not in priorities:
98 continue
99 entries.append(data)
100 except Exception as exc:
101 _LOGGER.warning("SUPERNOTIFY Unable to list archive entries: %s", exc)
102 return entries
105class ArchivableObject:
106 ha_context: HAContext | None = None
108 @abstractmethod
109 def base_filename(self) -> str:
110 pass
112 @abstractmethod
113 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> dict[str, Any]:
114 pass
116 def outcome(self) -> DeliveryOutcome:
117 return DeliveryOutcome.NO_DELIVERY
119 def diagnostics_selected(self, outcome_policy: OutcomeSelection) -> bool:
120 """Whether the archived copy should carry the full diagnostic content"""
121 return self.selected(outcome_policy)
123 def selected(self, outcome_policy: OutcomeSelection) -> bool:
124 return bool(
125 outcome_policy & OutcomeSelection.ALL
126 or (outcome_policy & OutcomeSelection.SUCCESS and self.outcome() == DeliveryOutcome.SUCCESS)
127 or (outcome_policy & OutcomeSelection.NO_DELIVERY and self.outcome() == DeliveryOutcome.NO_DELIVERY)
128 or (outcome_policy & OutcomeSelection.PARTIAL_DELIVERY and self.outcome() == DeliveryOutcome.PARTIAL_DELIVERY)
129 or (outcome_policy & OutcomeSelection.DUPE and self.outcome() == DeliveryOutcome.DUPE)
130 or (outcome_policy & OutcomeSelection.FALLBACK_DELIVERY and self.outcome() == DeliveryOutcome.FALLBACK_DELIVERY)
131 or (outcome_policy & OutcomeSelection.ERROR and self.outcome() == DeliveryOutcome.ERROR)
132 )
135class ArchiveDestination:
136 @abstractmethod
137 async def archive(self, archive_object: ArchivableObject) -> bool:
138 pass
141class EventArchiver(ArchiveDestination):
142 def __init__(
143 self,
144 hass_api: HomeAssistantAPI,
145 event_name: str,
146 selection: OutcomeSelection = OutcomeSelection.NONE,
147 diagnostics: OutcomeSelection = OutcomeSelection.ERROR,
148 ) -> None:
149 self.hass_api = hass_api
150 self.event_name = event_name
151 self.selection = selection
152 self.diagnostics = diagnostics
153 if selection:
154 _LOGGER.info("SUPERNOTIFY Archiving %s notifications as %s events", selection.name, event_name)
156 async def archive(self, archive_object: ArchivableObject) -> bool:
157 if not archive_object.selected(self.selection):
158 return False
159 try:
160 payload = archive_object.contents(diagnostics=archive_object.diagnostics_selected(self.diagnostics))
161 self.hass_api.fire_event(self.event_name, payload, context=archive_object.ha_context)
162 return True
163 except Exception:
164 _LOGGER.warning("SUPERNOTIFY Failed to archive to event %s", self.event_name)
165 return False
168class ArchiveTopic(ArchiveDestination):
169 def __init__(
170 self,
171 hass_api: HomeAssistantAPI,
172 topic: str,
173 qos: int = 0,
174 retain: bool = False,
175 diagnostics: OutcomeSelection = OutcomeSelection.ERROR,
176 ) -> None:
177 self.hass_api: HomeAssistantAPI = hass_api
178 self.topic: str = topic
179 self.qos: int = qos
180 self.retain: bool = retain
181 self.diagnostics: OutcomeSelection = diagnostics
182 self.enabled: bool = False
184 async def initialize(self) -> None:
185 if self.topic:
186 if await self.hass_api.mqtt_available(raise_on_error=False):
187 _LOGGER.info(f"SUPERNOTIFY Archiving to MQTT topic {self.topic}, qos {self.qos}, retain {self.retain}")
188 self.enabled = True
189 else:
190 _LOGGER.warning(
191 f"SUPERNOTIFY Archiving configured for topic {self.topic} but MQTT not available at startup, disabled"
192 )
194 async def archive(self, archive_object: ArchivableObject) -> bool:
195 if not self.enabled:
196 return False
197 payload = archive_object.contents(diagnostics=archive_object.diagnostics_selected(self.diagnostics))
198 topic = f"{self.topic}/{archive_object.base_filename()}"
199 _LOGGER.debug(f"SUPERNOTIFY Publishing notification to {topic}")
200 try:
201 await self.hass_api.mqtt_publish(
202 topic=topic,
203 payload=payload,
204 qos=self.qos,
205 retain=self.retain,
206 )
207 return True
208 except Exception:
209 _LOGGER.warning(f"SUPERNOTIFY Failed to archive to topic {self.topic}")
210 return False
213class ArchiveDirectory(ArchiveDestination):
214 def __init__(
215 self,
216 path: str,
217 diagnostics: OutcomeSelection = OutcomeSelection.ERROR,
218 hass_api: HomeAssistantAPI | None = None,
219 ) -> None:
220 self.hass_api: HomeAssistantAPI | None = hass_api
221 self.configured_path: str = path
222 self.archive_path: anyio.Path | None = None
223 self.enabled: bool = False
224 # why the path can't be used, if it can't
225 self.error: str | None = None
226 self.diagnostics: OutcomeSelection = diagnostics
228 async def initialize(self) -> None:
229 verify_archive_path: Path = Path(self.configured_path)
230 if not await verify_archive_path.exists():
231 _LOGGER.info("SUPERNOTIFY Archive path not found at %s", verify_archive_path)
232 try:
233 await verify_archive_path.mkdir(parents=True, exist_ok=True)
234 except Exception as e:
235 _LOGGER.warning("SUPERNOTIFY Archive path %s cannot be created: %s", verify_archive_path, e)
236 self.error = str(e)
237 return
238 if not await verify_archive_path.is_dir():
239 _LOGGER.warning("SUPERNOTIFY Archive path %s is not a directory", verify_archive_path)
240 self.error = "not a directory"
241 return
242 try:
243 await verify_archive_path.joinpath(WRITE_TEST).touch(exist_ok=True)
244 self.archive_path = verify_archive_path
245 _LOGGER.info("SUPERNOTIFY Archiving notifications to file system at %s", verify_archive_path)
246 self.enabled = True
247 except Exception as e:
248 _LOGGER.warning("SUPERNOTIFY Archive path %s cannot be written: %s", verify_archive_path, e)
249 self.error = str(e)
251 async def archive(self, archive_object: ArchivableObject) -> bool:
252 archived: bool = False
254 if self.enabled and self.archive_path: # archive_path to assuage mypy
255 archive_filepath: Path | None = None
256 diagnostics: bool = archive_object.diagnostics_selected(self.diagnostics)
257 try:
258 filename = f"{archive_object.base_filename()}.json"
259 archive_filepath = self.archive_path.joinpath(filename)
260 serialized: str = json.dumps(archive_object.contents(diagnostics=diagnostics), indent=2)
261 async with aiofiles.open(archive_filepath, mode="w") as file:
262 await file.write(serialized)
263 _LOGGER.debug("SUPERNOTIFY Archived notification %s", await archive_filepath.absolute())
264 archived = True
265 except Exception as e:
266 _LOGGER.warning("SUPERNOTIFY Unable to archive notification: %s", e)
267 if diagnostics and archive_filepath:
268 try:
269 serialized = json.dumps(archive_object.contents(diagnostics=False), indent=2)
270 async with aiofiles.open(archive_filepath, mode="w") as file:
271 await file.write(serialized)
272 _LOGGER.warning("SUPERNOTIFY Archived minimal notification %s", await archive_filepath.absolute())
273 archived = True
274 except Exception:
275 _LOGGER.exception("SUPERNOTIFY Unable to archive minimal notification")
276 return archived
278 async def list_entries(
279 self,
280 limit: int = 20,
281 after: dt.datetime | None = None,
282 before: dt.datetime | None = None,
283 outcome: str | None = None,
284 priority: list[str] | None = None,
285 ) -> list[dict[str, Any]]:
286 """Return up to *limit* archived notifications, newest first.
288 Optional *after* and *before* filter by the ``created`` timestamp stored in
289 each archive file. Optional *outcome* keeps only notifications whose top-level
290 ``outcome`` field matches, ignoring case: the archive stores the ``DeliveryOutcome``
291 value (``"success"``) and the action's selector offers ``"SUCCESS"``. Optional
292 *priority* keeps only notifications with one of these priorities, also ignoring case.
293 Filters apply before *limit*, so the newest *limit* matching notifications are returned.
295 Files are named for the minute they were created, so any that fall outside
296 *after* and *before* are passed over without being opened.
297 """
298 if not self.archive_path or not await self.archive_path.exists():
299 return []
300 # one executor job for the whole scan: a file at a time through aiofiles is a thread hop per
301 # open and per read, seconds for a month of archive
302 if self.hass_api is None: # only a directory made outside Home Assistant, as in some tests
303 return _list_entries_sync(str(self.archive_path), limit, after, before, outcome, priority)
304 entries: list[dict[str, Any]] = await self.hass_api.create_job(
305 _list_entries_sync, str(self.archive_path), limit, after, before, outcome, priority
306 )
307 return entries
309 async def read_entry(self, notification_id: str) -> dict[str, Any] | None:
310 """Return the full JSON of one archived notification by its id, or ``None``."""
311 if not self.archive_path or not await self.archive_path.exists():
312 return None
313 try:
314 raw_scan = await aiofiles.os.scandir(self.archive_path)
315 for entry in raw_scan:
316 if entry.name.endswith(".json") and notification_id in entry.name:
317 async with aiofiles.open(entry.path, mode="r") as fh:
318 return json.loads(await fh.read())
319 except Exception as exc:
320 _LOGGER.warning("SUPERNOTIFY Unable to read archive entry %s: %s", notification_id, exc)
321 return None
323 async def size(self) -> int:
324 path = self.archive_path
325 if path and await path.exists():
326 return sum(1 for p in await aiofiles.os.listdir(path) if p != WRITE_TEST)
327 return 0
329 async def cleanup(self, days: int) -> int:
330 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
331 cutoff = cutoff.astimezone(dt.UTC)
332 purged = 0
333 if self.archive_path and await self.archive_path.exists():
334 try:
335 archive = await aiofiles.os.scandir(self.archive_path)
336 for entry in archive:
337 if entry.name == WRITE_TEST:
338 continue
339 # st_ctime is used deliberately here (not st_birthtime): archive files are
340 # written once and never modified afterwards, so ctime reflects creation time
341 # on the platforms this integration targets; st_birthtime is not guaranteed to
342 # be available on all Linux filesystems.
343 if dt_util.utc_from_timestamp(entry.stat().st_ctime) <= cutoff: # ty: ignore[deprecated,unused-ignore-comment]
344 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
345 await aiofiles.os.unlink(entry.path)
346 purged += 1
347 except Exception as e:
348 _LOGGER.warning("SUPERNOTIFY Unable to clean up archive at %s: %s", self.archive_path, e, exc_info=True)
349 _LOGGER.info("SUPERNOTIFY Purged %s archived notifications for cutoff %s", purged, cutoff)
350 else:
351 _LOGGER.debug("SUPERNOTIFY Skipping archive purge for unknown path %s", self.archive_path)
352 return purged
355class NotificationArchive:
356 def __init__(
357 self,
358 config: ConfigType,
359 hass_api: HomeAssistantAPI,
360 ) -> None:
361 self.hass_api = hass_api
362 self.enabled = bool(config.get(CONF_ENABLED, False))
363 self.archive_directory: ArchiveDirectory | None = None
364 self.archive_topic: ArchiveTopic | None = None
365 self.event_archiver: EventArchiver | None = None
366 self.event_selection: OutcomeSelection = config.get(CONF_ARCHIVE_EVENT_SELECTION, OutcomeSelection.NONE)
367 self.diagnostics: OutcomeSelection = config.get(CONF_ARCHIVE_DIAGNOSTICS, OutcomeSelection.ERROR)
368 self.archive_event_name: str = config.get(CONF_ARCHIVE_EVENT_NAME, "supernotification")
369 self.configured_archive_path: str | None = config.get(CONF_ARCHIVE_PATH)
370 self.archive_days = int(config.get(CONF_ARCHIVE_DAYS, ARCHIVE_DEFAULT_DAYS))
371 self.mqtt_topic: str | None = config.get(CONF_ARCHIVE_MQTT_TOPIC)
372 self.mqtt_qos: int = int(config.get(CONF_ARCHIVE_MQTT_QOS, 0))
373 self.mqtt_retain: bool = bool(config.get(CONF_ARCHIVE_MQTT_RETAIN, False))
375 async def initialize(self) -> None:
376 if not self.enabled:
377 _LOGGER.info("SUPERNOTIFY Archive disabled")
378 return
379 if not self.configured_archive_path:
380 _LOGGER.warning("SUPERNOTIFY Archive path not configured")
381 else:
382 self.archive_directory = ArchiveDirectory(
383 self.hass_api.config_path(self.configured_archive_path),
384 diagnostics=self.diagnostics,
385 hass_api=self.hass_api,
386 )
387 await self.archive_directory.initialize()
388 if self.archive_directory.error is None:
389 self.hass_api.delete_issue(ARCHIVE_PATH_ISSUE)
390 else:
391 self.hass_api.raise_issue(
392 ARCHIVE_PATH_ISSUE,
393 ARCHIVE_PATH_ISSUE,
394 {"path": self.archive_directory.configured_path, "error": self.archive_directory.error},
395 learn_more_url="https://supernotify.rhizomatics.org.uk/latest/configuration/archiving/",
396 )
398 if self.mqtt_topic:
399 self.archive_topic = ArchiveTopic(self.hass_api, self.mqtt_topic, self.mqtt_qos, self.mqtt_retain, self.diagnostics)
400 await self.archive_topic.initialize()
402 self.event_archiver = EventArchiver(self.hass_api, self.archive_event_name, self.event_selection, self.diagnostics)
404 async def size(self) -> int:
405 return await self.archive_directory.size() if self.archive_directory else 0
407 async def cleanup(self, days: int | None = None) -> int:
408 days = self.archive_days if days is None else days
409 return await self.archive_directory.cleanup(days) if self.archive_directory else 0
411 async def archive(self, archive_object: ArchivableObject) -> bool:
412 archived: bool = False
413 if self.archive_topic and await self.archive_topic.archive(archive_object):
414 archived = True
415 if self.archive_directory and await self.archive_directory.archive(archive_object):
416 archived = True
417 if self.event_archiver:
418 await self.event_archiver.archive(archive_object)
420 return archived
423def _names_for(engine: SupernotifyEngine, person_ids: list[str]) -> list[str]:
424 def _display_name(recipient: Recipient) -> str:
425 return recipient.alias or recipient.name
427 people = engine.context.people_registry.people
428 return sorted(_display_name(people[p]) if p in people else p for p in person_ids)
431def _scenario_names(scenarios: Any) -> list[str]: # noqa: ANN401 - as stored, which has varied
432 """The scenario names of an archived notification: a list, or in older archive files a mapping by
433 name, or a single name."""
434 if not scenarios:
435 return []
436 if isinstance(scenarios, list | tuple | dict):
437 return [str(name) for name in scenarios]
438 return [str(scenarios)]
441def _bump(counts: dict[str, int], key: str) -> None:
442 counts[key] = counts.get(key, 0) + 1
445def summarize_by_day(entries: list[dict[str, Any]]) -> dict[str, Any]:
446 """Count archived notifications per local day, oldest first: how many, by outcome, priority,
447 local hour and applied scenario, and for each delivery in how many it sent or failed.
449 Older archive files keep `enabled_scenarios` as a mapping by name rather than a list - either
450 is counted by name. A notification without a readable `created` is left out."""
451 days: dict[str, dict[str, Any]] = {}
452 for contents in entries:
453 created: dt.datetime | None = dt_util.parse_datetime(str(contents.get("created") or ""))
454 if created is None:
455 continue
456 local: dt.datetime = dt_util.as_local(created)
457 key: str = local.date().isoformat()
458 day: dict[str, Any] = days.setdefault(
459 key, {"date": key, "count": 0, "outcome": {}, "priority": {}, "hour": [0] * 24, "deliveries": {}, "scenarios": {}}
460 )
461 day["count"] += 1
462 day["hour"][local.hour] += 1
463 _bump(day["outcome"], str(contents.get("outcome") or "unknown").lower())
464 _bump(day["priority"], str(contents.get("priority") or "unknown").lower())
465 for name, outcomes in (contents.get("deliveries") or {}).items():
466 if not isinstance(outcomes, dict):
467 continue
468 sent, failed = bool(outcomes.get("success")), bool(outcomes.get("error"))
469 if sent or failed:
470 counts: dict[str, int] = day["deliveries"].setdefault(name, {"success": 0, "failed": 0})
471 counts["success"] += sent
472 counts["failed"] += failed
473 for scenario in _scenario_names(contents.get("enabled_scenarios")):
474 _bump(day["scenarios"], scenario)
475 ordered: list[dict[str, Any]] = [days[key] for key in sorted(days)]
476 return {"days": ordered, "count": sum(day["count"] for day in ordered)}
479def summarize_notification(
480 engine: SupernotifyEngine, contents: dict[str, Any], include_provenance: bool = True
481) -> dict[str, Any]:
482 """Cut an archived or live notification down to what explains what happened to it.
484 `missed` counts the deliveries that were asked for but could not go out - the reason
485 a notification is a `partial_delivery` when nothing failed. `delivery_provenance`, why
486 each delivery was enabled or disabled, is about half of a summary: `enquire_archive`
487 leaves it out of its `summary` verbosity, the assistant tools keep it."""
488 deliveries: dict[str, dict[str, Any]] = {}
489 for name, outcomes in (contents.get("deliveries") or {}).items():
490 if skipped := outcomes.get("skipped"):
491 deliveries[name] = {"skipped": skipped.get("suppression_reason")}
492 continue
493 summary: dict[str, Any] = {}
494 recipients: set[str] = set()
495 for outcome in ("success", "suppressed", "error"):
496 envelopes: list[dict[str, Any]] = outcomes.get(outcome) or []
497 if not envelopes:
498 continue
499 summary[outcome] = len(envelopes)
500 for envelope in envelopes:
501 recipients.update(((envelope.get("target") or {}).get("person_id")) or [])
502 if outcome == "suppressed" and envelope.get("skip_reason"):
503 summary.setdefault("reasons", []).append(envelope["skip_reason"])
504 if outcome == "error":
505 summary.setdefault("errors", []).extend(
506 call.get("exception") for call in envelope.get("failed_calls") or [] if call.get("exception")
507 )
508 if recipients:
509 summary["recipients"] = _names_for(engine, sorted(recipients))
510 deliveries[name] = summary
511 condition_variables: dict[str, Any] = contents.get("condition_variables") or {}
512 result: dict[str, Any] = {
513 "id": contents.get("id"),
514 "created": contents.get("created"),
515 "outcome": contents.get("outcome"),
516 "missed": contents.get("missed") or 0,
517 "message": contents.get("message"),
518 "title": condition_variables.get("notification_title"),
519 "priority": contents.get("priority"),
520 "scenarios": _scenario_names(contents.get("enabled_scenarios")),
521 "occupancy": {
522 state: _names_for(engine, [p.get("person") for p in people if p.get("person")])
523 for state, people in (contents.get("occupancy") or {}).items()
524 },
525 "deliveries": deliveries,
526 }
527 if include_provenance:
528 result["delivery_provenance"] = contents.get("delivery_provenance") or {}
529 if contents.get("unknown_names"):
530 result["unknown_names"] = contents["unknown_names"]
531 if contents.get("_suppression_reason"):
532 result["suppressed"] = contents["_suppression_reason"]
533 if requester := (contents.get("original_context") or {}).get("user"):
534 result["requested_by"] = requester
535 return result