Coverage for custom_components/supernotify/archive.py: 94%
270 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 21:14 +0000
1from __future__ import annotations
3import datetime as dt
4import json
5import logging
6from abc import abstractmethod
7from typing import TYPE_CHECKING, Any
9import aiofiles.os
10import anyio
11import homeassistant.util.dt as dt_util
12from anyio import Path
13from homeassistant.const import (
14 CONF_DEBUG,
15 CONF_ENABLED,
16)
18from .const import (
19 CONF_ARCHIVE_DAYS,
20 CONF_ARCHIVE_DIAGNOSTICS,
21 CONF_ARCHIVE_EVENT_NAME,
22 CONF_ARCHIVE_EVENT_SELECTION,
23 CONF_ARCHIVE_MQTT_QOS,
24 CONF_ARCHIVE_MQTT_RETAIN,
25 CONF_ARCHIVE_MQTT_TOPIC,
26 CONF_ARCHIVE_PATH,
27 CONF_ARCHIVE_PURGE_INTERVAL,
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_LOGGER = logging.getLogger(__name__)
39ARCHIVE_PURGE_MIN_INTERVAL = 3 * 60
40ARCHIVE_DEFAULT_DAYS = 1
41WRITE_TEST = ".startup"
44class ArchivableObject:
45 ha_context: HAContext | None = None
47 @abstractmethod
48 def base_filename(self) -> str:
49 pass
51 @abstractmethod
52 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> dict[str, Any]:
53 pass
55 def outcome(self) -> DeliveryOutcome:
56 return DeliveryOutcome.NO_DELIVERY
58 def diagnostics_selected(self, outcome_policy: OutcomeSelection) -> bool:
59 """Whether the archived copy should carry the full diagnostic content"""
60 return self.selected(outcome_policy)
62 def selected(self, outcome_policy: OutcomeSelection) -> bool:
63 if outcome_policy & OutcomeSelection.NONE:
64 return False
65 return bool(
66 outcome_policy & OutcomeSelection.ALL
67 or (outcome_policy & OutcomeSelection.SUCCESS and self.outcome() == DeliveryOutcome.SUCCESS)
68 or (outcome_policy & OutcomeSelection.NO_DELIVERY and self.outcome() == DeliveryOutcome.NO_DELIVERY)
69 or (outcome_policy & OutcomeSelection.PARTIAL_DELIVERY and self.outcome() == DeliveryOutcome.PARTIAL_DELIVERY)
70 or (outcome_policy & OutcomeSelection.DUPE and self.outcome() == DeliveryOutcome.DUPE)
71 or (outcome_policy & OutcomeSelection.FALLBACK_DELIVERY and self.outcome() == DeliveryOutcome.FALLBACK_DELIVERY)
72 or (outcome_policy & OutcomeSelection.ERROR and self.outcome() == DeliveryOutcome.ERROR)
73 )
76class ArchiveDestination:
77 @abstractmethod
78 async def archive(self, archive_object: ArchivableObject) -> bool:
79 pass
82class EventArchiver(ArchiveDestination):
83 def __init__(
84 self, hass_api: HomeAssistantAPI, event_name: str, diagnostics: OutcomeSelection = OutcomeSelection.ERROR
85 ) -> None:
86 self.hass_api = hass_api
87 self.event_name = event_name
88 self.diagnostics = diagnostics
89 if diagnostics & OutcomeSelection.NONE:
90 pass
91 elif diagnostics & OutcomeSelection.ALL:
92 _LOGGER.info("SUPERNOTIFY Archiving all notifications as %s events", event_name)
93 else:
94 if diagnostics & OutcomeSelection.SUCCESS:
95 _LOGGER.info("SUPERNOTIFY Archiving successful notifications as %s events", event_name)
96 if diagnostics & OutcomeSelection.PARTIAL_DELIVERY:
97 _LOGGER.info("SUPERNOTIFY Archiving partial delivery notifications as %s events", event_name)
99 if diagnostics & OutcomeSelection.FALLBACK_DELIVERY:
100 _LOGGER.info("SUPERNOTIFY Archiving fallback notifications as %s events", event_name)
101 if diagnostics & OutcomeSelection.NO_DELIVERY:
102 _LOGGER.info("SUPERNOTIFY Archiving no delivery notifications as %s events", event_name)
104 if diagnostics & OutcomeSelection.ERROR:
105 _LOGGER.info("SUPERNOTIFY Archiving error notifications as %s events", event_name)
107 if diagnostics & OutcomeSelection.DUPE:
108 _LOGGER.info("SUPERNOTIFY Archiving dupe notifications as %s events", event_name)
110 async def archive(self, archive_object: ArchivableObject) -> bool:
111 try:
112 payload = archive_object.contents(diagnostics=archive_object.diagnostics_selected(self.diagnostics))
113 self.hass_api.fire_event(self.event_name, payload, context=archive_object.ha_context)
114 return True
115 except Exception:
116 _LOGGER.warning(f"SUPERNOTIFY Failed to archive to event {self.event_name}")
117 return False
120class ArchiveTopic(ArchiveDestination):
121 def __init__(
122 self,
123 hass_api: HomeAssistantAPI,
124 topic: str,
125 qos: int = 0,
126 retain: bool = True,
127 diagnostics: OutcomeSelection = OutcomeSelection.ERROR,
128 ) -> None:
129 self.hass_api: HomeAssistantAPI = hass_api
130 self.topic: str = topic
131 self.qos: int = qos
132 self.retain: bool = retain
133 self.diagnostics: OutcomeSelection = diagnostics
134 self.enabled: bool = False
136 async def initialize(self) -> None:
137 if self.topic:
138 if await self.hass_api.mqtt_available(raise_on_error=False):
139 _LOGGER.info(f"SUPERNOTIFY Archiving to MQTT topic {self.topic}, qos {self.qos}, retain {self.retain}")
140 self.enabled = True
141 else:
142 _LOGGER.warning(
143 f"SUPERNOTIFY Archiving configured for topic {self.topic} but MQTT not available at startup, disabled"
144 )
146 async def archive(self, archive_object: ArchivableObject) -> bool:
147 if not self.enabled:
148 return False
149 payload = archive_object.contents(diagnostics=archive_object.diagnostics_selected(self.diagnostics))
150 topic = f"{self.topic}/{archive_object.base_filename()}"
151 _LOGGER.debug(f"SUPERNOTIFY Publishing notification to {topic}")
152 try:
153 await self.hass_api.mqtt_publish(
154 topic=topic,
155 payload=payload,
156 qos=self.qos,
157 retain=self.retain,
158 )
159 return True
160 except Exception:
161 _LOGGER.warning(f"SUPERNOTIFY Failed to archive to topic {self.topic}")
162 return False
165class ArchiveDirectory(ArchiveDestination):
166 def __init__(self, path: str, purge_minute_interval: int, diagnostics: OutcomeSelection = OutcomeSelection.ERROR) -> None:
167 self.configured_path: str = path
168 self.archive_path: anyio.Path | None = None
169 self.enabled: bool = False
170 self.diagnostics: OutcomeSelection = diagnostics
171 self.last_purge: dt.datetime | None = None
172 self.purge_minute_interval: int = purge_minute_interval
174 async def initialize(self) -> None:
175 verify_archive_path: Path = Path(self.configured_path)
176 if verify_archive_path and not await verify_archive_path.exists():
177 _LOGGER.info("SUPERNOTIFY Archive path not found at %s", verify_archive_path)
178 try:
179 await verify_archive_path.mkdir(parents=True, exist_ok=True)
180 except Exception as e:
181 _LOGGER.warning("SUPERNOTIFY Archive path %s cannot be created: %s", verify_archive_path, e)
182 if verify_archive_path and await verify_archive_path.exists() and await verify_archive_path.is_dir():
183 try:
184 await verify_archive_path.joinpath(WRITE_TEST).touch(exist_ok=True)
185 self.archive_path = verify_archive_path
186 _LOGGER.info("SUPERNOTIFY Archiving notifications to file system at %s", verify_archive_path)
187 self.enabled = True
188 except Exception as e:
189 _LOGGER.warning("SUPERNOTIFY Archive path %s cannot be written: %s", verify_archive_path, e)
190 else:
191 _LOGGER.warning("SUPERNOTIFY Archive path %s is not a directory or does not exist", verify_archive_path)
193 async def archive(self, archive_object: ArchivableObject) -> bool:
194 archived: bool = False
196 if self.enabled and self.archive_path: # archive_path to assuage mypy
197 archive_filepath: Path | None = None
198 diagnostics: bool = archive_object.diagnostics_selected(self.diagnostics)
199 try:
200 filename = f"{archive_object.base_filename()}.json"
201 archive_filepath = self.archive_path.joinpath(filename)
202 serialized: str = json.dumps(archive_object.contents(diagnostics=diagnostics), indent=2)
203 async with aiofiles.open(archive_filepath, mode="w") as file:
204 await file.write(serialized)
205 _LOGGER.debug("SUPERNOTIFY Archived notification %s", await archive_filepath.absolute())
206 archived = True
207 except Exception as e:
208 _LOGGER.warning("SUPERNOTIFY Unable to archive notification: %s", e)
209 if diagnostics and archive_filepath:
210 try:
211 serialized = json.dumps(archive_object.contents(diagnostics=False), indent=2)
212 async with aiofiles.open(archive_filepath, mode="w") as file:
213 await file.write(serialized)
214 _LOGGER.warning("SUPERNOTIFY Archived minimal notification %s", await archive_filepath.absolute())
215 archived = True
216 except Exception:
217 _LOGGER.exception("SUPERNOTIFY Unable to archive minimal notification")
218 return archived
220 async def list_entries(
221 self,
222 limit: int = 20,
223 after: dt.datetime | None = None,
224 before: dt.datetime | None = None,
225 outcome: str | None = None,
226 ) -> list[dict[str, Any]]:
227 """Return up to *limit* archived notifications, newest first.
229 Optional *after* and *before* filter by the ``created`` timestamp stored in
230 each archive file. Optional *outcome* keeps only notifications whose top-level
231 ``outcome`` field matches (case-sensitive, e.g. ``"SUCCESS"``).
232 """
233 if not self.archive_path or not await self.archive_path.exists():
234 return []
235 entries: list[dict[str, Any]] = []
236 try:
237 raw_scan = await aiofiles.os.scandir(self.archive_path)
238 files = [e for e in raw_scan if e.name.endswith(".json") and e.name != WRITE_TEST]
239 files.sort(key=lambda e: e.stat().st_ctime, reverse=True)
240 for entry in files:
241 if len(entries) >= limit:
242 break
243 try:
244 async with aiofiles.open(entry.path, mode="r") as fh:
245 data: dict[str, Any] = json.loads(await fh.read())
246 except Exception as exc:
247 _LOGGER.debug("SUPERNOTIFY Skipping unreadable archive file %s: %s", entry.name, exc)
248 continue
249 created_raw: str | None = data.get("created")
250 if after is not None and created_raw is not None and created_raw < after.isoformat():
251 continue
252 if before is not None and created_raw is not None and created_raw > before.isoformat():
253 continue
254 if outcome is not None and data.get("outcome") != outcome:
255 continue
256 entries.append(data)
257 except Exception as exc:
258 _LOGGER.warning("SUPERNOTIFY Unable to list archive entries: %s", exc)
259 return entries
261 async def read_entry(self, notification_id: str) -> dict[str, Any] | None:
262 """Return the full JSON of one archived notification by its id, or ``None``."""
263 if not self.archive_path or not await self.archive_path.exists():
264 return None
265 try:
266 raw_scan = await aiofiles.os.scandir(self.archive_path)
267 for entry in raw_scan:
268 if entry.name.endswith(".json") and notification_id in entry.name:
269 async with aiofiles.open(entry.path, mode="r") as fh:
270 return json.loads(await fh.read())
271 except Exception as exc:
272 _LOGGER.warning("SUPERNOTIFY Unable to read archive entry %s: %s", notification_id, exc)
273 return None
275 async def size(self) -> int:
276 path = self.archive_path
277 if path and await path.exists():
278 return sum(1 for p in await aiofiles.os.listdir(path) if p != WRITE_TEST)
279 return 0
281 async def recent(self, since: dt.datetime, limit: int) -> list[dict[str, Any]]:
282 """Archived notifications written since a given time, newest first"""
283 if not self.archive_path or not await self.archive_path.exists():
284 return []
285 cutoff: float = since.timestamp()
286 candidates: list[tuple[float, str]] = []
287 for entry in await aiofiles.os.scandir(self.archive_path):
288 if entry.name.endswith(".json") and (modified := entry.stat().st_mtime) >= cutoff:
289 candidates.append((modified, entry.path))
290 results: list[dict[str, Any]] = []
291 for _modified, path in sorted(candidates, reverse=True)[:limit]:
292 try:
293 async with aiofiles.open(path) as file:
294 results.append(json.loads(await file.read()))
295 except (OSError, ValueError) as e:
296 _LOGGER.warning("SUPERNOTIFY Unable to read archived notification %s: %s", path, e)
297 return results
299 async def cleanup(self, days: int, force: bool) -> int:
300 if (
301 not force
302 and self.last_purge is not None
303 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval)
304 ):
305 return 0
307 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
308 cutoff = cutoff.astimezone(dt.UTC)
309 purged = 0
310 if self.archive_path and await self.archive_path.exists():
311 try:
312 archive = await aiofiles.os.scandir(self.archive_path)
313 for entry in archive:
314 if entry.name == ".startup":
315 continue
316 # st_ctime is used deliberately here (not st_birthtime): archive files are
317 # written once and never modified afterwards, so ctime reflects creation time
318 # on the platforms this integration targets; st_birthtime is not guaranteed to
319 # be available on all Linux filesystems.
320 if dt_util.utc_from_timestamp(entry.stat().st_ctime) <= cutoff: # ty: ignore[deprecated,unused-ignore-comment]
321 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
322 await aiofiles.os.unlink(entry.path)
323 purged += 1
324 except Exception as e:
325 _LOGGER.warning("SUPERNOTIFY Unable to clean up archive at %s: %s", self.archive_path, e, exc_info=True)
326 _LOGGER.info("SUPERNOTIFY Purged %s archived notifications for cutoff %s", purged, cutoff)
327 self.last_purge = dt.datetime.now(dt.UTC)
328 else:
329 _LOGGER.debug("SUPERNOTIFY Skipping archive purge for unknown path %s", self.archive_path)
330 return purged
333class NotificationArchive:
334 def __init__(
335 self,
336 config: ConfigType,
337 hass_api: HomeAssistantAPI,
338 ) -> None:
339 self.hass_api = hass_api
340 self.enabled = bool(config.get(CONF_ENABLED, False))
341 self.archive_directory: ArchiveDirectory | None = None
342 self.archive_topic: ArchiveTopic | None = None
343 self.event_archiver: EventArchiver | None = None
344 self.event_selection: OutcomeSelection = config.get(CONF_ARCHIVE_EVENT_SELECTION, OutcomeSelection.NONE)
345 self.diagnostics: OutcomeSelection = config.get(CONF_ARCHIVE_DIAGNOSTICS, OutcomeSelection.ERROR)
346 self.archive_event_name: str = config.get(CONF_ARCHIVE_EVENT_NAME, "supernotification")
347 self.configured_archive_path: str | None = config.get(CONF_ARCHIVE_PATH)
348 self.archive_days = int(config.get(CONF_ARCHIVE_DAYS, ARCHIVE_DEFAULT_DAYS))
349 self.mqtt_topic: str | None = config.get(CONF_ARCHIVE_MQTT_TOPIC)
350 self.mqtt_qos: int = int(config.get(CONF_ARCHIVE_MQTT_QOS, 0))
351 self.mqtt_retain: bool = bool(config.get(CONF_ARCHIVE_MQTT_RETAIN, True))
352 self.debug: bool = bool(config.get(CONF_DEBUG, False))
354 self.purge_minute_interval = int(config.get(CONF_ARCHIVE_PURGE_INTERVAL, ARCHIVE_PURGE_MIN_INTERVAL))
356 async def initialize(self) -> None:
357 if not self.enabled:
358 _LOGGER.info("SUPERNOTIFY Archive disabled")
359 return
360 if not self.configured_archive_path:
361 _LOGGER.warning("SUPERNOTIFY Archive path not configured")
362 else:
363 self.archive_directory = ArchiveDirectory(
364 self.configured_archive_path, purge_minute_interval=self.purge_minute_interval, diagnostics=self.diagnostics
365 )
366 await self.archive_directory.initialize()
368 if self.mqtt_topic:
369 self.archive_topic = ArchiveTopic(self.hass_api, self.mqtt_topic, self.mqtt_qos, self.mqtt_retain, self.diagnostics)
370 await self.archive_topic.initialize()
372 self.event_archiver = EventArchiver(self.hass_api, self.archive_event_name, self.diagnostics)
374 async def size(self) -> int:
375 return await self.archive_directory.size() if self.archive_directory else 0
377 async def recent(self, since: dt.datetime, limit: int) -> list[dict[str, Any]]:
378 return await self.archive_directory.recent(since, limit) if self.archive_directory else []
380 async def cleanup(self, days: int | None = None, force: bool = False) -> int:
381 days = days or self.archive_days
382 return await self.archive_directory.cleanup(days, force) if self.archive_directory else 0
384 async def archive(self, archive_object: ArchivableObject) -> bool:
385 archived: bool = False
386 if self.archive_topic and await self.archive_topic.archive(archive_object):
387 archived = True
388 if self.archive_directory and await self.archive_directory.archive(archive_object):
389 archived = True
390 if self.event_archiver and archive_object.selected(self.event_selection):
391 await self.event_archiver.archive(archive_object)
393 return archived