From ee3e87a6ec19878d56e8f386b20c58e4d9b211b3 Mon Sep 17 00:00:00 2001 From: Dmitrii Morozov Date: Tue, 7 May 2024 17:15:25 +0200 Subject: Modules --- fortnite_status/__init__.py | 119 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 fortnite_status/__init__.py (limited to 'fortnite_status/__init__.py') diff --git a/fortnite_status/__init__.py b/fortnite_status/__init__.py new file mode 100644 index 0000000..244502b --- /dev/null +++ b/fortnite_status/__init__.py @@ -0,0 +1,119 @@ +# Simple utility class which provides an access to Fortnite service status report. +# Usage: ForniteStatus().getStatus() + +import requests as req +import html5lib +from bs4 import BeautifulSoup + +# Polling interval in seconds +__polling_interval__ = 5 * 60 # 5 minutes + +class FortniteStatusObserver: + async def update(self, fortnite_status) -> None: + pass + +class FortniteStatus: + """ + Instantiate a FortniteStatus class. + """ + + class Status: + serviceStatuses = [] + + def __init__(self, serviceStatuses): + self.serviceStatuses = serviceStatuses + + def __eq__(self, other): + if not isinstance(other, FortniteStatus.Status): + return NotImplemented + return sorted(self.serviceStatuses) == sorted(other.serviceStatuses) + + def prettify(self): + return 'Fortnite services status:\n' + '\n'.join([serviceStatus.prettify() for serviceStatus in self.serviceStatuses]) + + class ServiceStatus: + serviceName = '' + status = False + + def __init__(self, serviceName, status): + self.serviceName = serviceName + self.status = status + + def __lt__(self, other): + if not isinstance(other, FortniteStatus.ServiceStatus): + return NotImplemented + return self.serviceName < other.serviceName + + def __eq__(self, other): + if not isinstance(other, FortniteStatus.ServiceStatus): + return NotImplemented + return self.serviceName == other.serviceName and self.status == other.status + + + def prettify(self): + return f'{self.serviceName}, {self.status}' + + def __findFortniteStatusHtmlComponent(self, html): + for component in html.findAll('div', {'class': 'component-container'}): + innerContainers = component.findAll('div', {'class': 'component-inner-container'}) + for innerContainer in innerContainers: + for names in innerContainer.findAll('span', {'class': 'name'}): + for name in names.findAll('span'): + if 'class' not in name.attrs and 'Fortnite' in name.text: + return component + + def __parseFortniteStatus(self, html): + component = self.__findFortniteStatusHtmlComponent(html) + serviceStatuses = [] + childContainer = component.find('div', {'class': 'child-components-container'}) + for innerContainer in childContainer.findAll('div', {'class': 'component-inner-container'}): + name = innerContainer.find('span', {'class': 'name'}).text.strip() + statusString = innerContainer.find('span', {'class': 'component-status'}).text.strip() + if statusString == 'Operational': + statusCode = True + else: + statusCode = False + serviceStatuses.append(self.ServiceStatus(name, statusCode)) + return self.Status(serviceStatuses) + + def getStatus(self): + webContent = req.get("https://status.epicgames.com/") + parsedHtml = BeautifulSoup(webContent.text, 'html5lib') + + return self.__parseFortniteStatus(parsedHtml) + + + + def printStatus(self): + """ + Prints a current Fortnite services status in stdout. + Example: + + """ + print(self.getStatus().prettify()) + +class FortniteStatusNotifier: + + __fortnite_status_observer: FortniteStatusObserver + __fortnite_status: FortniteStatus + __last_fortnite_status: any + + def __init__(self, fortnite_status_observer: FortniteStatusObserver): + self.__fortnite_status_observer = fortnite_status_observer + self.__fortnite_status = FortniteStatus() + + async def run(self): + # Initialize status + self.__last_fortnite_status = self.__fortnite_status.getStatus() + while True: + await self.__read_status() + await asyncio.sleep(__polling_interval__) + + async def __read_status(self): + service_status_tmp = self.__fortnite_status.getStatus() + if service_status_tmp != self.__last_fortnite_status: + await self.__notify(service_status_tmp) + self.__last_fortnite_status = service_status_tmp + + async def __notify(self, fortnite_status): + await self.__fortnite_status_observer.update(fortnite_status) \ No newline at end of file -- cgit v1.2.3