Ping checker, with email alerts and reminder when host is down

pingchecker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# All3kcis 2020

from time import sleep
import platform
import datetime
import smtplib
import signal
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from subprocess import PIPE, call



class pingStatus():

	hosts = {
		'8.8.8.8' : 'Google DNS',
	}
	email = {
		'from':'noreply@yourdomain.com',
		'to':'name@yourdomain.com',
	}
	notifDelay = 60*30 # In seconds, default : 60*30, Delay for new email when host is still down
	checkDelay = 60*5  # In seconds, default : 60*5, Interval of pings
	debug = False

	

	# --------------------------------------------------------------#
	#             Do not update values after that !!				#
	# --------------------------------------------------------------#
	status = {}
	first = True
	status_html = {
		'up' : '<span style="font-weight:bold;color:green;">up</span>',
		'down' : '<span style="font-weight:bold;color:red;">down</span>'
	}

	def start(self):
		self.init()
		while 1:
			self.work()
			sleep(self.checkDelay)

	def init(self):
		signal.signal(signal.SIGINT, self.exit_gracefully)
		signal.signal(signal.SIGTERM, self.exit_gracefully)
		self.s = smtplib.SMTP(host='127.0.0.1', port=25)
		for host, alias in self.hosts.items():
			#if not host in self.status:
			self.status[host] = {'last_status':'down','status':'','last_up':'', 'last_verify_down':''} 

	def work(self):
		notif = ''
		host_counter=0
		up_counter=0
		down_counter=0
		sendEmail=False
		if self.first:
			notif += 'Starting ping checker !<br>'
			notif += 'Date : %s<br>'%(datetime.datetime.now())

		for host, alias in self.hosts.items():
			#self.status[host]['last_check'] = datetime.datetime.now()
			
			host_counter+=1
			if self.ping(host):
				status = 'up'
				up_counter+=1
				self.status[host]['last_up'] = datetime.datetime.now()
			else:
				status = 'down'
				down_counter+=1

			if not self.first:
				self.status[host]['last_status'] = self.status[host]['status']
			self.status[host]['status'] = status

			if self.first:
				# Initial state, notifi for actual status
				sendEmail=True
				self.status[host]['last_verify_down'] = datetime.datetime.now() # Init for first
				if self.debug:
					print ('Initial state,',host, 'is', status)
				notif += 'Initial state for host {alias} ({host}) is {status}<br>'.format(alias=alias, host=host, status=self.status_html[status])
				

			elif self.status[host]['last_status'] != self.status[host]['status']:
				# State changed, notifie for
				sendEmail=True
				if self.debug:
					print ('State changed for,',host, 'to', status)
				notif += 'State changed for {alias} ({host}) is now {status}<br>'.format(alias=alias, host=host, status=self.status_html[status])
				
			elif self.status[host]['status'] == 'down' and ( (not self.status[host]['last_up'] and ((datetime.datetime.now() - self.status[host]['last_verify_down']).total_seconds()) > self.notifDelay) or (self.status[host]['last_up'] and ((datetime.datetime.now() - self.status[host]['last_up']).total_seconds()) > self.notifDelay)):
				sendEmail=True
				self.status[host]['last_verify_down'] = datetime.datetime.now()
				if not self.status[host]['last_up']:
					if self.debug:
						print ('Host,',host, 'is still', status, '(Never up !)')
					notif += '{alias} ({host}) is still {status} (Never up !)<br>'.format(alias=alias, host=host, status=self.status_html[status])
				else:
					if self.debug:
						print ('Host,',host, 'is still', status)
					notif += '{alias} ({host}) is still {status}<br>'.format(alias=alias, host=host, status=self.status_html[status])
			else:
				notif += '{alias} ({host}) is {status}<br>'.format(alias=alias, host=host, status=self.status_html[status])
				if self.debug:
					print ('.', end='', flush=True)
			
		self.first = False
		notif += '<br>hosts:%s<br>'%(host_counter)
		notif += 'hosts_up:%s<br>'%(up_counter)
		notif += 'hosts_down:%s<br>'%(down_counter)
		if sendEmail:
			self.sendEmail(notif)
	
	def sendEmail(self, text):
		msg = MIMEMultipart()       # create a message
		# add in the actual person name to the message template
		message = '' # \r\n
		message += text
		# setup the parameters of the message
		msg['From']=self.email['from']
		msg['To']=self.email['to']
		msg['Subject']="Ping Checker status"
		msg.attach(MIMEText(message, 'html'))
		self.s.send_message(msg)
		del msg	

	def ping(self, host):
		"""
		Returns True if host (str) responds to a ping request.
		Remember that a host may not respond to a ping (ICMP) request even if the host name is valid.
		"""

		# Option for the number of packets as a function of
		param = '-n' if platform.system().lower()=='windows' else '-c'

		# Building the command. Ex: "ping -c 1 google.com"
		command = ['ping', param, '1', host]

		return call(command, stdout=PIPE, stderr=PIPE, universal_newlines=True) == 0

	def stop(self):
		print('Stopping ping checker...')
		notif = 'Stopping ping checker...<br>'
		notif += 'Date : %s<br>'%(datetime.datetime.now())
		for host, alias in self.hosts.items():
			if self.debug:
				print ('Last state for, ',host, ' is ', self.status[host]['status'])
			notif += 'Last state for {alias} ({host}) is {status}<br>'.format(alias=alias, host=host, status=self.status_html[self.status[host]['status']])
		if notif:
			self.sendEmail(notif)

	def exit_gracefully(self, signum, frame):
		raise(SystemExit)

pingdaemon = pingStatus()

try:
	# Starting app
	pingdaemon.start()
except (KeyboardInterrupt, SystemExit):
	pingdaemon.stop()

pingchecker.service

#/etc/systemd/system/pingchecker.service
[Unit]
Description= Ping Checker
After=network.target

[Service]
Type=simple
Restart=always
ExecStart=/usr/bin/python3 /usr/local/bin/pingchecker.py

[Install]
WantedBy=multi-user.target