diff --git a/.travis.yml b/.travis.yml index e44b438..2b72ff5 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,4 +5,4 @@ python: - "3.7.1" script: - pytest - - pycodestyle checkin.py --show-source --ignore=E501 \ No newline at end of file + - pycodestyle *.py --show-source --ignore=E501 \ No newline at end of file diff --git a/checkin.py b/checkin.py index 314846b..7c76b4b 100755 --- a/checkin.py +++ b/checkin.py @@ -20,17 +20,17 @@ from dateutil.parser import parse from docopt import docopt from math import trunc from pytz import utc +from southwest import Reservation from threading import Thread from tzlocal import get_localzone import openflights -import southwest import sys import time CHECKIN_EARLY_SECONDS = 5 -def schedule_checkin(flight_time, number, first, last, notify=[]): +def schedule_checkin(flight_time, reservation): checkin_time = flight_time - timedelta(days=1) current_time = datetime.now(utc).astimezone(get_localzone()) # check to see if we need to sleep until 24 hours before flight @@ -42,16 +42,15 @@ def schedule_checkin(flight_time, number, first, last, notify=[]): h, m = divmod(m, 60) print("Too early to check in. Waiting {} hours, {} minutes, {} seconds".format(trunc(h), trunc(m), s)) time.sleep(delta) - data = southwest.checkin(number, first, last) + data = reservation.checkin() for flight in data['flights']: for doc in flight['passengers']: print("{} got {}{}!".format(doc['name'], doc['boardingGroup'], doc['boardingPosition'])) - if len(notify) > 0: - southwest.send_notification(data, notify) def auto_checkin(reservation_number, first_name, last_name, notify=[]): - body = southwest.lookup_existing_reservation(reservation_number, first_name, last_name) + r = Reservation(reservation_number, first_name, last_name, notify) + body = r.lookup_existing_reservation() # Get our local current time now = datetime.now(utc).astimezone(get_localzone()) @@ -70,7 +69,7 @@ def auto_checkin(reservation_number, first_name, last_name, notify=[]): # found a flight for checkin! print("Flight information found, departing {} at {}".format(airport, date.strftime('%b %d %I:%M%p'))) # Checkin with a thread - t = Thread(target=schedule_checkin, args=(date, reservation_number, first_name, last_name, notify)) + t = Thread(target=schedule_checkin, args=(date, r)) t.daemon = True t.start() threads.append(t) diff --git a/checkin_test.py b/checkin_test.py index cc6b555..d431400 100644 --- a/checkin_test.py +++ b/checkin_test.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta from pytz import timezone, utc from tzlocal import get_localzone + def template_time(file_name): in_string = open(file_name, 'r').read() t = datetime.now() + timedelta(seconds=5) @@ -15,6 +16,7 @@ def template_time(file_name): in_string = in_string.replace('TIME_HERE', t.strftime('%H:%M')) return in_string + def test_checkin(requests_mock): requests_mock.register_uri('GET', '/api/mobile-misc/v1/mobile-misc/page/view-reservation/XXXX?first-name=John&last-name=Smith', text=template_time('fixtures/view-reservation.json')) requests_mock.register_uri('GET', '/api/mobile-air-operations/v1/mobile-air-operations/page/check-in/XXXX?first-name=John&last-name=Smith', text=template_time('fixtures/checkin-get.json')) @@ -22,14 +24,16 @@ def test_checkin(requests_mock): requests_mock.register_uri('POST', '/php/apsearch.php', text=template_time('fixtures/openflights.json')) try: checkin.auto_checkin('XXXX', 'John', 'Smith') - except: + except Exception: pytest.fail("Error checking in") + def test_timezone_localization(): tz = timezone('America/Los_Angeles') date = tz.localize(datetime.strptime('2018-01-01 13:00', '%Y-%m-%d %H:%M')) assert date.strftime('%z') == '-0800' + @vcr.use_cassette() def test_openflights_api(): tzrequest = {'iata': 'LAX', @@ -42,24 +46,22 @@ def test_openflights_api(): airport_tz = timezone(json.loads(tzresult.text)['airports'][0]['tz_id']) assert airport_tz.zone == "America/Los_Angeles" + def test_notifications(requests_mock, mocker): requests_mock.register_uri('GET', '/api/mobile-air-operations/v1/mobile-air-operations/page/check-in/XXXX?first-name=John&last-name=Smith', text=template_time('fixtures/checkin-get.json')) data = template_time('fixtures/checkin-post.json') requests_mock.register_uri('POST', '/api/mobile-air-operations/v1/mobile-air-operations/page/check-in', text=data) data = json.loads(data) requests_mock.register_uri('POST', '/php/apsearch.php', text=template_time('fixtures/openflights.json')) - mocked_checkin = mocker.patch('southwest.send_notification') + mocked_checkin = mocker.patch('southwest.Reservation.send_notification') t = datetime.now(utc).astimezone(get_localzone()) + timedelta(minutes=5) try: - checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith') - southwest.send_notification.assert_not_called() - email = [{'mediaType': 'EMAIL', 'emailAddress': 'test@example.com'}] - checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith', email) - southwest.send_notification.assert_called_once_with(data['checkInConfirmationPage'], email) - southwest.send_notification.reset_mock() - phone = [{'mediaType': 'SMS', 'phoneNumber': '1234567890'}] - checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith', phone) - southwest.send_notification.assert_called_once_with(data['checkInConfirmationPage'], phone) - except: + r = southwest.Reservation('XXXX', 'John', 'Smith') + checkin.schedule_checkin(t, r) + mocked_checkin.assert_not_called() + r.notifications = [{'mediaType': 'EMAIL', 'emailAddress': 'test@example.com'}] + checkin.schedule_checkin(t, r) + mocked_checkin.assert_called_once() + except Exception: pytest.fail("Error checking in") diff --git a/openflights.py b/openflights.py index 364c449..9efb7d9 100644 --- a/openflights.py +++ b/openflights.py @@ -2,6 +2,7 @@ import requests import json import pytz + def timezone_for_airport(airport_code): tzrequest = {'iata': airport_code, 'country': 'ALL', diff --git a/southwest.py b/southwest.py index 721549c..58b5ea2 100644 --- a/southwest.py +++ b/southwest.py @@ -10,58 +10,68 @@ MAX_ATTEMPTS = 40 # Pulled from proxying the Southwest iOS App headers = {'Host': 'mobile.southwest.com', 'Content-Type': 'application/json', 'X-API-Key': API_KEY, 'X-User-Experience-Id': USER_EXPERIENCE_KEY, 'Accept': '*/*'} -# You might ask yourself, "Why the hell does this exist?" -# Basically, there sometimes appears a "hiccup" in Southwest where things -# aren't exactly available 24-hours before, so we try a few times -def safe_request(url, body=None): - try: - attempts = 0 - while True: - if body is not None: - r = requests.post(url, headers=headers, json=body) - else: - r = requests.get(url, headers=headers) - data = r.json() - if 'httpStatusCode' in data and data['httpStatusCode'] in ['NOT_FOUND', 'BAD_REQUEST', 'FORBIDDEN']: - attempts += 1 - print(data['message']) - if attempts > MAX_ATTEMPTS: - sys.exit("Unable to get data, killing self") - time.sleep(CHECKIN_INTERVAL_SECONDS) - continue - return data - except ValueError: - # Ignore responses with no json data in body - pass -def lookup_existing_reservation(number, first, last): - # Find our existing record - url = "{}mobile-misc/v1/mobile-misc/page/view-reservation/{}?first-name={}&last-name={}".format(BASE_URL, number, first, last) - data = safe_request(url) - return data['viewReservationViewPage'] +class Reservation(): -def get_checkin_data(number, first, last): - url = "{}mobile-air-operations/v1/mobile-air-operations/page/check-in/{}?first-name={}&last-name={}".format(BASE_URL, number, first, last) - data = safe_request(url) - return data['checkInViewReservationPage'] + def __init__(self, number, first, last, notifications=[]): + self.number = number + self.first = first + self.last = last + self.notifications = [] -def checkin(number, first, last): - data = get_checkin_data(number, first, last) - info_needed = data['_links']['checkIn'] - url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) - print("Attempting check-in...") - return safe_request(url, info_needed['body'])['checkInConfirmationPage'] + # You might ask yourself, "Why the hell does this exist?" + # Basically, there sometimes appears a "hiccup" in Southwest where things + # aren't exactly available 24-hours before, so we try a few times + def safe_request(self, url, body=None): + try: + attempts = 0 + while True: + if body is not None: + r = requests.post(url, headers=headers, json=body) + else: + r = requests.get(url, headers=headers) + data = r.json() + if 'httpStatusCode' in data and data['httpStatusCode'] in ['NOT_FOUND', 'BAD_REQUEST', 'FORBIDDEN']: + attempts += 1 + print(data['message']) + if attempts > MAX_ATTEMPTS: + sys.exit("Unable to get data, killing self") + time.sleep(CHECKIN_INTERVAL_SECONDS) + continue + return data + except ValueError: + # Ignore responses with no json data in body + pass -def send_notification(checkindata, notify): - if len(notify) < 1: - return - info_needed = checkindata['_links']['boardingPasses'] - url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) - mbpdata = safe_request(url, info_needed['body']) - info_needed = mbpdata['checkInViewBoardingPassPage']['_links'] - url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) - print("Attempting to send boarding pass...") - body = info_needed['body'] - for n in notify: - body.update(n) - safe_request(url, body) + def lookup_existing_reservation(self): + # Find our existing record + url = "{}mobile-misc/v1/mobile-misc/page/view-reservation/{}?first-name={}&last-name={}".format(BASE_URL, self.number, self.first, self.last) + data = self.safe_request(url) + return data['viewReservationViewPage'] + + def get_checkin_data(self): + url = "{}mobile-air-operations/v1/mobile-air-operations/page/check-in/{}?first-name={}&last-name={}".format(BASE_URL, self.number, self.first, self.last) + data = self.safe_request(url) + return data['checkInViewReservationPage'] + + def checkin(self): + data = self.get_checkin_data() + info_needed = data['_links']['checkIn'] + url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) + print("Attempting check-in...") + confirmation = self.safe_request(url, info_needed['body'])['checkInConfirmationPage'] + if len(self.notifications) > 0: + self.send_notification(confirmation) + return confirmation + + def send_notification(self, checkindata): + info_needed = checkindata['_links']['boardingPasses'] + url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) + mbpdata = self.safe_request(url, info_needed['body']) + info_needed = mbpdata['checkInViewBoardingPassPage']['_links'] + url = "{}mobile-air-operations{}".format(BASE_URL, info_needed['href']) + print("Attempting to send boarding pass...") + body = info_needed['body'] + for n in self.notifications: + body.update(n) + self.safe_request(url, body)