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

1import datetime as dt 

2import logging 

3from abc import abstractmethod 

4from pathlib import Path 

5from typing import Any, cast 

6 

7import aiofiles.os 

8import homeassistant.util.dt as dt_util 

9from homeassistant.helpers.json import save_json 

10 

11_LOGGER = logging.getLogger(__name__) 

12 

13ARCHIVE_PURGE_MIN_INTERVAL = 3 * 60 

14ARCHIVE_DEFAULT_DAYS = 1 

15WRITE_TEST = ".startup" 

16 

17 

18class ArchivableObject: 

19 @abstractmethod 

20 def base_filename(self) -> str: 

21 pass 

22 

23 def contents(self, minimal: bool = False) -> Any: 

24 pass 

25 

26 

27class NotificationArchive: 

28 def __init__(self, archive_path: str | None, archive_days: str | None, purge_minute_interval: str | None = None) -> None: 

29 self.enabled = False 

30 self.last_purge: dt.datetime | None = None 

31 self.configured_archive_path: str | None = archive_path 

32 self.archive_path: Path | None = None 

33 self.archive_days: int = int(archive_days) if archive_days else ARCHIVE_DEFAULT_DAYS 

34 self.purge_minute_interval = int(purge_minute_interval) if purge_minute_interval else ARCHIVE_PURGE_MIN_INTERVAL 

35 

36 def initialize(self) -> None: 

37 if not self.configured_archive_path: 

38 _LOGGER.warning("SUPERNOTIFY archive path not configured") 

39 return 

40 verify_archive_path: Path = Path(cast(str, self.configured_archive_path)) 

41 if verify_archive_path and not verify_archive_path.exists(): 

42 _LOGGER.info("SUPERNOTIFY archive path not found at %s", verify_archive_path) 

43 try: 

44 verify_archive_path.mkdir(parents=True, exist_ok=True) 

45 except Exception as e: 

46 _LOGGER.warning("SUPERNOTIFY archive path %s cannot be created: %s", verify_archive_path, e) 

47 if verify_archive_path and verify_archive_path.exists() and verify_archive_path.is_dir(): 

48 try: 

49 verify_archive_path.joinpath(WRITE_TEST).touch(exist_ok=True) 

50 self.enabled = True 

51 self.archive_path = verify_archive_path 

52 except Exception as e: 

53 _LOGGER.warning("SUPERNOTIFY archive path %s cannot be written: %s", verify_archive_path, e) 

54 else: 

55 _LOGGER.warning("SUPERNOTIFY archive path %s is not a directory or does not exist", verify_archive_path) 

56 

57 async def size(self) -> int: 

58 path = self.archive_path 

59 if path and Path(path).exists(): 

60 return sum(1 for p in await aiofiles.os.listdir(path) if p != WRITE_TEST) 

61 return 0 

62 

63 async def cleanup(self, days: int | None = None, force: bool = False) -> int: 

64 if ( 

65 not force 

66 and self.last_purge is not None 

67 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval) 

68 ): 

69 return 0 

70 days = days or self.archive_days 

71 

72 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=self.archive_days) 

73 cutoff = cutoff.astimezone(dt.UTC) 

74 purged = 0 

75 if self.archive_path and Path(self.archive_path).exists(): 

76 try: 

77 archive = await aiofiles.os.scandir(self.archive_path) 

78 for entry in archive: 

79 if entry.name == ".startup": 

80 continue 

81 if dt_util.utc_from_timestamp(entry.stat().st_ctime) <= cutoff: 

82 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path) 

83 await aiofiles.os.unlink(Path(entry.path)) 

84 purged += 1 

85 except Exception as e: 

86 _LOGGER.warning("SUPERNOTIFY Unable to clean up archive at %s: %s", self.archive_path, e, exc_info=True) 

87 _LOGGER.info("SUPERNOTIFY Purged %s archived notifications for cutoff %s", purged, cutoff) 

88 self.last_purge = dt.datetime.now(dt.UTC) 

89 else: 

90 _LOGGER.debug("SUPERNOTIFY Skipping archive purge for unknown path %s", self.archive_path) 

91 return purged 

92 

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

94 if not self.enabled or not self.archive_path: 

95 return False 

96 archive_path: str = "" 

97 try: 

98 filename = f"{archive_object.base_filename()}.json" 

99 archive_path = str(self.archive_path.joinpath(filename)) 

100 save_json(archive_path, archive_object.contents()) 

101 _LOGGER.debug("SUPERNOTIFY Archived notification %s", archive_path) 

102 return True 

103 except Exception as e: 

104 _LOGGER.warning("SUPERNOTIFY Unable to archived notification: %s", e) 

105 try: 

106 save_json(archive_path, archive_object.contents(minimal=True)) 

107 _LOGGER.debug("SUPERNOTIFY Archived minimal notification %s", archive_path) 

108 return True 

109 except Exception as e2: 

110 _LOGGER.warning("SUPERNOTIFY Unable to archived minimal notification: %s", e2) 

111 return False