ping/index.py

from flask import Flask, send_file, redirect, render_template
import threading
import requests
import tomllib
import logging
import socket
import time

with open("config.toml", "rb") as f:
	config = tomllib.load(f)

app = Flask("velping")
logging.disable(logging.CRITICAL)
http = requests.Session()
http.headers["User-Agent"] = "velping"


uptime = {}


def handle_service_status(
	service_name: str,
	down: bool = False,
	error: str | None = None,
):
	if (
		uptime[service_name][config["frontend"]["pings"] - 1] != "O"
		and config["ntfy"]["enabled"]
		and down
	):
		http.post(
			url=f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
			data=f"Service {service_name} is offline",
		)
	elif (
		uptime[service_name][config["frontend"]["pings"] - 1] != "I"
		and config["ntfy"]["enabled"]
		and not down
	):
		http.post(
			url=f"https://{config['ntfy']['server']}/{config['ntfy']['topic']}",
			data=f"Service {service_name} is online",
		)

	uptime[service_name].pop(0)
	if down:
		uptime[service_name].append("O")
		print(f"[E] An error happened while pinging for {service_name}: {error}")
	else:
		uptime[service_name].append("I")
		print(f"[I] Pinging {service_name} succeeded!")


def tcp_ping(service):
	ip = service["ip_version"]
	name = service["name"]
	host = service["hostname"]
	port = service["port"]
	timeout = service["timeout"]

	if ip == 4:
		socket_type = socket.AF_INET
	elif ip == 6:
		socket_type = socket.AF_INET6
	else:
		print(f"[E] IP version for {name} is not 4 or 6")
		return

	while True:
		print(f"[I] Pinging {name}")

		try:
			with socket.socket(family=socket_type, type=socket.SOCK_STREAM) as sock:
				sock.settimeout(timeout)
				sock.connect((host, port))

			handle_service_status(
				service_name=name,
				down=False,
			)

		except socket.timeout:
			handle_service_status(
				service_name=name,
				down=True,
				error=f"Connection to {name} timed out",
			)

		except Exception as err:
			handle_service_status(
				service_name=name,
				down=True,
				error=str(err),
			)

		time.sleep(config["pinging"]["seconds_between_ping"])


def http_ping(service):
	name = service["name"]
	address = service["address"]
	timeout = service["timeout"]

	while True:
		print(f"[I] Pinging {service['name']}")
		try:
			resp = http.get(
				url=address,
				timeout=timeout,
			)
			resp.raise_for_status()

			handle_service_status(
				service_name=name,
				down=False,
			)
		except Exception as e:
			handle_service_status(
				service_name=name,
				down=True,
				error=str(e),
			)

		time.sleep(
			config.get("pinging", {}).get("seconds", 60)
		)


for service_name in config["services"]:
	service = config["services"][service_name]
	if "type" in service and "name" in service and "timeout" in service:
		uptime[service["name"]] = []
		for _ in range(config["frontend"]["pings"]):
			uptime[service["name"]].append("?")

		if service["type"] == "http" and "address" in service:
			threading.Thread(
				target=http_ping,
				args=(service,),
			).start()
		elif (
			service["type"] == "tcp"
			and "port" in service
			and "hostname" in service
			and "ip_version" in service
		):
			threading.Thread(
				target=tcp_ping,
				args=(service,),
			).start()
		else:
			print(
				f"[W] Service {service['name']} is missing required fields, ignoring."
			)
	else:
		print("[W] Invalid service entry found, ignoring.")


@app.errorhandler(404)
def not_found(error):
	return redirect("/", 308)


@app.errorhandler(500)
def internal_error(error):
	return render_template(
		"internal_error.xht",
		config=config,
	), 500


@app.route("/style.css")
def css():
	return send_file("assets/style.css")


@app.route("/")
def root():
	return render_template(
		"index.xht",
		config=config,
		services=config["services"],
		uptime=uptime,
	), 200


app.run(
	port=config["frontend"]["port"],
	host=config["frontend"]["address"],
)