From 7a43ff00c81a5b3f67258f1db6823077116769a7 Mon Sep 17 00:00:00 2001 From: pyro2927 Date: Sat, 16 Feb 2019 14:17:43 -0800 Subject: [PATCH] Refactored, added in pycodestyle for linting --- .travis.yml | 3 +- checkin.py | 97 +++++++----------------------------------------- checkin_test.py | 11 +++--- openflights.py | 14 +++++++ requirements.txt | 1 + southwest.py | 68 +++++++++++++++++++++++++++++++++ 6 files changed, 104 insertions(+), 90 deletions(-) create mode 100644 openflights.py create mode 100644 southwest.py diff --git a/.travis.yml b/.travis.yml index 5c628f3..e44b438 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,4 +4,5 @@ python: - "2.7.15" - "3.7.1" script: - - pytest \ No newline at end of file + - pytest + - pycodestyle checkin.py --show-source --ignore=E501 \ No newline at end of file diff --git a/checkin.py b/checkin.py index 4337572..eb60b66 100755 --- a/checkin.py +++ b/checkin.py @@ -19,84 +19,20 @@ from datetime import timedelta from dateutil.parser import parse from docopt import docopt from math import trunc +from pytz import utc from threading import Thread from tzlocal import get_localzone -import json -import pytz -import requests +import openflights +import southwest import sys import time -API_KEY = 'l7xxb3dcccc4a5674bada48fc6fcf0946bc8' -USER_EXPERIENCE_KEY = 'AAAA3198-4545-46F4-9A05-BB3E868BEFF5' -BASE_URL = 'https://mobile.southwest.com/api/' CHECKIN_EARLY_SECONDS = 5 -CHECKIN_INTERVAL_SECONDS = 0.25 -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'] - -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 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'] - -def send_notification(checkindata, emailaddr=None, mobilenum=None): - 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']) - if emailaddr: - info_needed['body']['mediaType'] = 'EMAIL' - info_needed['body']['emailAddress'] = emailaddr - if mobilenum: - info_needed['body']['mediaType'] = 'SMS' - info_needed['body']['phoneNumber'] = mobilenum - print("Attempting to send boarding pass...") - safe_request(url, info_needed['body']) def schedule_checkin(flight_time, number, first, last, email, mobile): checkin_time = flight_time - timedelta(days=1) - current_time = datetime.now(pytz.utc).astimezone(get_localzone()) + current_time = datetime.now(utc).astimezone(get_localzone()) # check to see if we need to sleep until 24 hours before flight if checkin_time > current_time: # calculate duration to sleep @@ -106,21 +42,21 @@ def schedule_checkin(flight_time, number, first, last, email, mobile): 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 = checkin(number, first, last) + data = southwest.checkin(number, first, last) for flight in data['flights']: for doc in flight['passengers']: print("{} got {}{}!".format(doc['name'], doc['boardingGroup'], doc['boardingPosition'])) if email: - send_notification(data, emailaddr=email) + southwest.send_notification(data, emailaddr=email) elif mobile: - send_notification(data, mobilenum=mobile) + southwest.send_notification(data, mobilenum=mobile) def auto_checkin(reservation_number, first_name, last_name, email=None, mobile=None): - body = lookup_existing_reservation(reservation_number, first_name, last_name) + body = southwest.lookup_existing_reservation(reservation_number, first_name, last_name) # Get our local current time - now = datetime.now(pytz.utc).astimezone(get_localzone()) + now = datetime.now(utc).astimezone(get_localzone()) tomorrow = now + timedelta(days=1) threads = [] @@ -130,14 +66,7 @@ def auto_checkin(reservation_number, first_name, last_name, email=None, mobile=N # calculate departure for this leg airport = "{}, {}".format(leg['departureAirport']['name'], leg['departureAirport']['state']) takeoff = "{} {}".format(leg['departureDate'], leg['departureTime']) - tzrequest = {'iata': leg['departureAirport']['code'], - 'country': 'ALL', - 'db': 'airports', - 'iatafilter': 'true', - 'action': 'SEARCH', - 'offset': '0'} - tzresult = requests.post("https://openflights.org/php/apsearch.php", tzrequest) - airport_tz = pytz.timezone(json.loads(tzresult.text)['airports'][0]['tz_id']) + airport_tz = openflights.timezone_for_airport(leg['departureAirport']['code']) date = airport_tz.localize(datetime.strptime(takeoff, '%Y-%m-%d %H:%M')) if date > now: # found a flight for checkin! @@ -158,10 +87,10 @@ def auto_checkin(reservation_number, first_name, last_name, email=None, mobile=N threads.remove(t) break -if __name__ == '__main__': - arguments = docopt(__doc__, version='Southwest Checkin 0.2') - # work work +if __name__ == '__main__': + + arguments = docopt(__doc__, version='Southwest Checkin 0.2') reservation_number = arguments['CONFIRMATION_NUMBER'] first_name = arguments['FIRST_NAME'] last_name = arguments['LAST_NAME'] diff --git a/checkin_test.py b/checkin_test.py index 20e6841..8e07c7c 100644 --- a/checkin_test.py +++ b/checkin_test.py @@ -2,6 +2,7 @@ import checkin import json import pytest import requests +import southwest from datetime import datetime, timedelta from pytz import timezone, utc from tzlocal import get_localzone @@ -45,16 +46,16 @@ def test_notifications(requests_mock, mocker): 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('checkin.send_notification') + mocked_checkin = mocker.patch('southwest.send_notification') t = datetime.now(utc).astimezone(get_localzone()) + timedelta(minutes=5) try: checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith', None, None) - checkin.send_notification.assert_not_called() + southwest.send_notification.assert_not_called() checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith', 'test@example.com', None) - checkin.send_notification.assert_called_once_with(data['checkInConfirmationPage'], emailaddr='test@example.com') - checkin.send_notification.reset_mock() + southwest.send_notification.assert_called_once_with(data['checkInConfirmationPage'], emailaddr='test@example.com') + southwest.send_notification.reset_mock() checkin.schedule_checkin(t, 'XXXX', 'John', 'Smith', None, '1234567890') - checkin.send_notification.assert_called_once_with(data['checkInConfirmationPage'], mobilenum='1234567890') + southwest.send_notification.assert_called_once_with(data['checkInConfirmationPage'], mobilenum='1234567890') except: pytest.fail("Error checking in") diff --git a/openflights.py b/openflights.py new file mode 100644 index 0000000..364c449 --- /dev/null +++ b/openflights.py @@ -0,0 +1,14 @@ +import requests +import json +import pytz + +def timezone_for_airport(airport_code): + tzrequest = {'iata': airport_code, + 'country': 'ALL', + 'db': 'airports', + 'iatafilter': 'true', + 'action': 'SEARCH', + 'offset': '0'} + tzresult = requests.post("https://openflights.org/php/apsearch.php", tzrequest) + airport_tz = pytz.timezone(json.loads(tzresult.text)['airports'][0]['tz_id']) + return airport_tz diff --git a/requirements.txt b/requirements.txt index 7c4d9ea..a24d149 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ datetime docopts +pycodestyle pytest pytest-mock python-dateutil diff --git a/southwest.py b/southwest.py new file mode 100644 index 0000000..2246a0b --- /dev/null +++ b/southwest.py @@ -0,0 +1,68 @@ +import requests +import sys + +API_KEY = 'l7xxb3dcccc4a5674bada48fc6fcf0946bc8' +USER_EXPERIENCE_KEY = 'AAAA3198-4545-46F4-9A05-BB3E868BEFF5' +BASE_URL = 'https://mobile.southwest.com/api/' +CHECKIN_INTERVAL_SECONDS = 0.25 +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'] + +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 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'] + +def send_notification(checkindata, emailaddr=None, mobilenum=None): + 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']) + if emailaddr: + info_needed['body']['mediaType'] = 'EMAIL' + info_needed['body']['emailAddress'] = emailaddr + if mobilenum: + info_needed['body']['mediaType'] = 'SMS' + info_needed['body']['phoneNumber'] = mobilenum + print("Attempting to send boarding pass...") + safe_request(url, info_needed['body'])