pingcheck.py

import requests
import datetime
import time
import logging
import sys
from logging.handlers import TimedRotatingFileHandler
import os
from dotenv import load_dotenv
import warnings
from urllib3.exceptions import InsecureRequestWarning

# Load environment variables from .env file
load_dotenv() 

# grab vars from .env file

## USER-DEFINED VARIABLES ##

# define quiet hours in 24H time
sleep_hours = range (1, 7) # sleeping hours from 1am to 7am

## SERVICE URLS ##
plex_url = os.environ.get("PLEX_URL")
overseerr_url = os.environ.get("OVERSEERR_URL")
tautulli_url = os.environ.get("TAUTULLI_URL")
unraid_url = os.environ.get("UNRAID_URL")
hass_url = os.environ.get("HASS_URL")
pihole_url = os.environ.get("PIHOLE_URL")
homepage_url = os.environ.get("HOMEPAGE_URL")
# example_service = os.environ.get("EXAMPLE_URL")
# add EXAMPLE_URL=https://example.url to .env
# this will be fixed with an interface/tui at some point 

## END USER-DEFINED VARIABLES ##

# Pushover API credentials
pushover_api_token = os.environ.get("PUSHOVER_API_TOKEN")
pushover_user_key = os.environ.get("PUSHOVER_USER_KEY")
pushover_url = "https://api.pushover.net/1/messages.json"

# set up logging
log_formatter = logging.Formatter("%(asctime)s %(levelname)s: %(message)s")
log_handler = TimedRotatingFileHandler(
    "service_monitor.log", when="midnight", interval=1, backupCount=7, encoding="utf-8"
)
log_handler.setFormatter(log_formatter)
log_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(log_handler)
logging.getLogger().setLevel(logging.INFO)

console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s %(levelname)s: %(message)s")
console.setFormatter(formatter)
logging.getLogger().addHandler(console)

# suppress insecure request warnings - common for self-hosted things
warnings.simplefilter("ignore", InsecureRequestWarning)

# define functions
def check_service_status(url):
    try:
        response = requests.get(url, timeout=5, verify=False)
        if response.status_code in [200, 401]:
            return True
        else:
            return False
    except requests.exceptions.RequestException as e:
        logging.error(f"Error checking {url}: {e}")
        return False
    
def send_pushover_notification(message):
    payload = {
        "token": pushover_api_token,
        "user": pushover_user_key,
        "message": message
    }
    try:
        response = requests.post(pushover_url, data=payload)
        if response.status_code != 200:
            logging.error(f"Failed to send pushover notification: {response.status_code} - {response.text}")
            return False
    except requests.exceptions.RequestException as e:
        logging.error(f"Error sending pushover notification: {e}")
        return False
    
# track the last known status for each service
last_status = {}
last_down_time = {}  # Tracks when a service first fell


# main loop
while True:
    # check the status of each service
    services = {
        "Plex": plex_url,
        "Overseerr": overseerr_url,
        "Tautulli": tautulli_url,
        "Baldr": unraid_url,
        "Frigg": hass_url,
        "Vigar": pihole_url,
        "Homepage": homepage_url
        # "Example Service": example_service  # Uncomment and replace with actual service if needed
    }
    current_hour = datetime.datetime.now().hour

    for service_name, service_url in services.items():
        is_up = check_service_status(service_url)
        was_up = last_status.get(service_name, True)
        current_time = time.time()

        if not is_up:
            if was_up:
                last_down_time[service_name] = current_time
                if current_hour not in sleep_hours:
                    logging.info(f"{service_name} has fallen in battle and enters Valhalla ⚔️ Sending raven to the gods...")
                    send_pushover_notification(f"{service_name} has entered Valhalla")
                else:
                    logging.info(f"{service_name} is down, but Odin sleeps — no raven shall fly")
            else:
                # Service is still down — check if it's been over an hour
                down_since = last_down_time.get(service_name)
                if down_since and (current_time - down_since) >= 3600:
                    if current_hour not in sleep_hours:
                        logging.info(f"{service_name} still lingers in Valhalla after 1 hour — sending another raven 🕊️")
                        send_pushover_notification(f"{service_name} is still in Valhalla after 1 hour")
                        # Reset the timer so we don’t get bombarded every loop
                        last_down_time[service_name] = current_time
                    else:
                        logging.info(f"{service_name} still in Valhalla past 1 hour, but night cloaks the realm 🛌")
                else:
                    logging.info(f"{service_name} remains lost in Valhalla — no new raven dispatched 🛡️")
        else:
            if not was_up:
                logging.info(f"{service_name} has returned from the halls of Valhalla — reborn and battle-ready 🔥")
                if current_hour not in sleep_hours:
                    send_pushover_notification(f"{service_name} has returned from Valhalla reborn")
                last_down_time.pop(service_name, None)  # Clear the down time
            else:
                logging.info(f"{service_name} stands vigilant upon the Bifrost 🌈")

        last_status[service_name] = is_up
    logging.info("The ravens rest - all realms checked. Sleeping for 5 minutes...")
    time.sleep(300)