add healthchecks.io integration

lib/checkin_handler.py

@@ -3,6 +3,7 @@
 import os
 import signal
 import time
+import requests
 from datetime import datetime, timedelta
 from multiprocessing import Lock, Process
 from typing import TYPE_CHECKING
@@ -152,3 +153,4 @@ def _check_in(self) -> None:
         self.notification_handler.successful_checkin(
             reservation["checkInConfirmationPage"], self.flight
         )
+

lib/config.py

@@ -26,6 +26,7 @@ def __init__(self) -> None:
         self.notification_level = NotificationLevel.INFO
         self.notification_urls = []
         self.retrieval_interval = 24 * 60 * 60
+        self.healthchecks_url = None
 
     def create(self, config_json: JSON, global_config: "GlobalConfig") -> None:
         self._merge_globals(global_config)
@@ -55,6 +56,14 @@ def _parse_config(self, config: JSON) -> None:
             if not isinstance(self.check_fares, bool):
                 raise ConfigError("'check_fares' must be a boolean")
 
+        if "healthchecks_url" in config:
+            self.healthchecks_url = config["healthchecks_url"]
+
+            if not isinstance(self.healthchecks_url, str):
+                raise ConfigError("'healthchecks_url' must be a string")
+
+            logger.debug("A Healthchecks URL has been provided")
+
         if "notification_level" in config:
             notification_level = config["notification_level"]
             try:

lib/fare_checker.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import requests
+
 from typing import TYPE_CHECKING, Any, Dict, List, Tuple
 
 from .checkin_scheduler import VIEW_RESERVATION_URL
@@ -35,6 +37,7 @@ def check_flight_price(self, flight: Flight) -> None:
         amount = int(flight_price["amount"].replace(",", ""))
         price_info = f"{sign}{amount} {flight_price['currencyCode']}"
         logger.debug("Flight price change found for %s", price_info)
+        self.reservation_monitor.notification_handler.healthchecks_success("Successful fare check, confirmation number=" + flight.confirmation_number)
 
         # The Southwest website can report a fare price difference of -1 USD. This is a
         # false positive as no credit is actually received when the flight is changed.

lib/notification_handler.py

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING, Any, Dict, List
 
 import apprise
+import requests
 
 from .flight import Flight
 from .log import get_logger
@@ -117,5 +118,13 @@ def lower_fare(self, flight: Flight, price_info: str) -> None:
         logger.debug("Sending lower fare notification...")
         self.send_notification(message, NotificationLevel.INFO)
 
+    def healthchecks_success(self, data: Any) -> None:
+        if self.reservation_monitor.config.healthchecks_url is not None:
+            requests.post(self.reservation_monitor.config.healthchecks_url, data=data)
+
+    def healthchecks_fail(self, data: Any) -> None:
+        if self.reservation_monitor.config.healthchecks_url is not None:
+            requests.post(self.reservation_monitor.config.healthchecks_url + "/fail", data=data)
+
     def _get_account_name(self) -> str:
         return f"{self.reservation_monitor.first_name} {self.reservation_monitor.last_name}"

lib/reservation_monitor.py

@@ -1,6 +1,7 @@
 import multiprocessing
 import sys
 import time
+import requests
 from datetime import datetime
 from typing import Any, Dict, List, Tuple
 
@@ -99,10 +100,12 @@ def _check_flight_fares(self) -> None:
             try:
                 fare_checker.check_flight_price(flight)
             except RequestError as err:
+                self.notification_handler.healthchecks_fail("Failed fare check, confirmation number=" + flight.confirmation_number)
                 logger.error("Requesting error during fare check. %s. Skipping...", err)
             except FlightChangeError as err:
                 logger.debug("%s. Skipping fare check", err)
             except Exception as err:
+                self.notification_handler.healthchecks_fail("Failed fare check, confirmation number=" + flight.confirmation_number)
                 logger.exception("Unexpected error during fare check: %s", repr(err))
 
     def _smart_sleep(self, previous_time: datetime) -> None: