Update Signal alert handling

python/tools/snapshot_manager.py

@@ -6,6 +6,7 @@ import logging
 import sys
 import tomllib
 from functools import cache
+from os import getenv
 from pathlib import Path  # noqa: TC003 This is required for the typer CLI
 from re import compile as re_compile
 from re import search
@@ -24,6 +25,15 @@ def main(config_file: Path) -> None:
     configure_logger(level="DEBUG")
     logger.info("Starting snapshot_manager")
 
+    from_phone = getenv("SIGNAL_ALERT_FROM_PHONE")
+    to_phone = getenv("SIGNAL_ALERT_TO_PHONE")
+    if not from_phone:
+        error = "SIGNAL_ALERT_FROM_PHONE environment variable is not set"
+        raise ValueError(error)
+    if not to_phone:
+        error = "SIGNAL_ALERT_TO_PHONE environment variable is not set"
+        raise ValueError(error)
+
     try:
         time_stamp = get_time_stamp()
 
@@ -33,14 +43,14 @@ def main(config_file: Path) -> None:
             if status != "snapshot created":
                 msg = f"{dataset.name} failed to create snapshot {time_stamp}"
                 logger.error(msg)
-                signal_alert(msg)
+                signal_alert(msg, from_phone, to_phone)
                 continue
             count_lookup = get_count_lookup(config_file, dataset.name)
             logger.info(f"using {count_lookup} for {dataset.name}")
-            get_snapshots_to_delete(dataset, count_lookup)
+            get_snapshots_to_delete(dataset, count_lookup, from_phone, to_phone)
     except Exception:
         logger.exception("snapshot_manager failed")
-        signal_alert("snapshot_manager failed")
+        signal_alert("snapshot_manager failed", from_phone, to_phone)
         sys.exit(1)
     else:
         logger.info("snapshot_manager completed")
@@ -92,12 +102,16 @@ def load_config_data(config_file: Path) -> dict[str, dict[str, int]]:
 def get_snapshots_to_delete(
     dataset: Dataset,
     count_lookup: dict[str, int],
+    from_phone: str,
+    to_phone: str,
 ) -> None:
     """Get snapshots to delete.
 
     Args:
         dataset (Dataset): the dataset
         count_lookup (dict[str, int]): the count lookup
+        from_phone (str): The Signal account sending alerts.
+        to_phone (str): The Signal account receiving alerts.
     """
     snapshots = dataset.get_snapshots()
 
@@ -127,7 +141,7 @@ def get_snapshots_to_delete(
         for snapshot in snapshots_being_deleted:
             if error := dataset.delete_snapshot(snapshot):
                 error_message = f"{dataset.name}@{snapshot} failed to delete: {error}"
-                signal_alert(error_message)
+                signal_alert(error_message, from_phone, to_phone)
                 logger.error(error_message)