Niriel/rvo_provider_monitoring.py

## rvo_provider_monitoring.py
from http.server import BaseHTTPRequestHandler, HTTPServer
from datetime import datetime
import urllib.request
import urllib.error
from bs4 import BeautifulSoup


TIMEOUT = 2.0  # seconds
HOST_NAME = "localhost"
SERVER_PORT = 8080


def basic_load(url):
    try:
        response = urllib.request.urlopen(url, timeout=TIMEOUT)
    except urllib.error.HTTPError as e:
        response = e
    except urllib.error.URLError:
        response = None
    if response is None:
        return None
    if response.status != 200:
        return None
    content = response.read()
    return content


def kpn_load():
    return basic_load("https://www.kpn.com/zakelijk/eherkenning/service.htm")


def kpn_parse(content):
    if content is None:
        return None
    result = {}
    soup = BeautifulSoup(content, "html.parser")
    components = soup.find_all(**{"class": "notification-head"})
    for component_ in components:
        pp = component_.parent.parent
        status = pp['class'][1]
        component_ok = None
        if status == "success":
            component_ok = True
        elif status == "error":
            component_ok = False
        component_name = component_.text
        result[component_name] = component_ok
    return result


def ddy_load():
    return basic_load("https://ddy.statuspage.io/")


def ddy_parse(content):
    if content is None:
        return None
    result = {}
    soup = BeautifulSoup(content, "html.parser")
    components = soup.find_all(**{"class": "component-inner-container"})
    for component_ in components:
        status = component_['class'][1]
        component_ok = None
        if status == "status-green":
            component_ok = True
        elif status == "status-red":  # I think?
            component_ok = False
        component_name = component_.span.text.strip()
        result[component_name] = component_ok
    return result


def qvd_load():
    return basic_load("https://support.qv-eherkenning.nl/storing-onderhoud/")


def qvd_parse(content):
    if content is None:
        return None
    result = {}
    soup = BeautifulSoup(content, "html.parser")
    h2s = soup.find_all("h2")
    problems = None
    for h2 in h2s:
        if h2.text == "Storing":
            problems = h2.find_next("ul")
            break
        else:
            return result
    problem = problems.li
    while problem:
        problem_component = problem.span.text
        if problem_component != "Op dit moment zijn er geen storingen bij ons bekend.":
            result[problem_component] = False
        problem = problem.find_next_sibling("li")
    return result


def monitor():
    time_start = datetime.now()
    results = {
        "KPN": kpn_parse(kpn_load()),
        "Digidentity": ddy_parse(ddy_load()),
        "QuoVadis": qvd_parse(qvd_load()),
    }
    time_stop = datetime.now()
    query_time = time_stop - time_start

    lines = ["<!DOCTYPE html>", "<html>", "<body>",
             f"<p>Query beginning: {time_start.isoformat()}</p>",
             f"<p>Query end: {time_stop.isoformat()}</p>",
             f"<p>Query duration: {query_time}</p>",
             ]

    for site, components_ok in results.items():
        lines.append(f"<h1>{site}</h1>")
        if components_ok is None:
            lines.append('<p><span style="background-color:Tomato;">'
                         'Website seems unreachable or dysfunctional.</span></p>')
        elif components_ok:
            lines.append("<ul>")
            for component, ok in components_ok.items():
                status_color = "Violet"
                status_text = "status could not be retrieved (website HTML probably changed, " \
                              "this aggregator may need to be updated)"
                if ok is not None:
                    if ok:
                        status_color = "MediumSeaGreen"
                        status_text = "Healthy"
                    else:
                        status_color = "Tomato"
                        status_text = "Sick"
                lines.append(
                    f'<li>{component}: <span style="background-color:{status_color};">{status_text}</span></li>'
                )
            lines.append("</ul>")
        else:
            lines.append('<p><span style="background-color:MediumSeaGreen;">'
                         'Website does not report any problem.</span></p>')
    lines.extend(["</body>", "</html>"])

    page = "\n".join(lines)

    with open("latest_result.html", "wt") as f:
        f.write(page)
    return page


class MyServer(BaseHTTPRequestHandler):
    def do_GET(self):
        page = monitor()
        self.send_response(200)
        self.send_header("Content-type", "text/html")
        self.end_headers()
        self.wfile.write(bytes(page.encode()))


if __name__ == "__main__":
    webServer = HTTPServer((HOST_NAME, SERVER_PORT), MyServer)
    print("Server started http://%s:%s" % (HOST_NAME, SERVER_PORT))

    try:
        webServer.serve_forever()
    except KeyboardInterrupt:
        pass

    webServer.server_close()
    print("Server stopped.")
	from http.server import BaseHTTPRequestHandler, HTTPServer
	from datetime import datetime
	import urllib.request
	import urllib.error
	from bs4 import BeautifulSoup


	TIMEOUT = 2.0 # seconds
	HOST_NAME = "localhost"
	SERVER_PORT = 8080


	def basic_load(url):
	try:
	response = urllib.request.urlopen(url, timeout=TIMEOUT)
	except urllib.error.HTTPError as e:
	response = e
	except urllib.error.URLError:
	response = None
	if response is None:
	return None
	if response.status != 200:
	return None
	content = response.read()
	return content


	def kpn_load():
	return basic_load("https://www.kpn.com/zakelijk/eherkenning/service.htm")


	def kpn_parse(content):
	if content is None:
	return None
	result = {}
	soup = BeautifulSoup(content, "html.parser")
	components = soup.find_all(**{"class": "notification-head"})
	for component_ in components:
	pp = component_.parent.parent
	status = pp['class'][1]
	component_ok = None
	if status == "success":
	component_ok = True
	elif status == "error":
	component_ok = False
	component_name = component_.text
	result[component_name] = component_ok
	return result


	def ddy_load():
	return basic_load("https://ddy.statuspage.io/")


	def ddy_parse(content):
	if content is None:
	return None
	result = {}
	soup = BeautifulSoup(content, "html.parser")
	components = soup.find_all(**{"class": "component-inner-container"})
	for component_ in components:
	status = component_['class'][1]
	component_ok = None
	if status == "status-green":
	component_ok = True
	elif status == "status-red": # I think?
	component_ok = False
	component_name = component_.span.text.strip()
	result[component_name] = component_ok
	return result


	def qvd_load():
	return basic_load("https://support.qv-eherkenning.nl/storing-onderhoud/")


	def qvd_parse(content):
	if content is None:
	return None
	result = {}
	soup = BeautifulSoup(content, "html.parser")
	h2s = soup.find_all("h2")
	problems = None
	for h2 in h2s:
	if h2.text == "Storing":
	problems = h2.find_next("ul")
	break
	else:
	return result
	problem = problems.li
	while problem:
	problem_component = problem.span.text
	if problem_component != "Op dit moment zijn er geen storingen bij ons bekend.":
	result[problem_component] = False
	problem = problem.find_next_sibling("li")
	return result


	def monitor():
	time_start = datetime.now()
	results = {
	"KPN": kpn_parse(kpn_load()),
	"Digidentity": ddy_parse(ddy_load()),
	"QuoVadis": qvd_parse(qvd_load()),
	}
	time_stop = datetime.now()
	query_time = time_stop - time_start

	lines = ["<!DOCTYPE html>", "<html>", "<body>",
	f"<p>Query beginning: {time_start.isoformat()}</p>",
	f"<p>Query end: {time_stop.isoformat()}</p>",
	f"<p>Query duration: {query_time}</p>",
	]

	for site, components_ok in results.items():
	lines.append(f"<h1>{site}</h1>")
	if components_ok is None:
	lines.append('<p><span style="background-color:Tomato;">'
	'Website seems unreachable or dysfunctional.</span></p>')
	elif components_ok:
	lines.append("<ul>")
	for component, ok in components_ok.items():
	status_color = "Violet"
	status_text = "status could not be retrieved (website HTML probably changed, " \
	"this aggregator may need to be updated)"
	if ok is not None:
	if ok:
	status_color = "MediumSeaGreen"
	status_text = "Healthy"
	else:
	status_color = "Tomato"
	status_text = "Sick"
	lines.append(
	f'<li>{component}: <span style="background-color:{status_color};">{status_text}</span></li>'
	)
	lines.append("</ul>")
	else:
	lines.append('<p><span style="background-color:MediumSeaGreen;">'
	'Website does not report any problem.</span></p>')
	lines.extend(["</body>", "</html>"])

	page = "\n".join(lines)

	with open("latest_result.html", "wt") as f:
	f.write(page)
	return page


	class MyServer(BaseHTTPRequestHandler):
	def do_GET(self):
	page = monitor()
	self.send_response(200)
	self.send_header("Content-type", "text/html")
	self.end_headers()
	self.wfile.write(bytes(page.encode()))


	if __name__ == "__main__":
	webServer = HTTPServer((HOST_NAME, SERVER_PORT), MyServer)
	print("Server started http://%s:%s" % (HOST_NAME, SERVER_PORT))

	try:
	webServer.serve_forever()
	except KeyboardInterrupt:
	pass

	webServer.server_close()
	print("Server stopped.")