Coverage for custom_components/supernotify/archive.py: 99%

1from __future__ import annotations 

2 

3import datetime as dt 

4import json 

5import logging 

6from abc import abstractmethod 

7from typing import TYPE_CHECKING, Any 

8 

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) 

17 

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 

30 

31if TYPE_CHECKING: 

32 from homeassistant.helpers.typing import ConfigType 

33 

34 from custom_components.supernotify.hass_api import HomeAssistantAPI 

35 

36_LOGGER = logging.getLogger(__name__) 

37 

38ARCHIVE_PURGE_MIN_INTERVAL = 3 * 60 

39ARCHIVE_DEFAULT_DAYS = 1 

40WRITE_TEST = ".startup" 

41 

42 

43class ArchivableObject: 

44 @abstractmethod 

45 def base_filename(self) -> str: 

46 pass 

47 

48 @abstractmethod 

49 def contents(self, diagnostics: bool = False, **_kwargs: Any) -> Any: 

50 pass 

51 

52 def outcome(self) -> DeliveryOutcome: 

53 return DeliveryOutcome.NO_DELIVERY 

54 

55 def selected(self, outcome_policy: OutcomeSelection) -> bool: 

56 if outcome_policy & OutcomeSelection.NONE: 

57 return False 

58 return bool( 

59 outcome_policy & OutcomeSelection.ALL 

60 or (outcome_policy & OutcomeSelection.SUCCESS and self.outcome() == DeliveryOutcome.SUCCESS) 

61 or (outcome_policy & OutcomeSelection.NO_DELIVERY and self.outcome() == DeliveryOutcome.NO_DELIVERY) 

62 or (outcome_policy & OutcomeSelection.PARTIAL_DELIVERY and self.outcome() == DeliveryOutcome.PARTIAL_DELIVERY) 

63 or (outcome_policy & OutcomeSelection.DUPE and self.outcome() == DeliveryOutcome.DUPE) 

64 or (outcome_policy & OutcomeSelection.FALLBACK_DELIVERY and self.outcome() == DeliveryOutcome.FALLBACK_DELIVERY) 

65 or (outcome_policy & OutcomeSelection.ERROR and self.outcome() == DeliveryOutcome.ERROR) 

66 ) 

67 

68 

69class ArchiveDestination: 

70 @abstractmethod 

71 async def archive(self, archive_object: ArchivableObject) -> bool: 

72 pass 

73 

74 

75class EventArchiver(ArchiveDestination): 

76 def __init__( 

77 self, hass_api: HomeAssistantAPI, event_name: str, diagnostics: OutcomeSelection = OutcomeSelection.ERROR 

78 ) -> None: 

79 self.hass_api = hass_api 

80 self.event_name = event_name 

81 self.diagnostics = diagnostics 

82 if diagnostics & OutcomeSelection.NONE: 

83 pass 

84 elif diagnostics & OutcomeSelection.ALL: 

85 _LOGGER.info("SUPERNOTIFY Archiving all notifications as %s events", event_name) 

86 else: 

87 if diagnostics & OutcomeSelection.SUCCESS: 

88 _LOGGER.info("SUPERNOTIFY Archiving successful notifications as %s events", event_name) 

89 if diagnostics & OutcomeSelection.PARTIAL_DELIVERY: 

90 _LOGGER.info("SUPERNOTIFY Archiving partial delivery notifications as %s events", event_name) 

91 

92 if diagnostics & OutcomeSelection.FALLBACK_DELIVERY: 

93 _LOGGER.info("SUPERNOTIFY Archiving fallback notifications as %s events", event_name) 

94 if diagnostics & OutcomeSelection.NO_DELIVERY: 

95 _LOGGER.info("SUPERNOTIFY Archiving no delivery notifications as %s events", event_name) 

96 

97 if diagnostics & OutcomeSelection.ERROR: 

98 _LOGGER.info("SUPERNOTIFY Archiving error notifications as %s events", event_name) 

99 

100 if diagnostics & OutcomeSelection.DUPE: 

101 _LOGGER.info("SUPERNOTIFY Archiving dupe notifications as %s events", event_name) 

102 

103 async def archive(self, archive_object: ArchivableObject) -> bool: 

104 payload = archive_object.contents(diagnostics=archive_object.selected(self.diagnostics)) 

105 self.hass_api.fire_event(self.event_name, payload) 

106 return True 

107 

108 

109class ArchiveTopic(ArchiveDestination): 

110 def __init__( 

111 self, 

112 hass_api: HomeAssistantAPI, 

113 topic: str, 

114 qos: int = 0, 

115 retain: bool = True, 

116 diagnostics: OutcomeSelection = OutcomeSelection.ERROR, 

117 ) -> None: 

118 self.hass_api: HomeAssistantAPI = hass_api 

119 self.topic: str = topic 

120 self.qos: int = qos 

121 self.retain: bool = retain 

122 self.diagnostics: OutcomeSelection = diagnostics 

123 self.enabled: bool = False 

124 

125 async def initialize(self) -> None: 

126 if self.topic: 

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 MQTT not available at startup, disabled" 

133 ) 

134 

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 

152 

153 

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 

162 

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) 

181 

182 async def archive(self, archive_object: ArchivableObject) -> bool: 

183 archived: bool = False 

184 

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: 

206 _LOGGER.exception("SUPERNOTIFY Unable to archive minimal notification") 

207 return archived 

208 

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 

214 

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 

222 

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 

243 

244 

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)) 

265 

266 self.purge_minute_interval = int(config.get(CONF_ARCHIVE_PURGE_INTERVAL, ARCHIVE_PURGE_MIN_INTERVAL)) 

267 

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() 

279 

280 if self.mqtt_topic: 

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() 

283 

284 self.event_archiver = EventArchiver(self.hass_api, self.archive_event_name, self.diagnostics) 

285 

286 async def size(self) -> int: 

287 return await self.archive_directory.size() if self.archive_directory else 0 

288 

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 

292 

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) 

303 

304 return archived