diff --git a/slack-bot/__pycache__/pullTrafficInfo.cpython-37.pyc b/slack-bot/__pycache__/pullTrafficInfo.cpython-37.pyc new file mode 100644 index 0000000..e39ef70 Binary files /dev/null and b/slack-bot/__pycache__/pullTrafficInfo.cpython-37.pyc differ diff --git a/slack-bot/pullTrafficInfo.py b/slack-bot/pullTrafficInfo.py index 4a9a071..8c29791 100644 --- a/slack-bot/pullTrafficInfo.py +++ b/slack-bot/pullTrafficInfo.py @@ -2,13 +2,45 @@ from selenium import webdriver from selenium.common.exceptions import NoSuchElementException from datetime import datetime import re +import emoji +import random class getTrafficInfo(object): - """docstring for getTrafficInfo""" - def __init__(self, browser): + """ + Pulls the latest traffic information for a major road or + motorway from trafficdelays.co.uk. + + Parameters: + motorway: str -> a string containing the A-road or motorway + driver: str -> the browser/driver to run under (chrome default) + + Usage: + + ** required ** + Create a driver for your motorway: + getTrafficInfo.getTrafficURL(motorway, driver) + + **options** + Get information for incident types: + getTrafficInfo.findIncidents() + + Get HTML from the page: + getTrafficInfo.getIncidentHTML() + + Get a count of how many incidents for each type: + getTrafficInfo.getIncidentCount() + + Get the text detailing the incident: + getTrafficInfo.getIncidentInformation() + + Generate a sequential list for an output: + getTrafficInfo.getOutput() + """ + def __init__(self, browser, motorway): super(getTrafficInfo, self).__init__() self.browser = browser + self.motorway = motorway self.incidentTypes = ['congestion', 'accident'] @classmethod @@ -17,7 +49,7 @@ class getTrafficInfo(object): + '-traffic-delays') browser = getTrafficInfo.getWebDriver(driver, url) getTrafficInfo.verfiyMotorway(browser, motorway) - return cls(browser) + return cls(browser, motorway) @staticmethod def getWebDriver(driver, url): @@ -47,11 +79,80 @@ class getTrafficInfo(object): pass def findIncidents(self): + self.incidentBrowser = [] for item in self.incidentTypes: xpath = f'//*[@id="{item}"]' - setattr(self, f'item', self.browser.find_element_by_xpath(xpath)) - print(self.item.text) + self.incidentBrowser.append(self.browser.find_element_by_xpath + (xpath)) + self.getIncidentHTML() + return self + + def getIncidentHTML(self): + self.incidentHTML = [] + for item in self.incidentBrowser: + self.incidentHTML.append(item.get_attribute('innerHTML')) + return self + + def getIncidentCount(self): + self.incidentCount = [] + for item, i in zip(self.incidentBrowser, + range(0, len(self.incidentHTML))): + self.incidentCount.append(self.incidentHTML[i].count('
  • ')) + return self + + def getIncidentInformation(self): + self.incidentInformation = [] + pattern = r".*\.*title=\".*\".*\>((.|\n)*?)\" + for item in self.incidentHTML: + self.incidentInformation.append(re.findall(pattern, item)) + return self + + def generateOutput(self): + self.getIncidentCount() + self.output = [] + self.sarcasticMessage = [(f'Hey Andy, have you thought about getting' + ' the train?' + + emoji.emojize(f' :bullettrain_front:', + use_aliases=True)), + (f'Hey Andy, maybe flying would be quicker?' + + emoji.emojize(f' :helicopter:', + use_aliases=True)), + (f'Don\'t fret, Andy can always work from' + ' home!' + + emoji.emojize(f' :house_with_garden:', + use_aliases=True))] + currentTime = datetime.now().strftime('%H:%M') + self.output.append(emoji.emojize(f'Did someone say {self.motorway}!?' + ' :anguished:' + ' Let \'s check the latest updates' + ' from Highways England as of' + f' {currentTime}!' + ' :police_car::rotating_light:', + use_aliases=True)) + for item, i in zip(self.incidentCount, + range(0, len(self.incidentTypes))): + if item == 0: + self.output.append(emoji.emojize + (f'There are currently no' + ' reported' + f' {self.incidentTypes[i]} incidents on' + f' the {self.motorway} :thinking_face:', + use_aliases=True)) + pass + else: + self.output.append(emoji.emojize + (f'There are currently' + f' {self.incidentCount[i]} reported' + f' {self.incidentTypes[i]} incidents' + f' reported on the {self.motorway}' + f' :scream:', + use_aliases=True)) + self.output.append(self.incidentInformation[i][0][0]) + self.output.append(random.choice(self.sarcasticMessage)) return self -inst = getTrafficInfo.getTrafficURL('M62').findIncidents() +# inst = getTrafficInfo.getTrafficURL('A50').findIncidents() \ +# .getIncidentInformation().generateOutput() +# for i in inst.output: +# print(i) diff --git a/slack-bot/sendToSQS.py b/slack-bot/sendToSQS.py new file mode 100644 index 0000000..e5314eb --- /dev/null +++ b/slack-bot/sendToSQS.py @@ -0,0 +1,7 @@ +import boto3 +from pullTrafficInfo import getTrafficInfo + +session = boto3.Session(profile_name='plex-aws') +sqs = session.resource('sqs') + +queue = sqs.create_queue(QueueName='test', Attributes={'DelaySeconds': '5'})