Coverage for custom_components / supernotify / archive.py: 88%
207 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +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)
17from homeassistant.helpers import condition as condition
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 CONF_ARCHIVE_PURGE_INTERVAL,
29)
30from .schema import Outcome, OutcomeSelection
32if TYPE_CHECKING:
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 @abstractmethod
46 def base_filename(self) -> str:
47 pass
49 @abstractmethod
50 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> Any:
51 pass
53 def outcome(self) -> Outcome:
54 return Outcome.NO_DELIVERY
56 def selected(self, outcome_policy: OutcomeSelection) -> bool:
57 if outcome_policy & OutcomeSelection.NONE:
58 return False
59 return bool(
60 outcome_policy & OutcomeSelection.ALL
61 or (outcome_policy & OutcomeSelection.SUCCESS and self.outcome() == Outcome.SUCCESS)
62 or (outcome_policy & OutcomeSelection.NO_DELIVERY and self.outcome() == Outcome.NO_DELIVERY)
63 or (outcome_policy & OutcomeSelection.PARTIAL_DELIVERY and self.outcome() == Outcome.PARTIAL_DELIVERY)
64 or (outcome_policy & OutcomeSelection.DUPE and self.outcome() == Outcome.DUPE)
65 or (outcome_policy & OutcomeSelection.FALLBACK_DELIVERY and self.outcome() == Outcome.FALLBACK_DELIVERY)
66 or (outcome_policy & OutcomeSelection.ERROR and self.outcome() == Outcome.ERROR)
67 )
70class ArchiveDestination:
71 @abstractmethod
72 async def archive(self, archive_object: ArchivableObject) -> bool:
73 pass
76class EventArchiver(ArchiveDestination):
77 def __init__(
78 self, hass_api: HomeAssistantAPI, event_name: str, diagnostics: OutcomeSelection = OutcomeSelection.ERROR
79 ) -> None:
80 self.hass_api = hass_api
81 self.event_name = event_name
82 self.diagnostics = diagnostics
83 if diagnostics & OutcomeSelection.NONE:
84 pass
85 elif diagnostics & OutcomeSelection.ALL:
86 _LOGGER.info("SUPERNOTIFY archiving all notifications as %s events", event_name)
87 else:
88 if diagnostics & OutcomeSelection.SUCCESS:
89 _LOGGER.info("SUPERNOTIFY archiving successful notifications as %s events", event_name)
90 if diagnostics & OutcomeSelection.PARTIAL_DELIVERY:
91 _LOGGER.info("SUPERNOTIFY archiving partial delivery notifications as %s events", event_name)
93 if diagnostics & OutcomeSelection.FALLBACK_DELIVERY:
94 _LOGGER.info("SUPERNOTIFY archiving fallback notifications as %s events", event_name)
95 if diagnostics & OutcomeSelection.NO_DELIVERY:
96 _LOGGER.info("SUPERNOTIFY archiving no delivery notifications as %s events", event_name)
98 if diagnostics & OutcomeSelection.ERROR:
99 _LOGGER.info("SUPERNOTIFY archiving error notifications as %s events", event_name)
101 if diagnostics & OutcomeSelection.DUPE:
102 _LOGGER.info("SUPERNOTIFY archiving dupe notifications as %s events", event_name)
104 async def archive(self, archive_object: ArchivableObject) -> bool:
105 payload = archive_object.contents(diagnostics=archive_object.selected(self.diagnostics))
106 self.hass_api.fire_event(self.event_name, payload)
107 return True
110class ArchiveTopic(ArchiveDestination):
111 def __init__(
112 self,
113 hass_api: HomeAssistantAPI,
114 topic: str,
115 qos: int = 0,
116 retain: bool = True,
117 diagnostics: OutcomeSelection = OutcomeSelection.ERROR,
118 ) -> None:
119 self.hass_api: HomeAssistantAPI = hass_api
120 self.topic: str = topic
121 self.qos: int = qos
122 self.retain: bool = retain
123 self.diagnostics: OutcomeSelection = diagnostics
124 self.enabled: bool = False
126 async def initialize(self) -> None:
127 if await self.hass_api.mqtt_available(raise_on_error=False):
128 _LOGGER.info(f"SUPERNOTIFY Archiving to MQTT topic {self.topic}, qos {self.qos}, retain {self.retain}")
129 self.enabled = True
130 else:
131 _LOGGER.warning(
132 f"SUPERNOTIFY archiving configured for topic {self.topic} but MQTTT not available at startup, disabled"
133 )
135 async def archive(self, archive_object: ArchivableObject) -> bool:
136 if not self.enabled:
137 return False
138 payload = archive_object.contents(diagnostics=archive_object.selected(self.diagnostics))
139 topic = f"{self.topic}/{archive_object.base_filename()}"
140 _LOGGER.debug(f"SUPERNOTIFY Publishing notification to {topic}")
141 try:
142 await self.hass_api.mqtt_publish(
143 topic=topic,
144 payload=payload,
145 qos=self.qos,
146 retain=self.retain,
147 )
148 return True
149 except Exception:
150 _LOGGER.warning(f"SUPERNOTIFY failed to archive to topic {self.topic}")
151 return False
154class ArchiveDirectory(ArchiveDestination):
155 def __init__(self, path: str, purge_minute_interval: int, diagnostics: OutcomeSelection = OutcomeSelection.ERROR) -> None:
156 self.configured_path: str = path
157 self.archive_path: anyio.Path | None = None
158 self.enabled: bool = False
159 self.diagnostics: OutcomeSelection = diagnostics
160 self.last_purge: dt.datetime | None = None
161 self.purge_minute_interval: int = purge_minute_interval
163 async def initialize(self) -> None:
164 verify_archive_path: Path = Path(self.configured_path)
165 if verify_archive_path and not await verify_archive_path.exists():
166 _LOGGER.info("SUPERNOTIFY archive path not found at %s", verify_archive_path)
167 try:
168 await verify_archive_path.mkdir(parents=True, exist_ok=True)
169 except Exception as e:
170 _LOGGER.warning("SUPERNOTIFY archive path %s cannot be created: %s", verify_archive_path, e)
171 if verify_archive_path and await verify_archive_path.exists() and await verify_archive_path.is_dir():
172 try:
173 await verify_archive_path.joinpath(WRITE_TEST).touch(exist_ok=True)
174 self.archive_path = verify_archive_path
175 _LOGGER.info("SUPERNOTIFY archiving notifications to file system at %s", verify_archive_path)
176 self.enabled = True
177 except Exception as e:
178 _LOGGER.warning("SUPERNOTIFY archive path %s cannot be written: %s", verify_archive_path, e)
179 else:
180 _LOGGER.warning("SUPERNOTIFY archive path %s is not a directory or does not exist", verify_archive_path)
182 async def archive(self, archive_object: ArchivableObject) -> bool:
183 archived: bool = False
185 if self.enabled and self.archive_path: # archive_path to assuage mypy
186 archive_filepath: Path | None = None
187 diagnostics: bool = archive_object.selected(self.diagnostics)
188 try:
189 filename = f"{archive_object.base_filename()}.json"
190 archive_filepath = self.archive_path.joinpath(filename)
191 serialized: str = json.dumps(archive_object.contents(diagnostics=diagnostics), indent=2)
192 async with aiofiles.open(archive_filepath, mode="w") as file:
193 await file.write(serialized)
194 _LOGGER.debug("SUPERNOTIFY Archived notification %s", await archive_filepath.absolute())
195 archived = True
196 except Exception as e:
197 _LOGGER.warning("SUPERNOTIFY Unable to archive notification: %s", e)
198 if diagnostics and archive_filepath:
199 try:
200 serialized = json.dumps(archive_object.contents(diagnostics=False), indent=2)
201 async with aiofiles.open(archive_filepath, mode="w") as file:
202 await file.write(serialized)
203 _LOGGER.warning("SUPERNOTIFY Archived minimal notification %s", await archive_filepath.absolute())
204 archived = True
205 except Exception as e2:
206 _LOGGER.exception("SUPERNOTIFY Unable to archive minimal notification: %s", e2)
207 return archived
209 async def size(self) -> int:
210 path = self.archive_path
211 if path and await path.exists():
212 return sum(1 for p in await aiofiles.os.listdir(path) if p != WRITE_TEST)
213 return 0
215 async def cleanup(self, days: int, force: bool) -> int:
216 if (
217 not force
218 and self.last_purge is not None
219 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval)
220 ):
221 return 0
223 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
224 cutoff = cutoff.astimezone(dt.UTC)
225 purged = 0
226 if self.archive_path and await self.archive_path.exists():
227 try:
228 archive = await aiofiles.os.scandir(self.archive_path)
229 for entry in archive:
230 if entry.name == ".startup":
231 continue
232 if dt_util.utc_from_timestamp(entry.stat().st_ctime) <= cutoff:
233 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
234 await aiofiles.os.unlink(entry.path)
235 purged += 1
236 except Exception as e:
237 _LOGGER.warning("SUPERNOTIFY Unable to clean up archive at %s: %s", self.archive_path, e, exc_info=True)
238 _LOGGER.info("SUPERNOTIFY Purged %s archived notifications for cutoff %s", purged, cutoff)
239 self.last_purge = dt.datetime.now(dt.UTC)
240 else:
241 _LOGGER.debug("SUPERNOTIFY Skipping archive purge for unknown path %s", self.archive_path)
242 return purged
245class NotificationArchive:
246 def __init__(
247 self,
248 config: ConfigType,
249 hass_api: HomeAssistantAPI,
250 ) -> None:
251 self.hass_api = hass_api
252 self.enabled = bool(config.get(CONF_ENABLED, False))
253 self.archive_directory: ArchiveDirectory | None = None
254 self.archive_topic: ArchiveTopic | None = None
255 self.event_archiver: EventArchiver | None = None
256 self.event_selection: OutcomeSelection = config.get(CONF_ARCHIVE_EVENT_SELECTION, OutcomeSelection.NONE)
257 self.diagnostics: OutcomeSelection = config.get(CONF_ARCHIVE_DIAGNOSTICS, OutcomeSelection.ERROR)
258 self.archive_event_name: str = config.get(CONF_ARCHIVE_EVENT_NAME, "supernotification")
259 self.configured_archive_path: str | None = config.get(CONF_ARCHIVE_PATH)
260 self.archive_days = int(config.get(CONF_ARCHIVE_DAYS, ARCHIVE_DEFAULT_DAYS))
261 self.mqtt_topic: str | None = config.get(CONF_ARCHIVE_MQTT_TOPIC)
262 self.mqtt_qos: int = int(config.get(CONF_ARCHIVE_MQTT_QOS, 0))
263 self.mqtt_retain: bool = bool(config.get(CONF_ARCHIVE_MQTT_RETAIN, True))
264 self.debug: bool = bool(config.get(CONF_DEBUG, False))
266 self.purge_minute_interval = int(config.get(CONF_ARCHIVE_PURGE_INTERVAL, ARCHIVE_PURGE_MIN_INTERVAL))
268 async def initialize(self) -> None:
269 if not self.enabled:
270 _LOGGER.info("SUPERNOTIFY Archive disabled")
271 return
272 if not self.configured_archive_path:
273 _LOGGER.warning("SUPERNOTIFY archive path not configured")
274 else:
275 self.archive_directory = ArchiveDirectory(
276 self.configured_archive_path, purge_minute_interval=self.purge_minute_interval, diagnostics=self.diagnostics
277 )
278 await self.archive_directory.initialize()
280 if self.mqtt_topic is not None:
281 self.archive_topic = ArchiveTopic(self.hass_api, self.mqtt_topic, self.mqtt_qos, self.mqtt_retain, self.diagnostics)
282 await self.archive_topic.initialize()
284 self.event_archiver = EventArchiver(self.hass_api, self.archive_event_name, self.diagnostics)
286 async def size(self) -> int:
287 return await self.archive_directory.size() if self.archive_directory else 0
289 async def cleanup(self, days: int | None = None, force: bool = False) -> int:
290 days = days or self.archive_days
291 return await self.archive_directory.cleanup(days, force) if self.archive_directory else 0
293 async def archive(self, archive_object: ArchivableObject) -> bool:
294 archived: bool = False
295 if self.archive_topic:
296 if await self.archive_topic.archive(archive_object):
297 archived = True
298 if self.archive_directory:
299 if await self.archive_directory.archive(archive_object):
300 archived = True
301 if self.event_archiver and archive_object.selected(self.event_selection):
302 await self.event_archiver.archive(archive_object)
304 return archived