pullTrafficInfo completed

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('<li>'))
+        return self
+
+    def getIncidentInformation(self):
+        self.incidentInformation = []
+        pattern = r".*\<li\>.*title=\".*\".*\>((.|\n)*?)\<br\>"
+        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)

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'})