jgaunt/SouthwestCheckin
1
0
Fork 0
mirror of https://github.com/pyro2927/SouthwestCheckin.git synced 2025-12-06 01:13:19 +00:00
Code Issues Releases Wiki Activity

Refactoring Southwest class to be OOP

Browse Source
This commit is contained in:
pyro2927
2019-02-16 15:25:43 -08:00
parent 170519d316
commit d7722fa1ce
5 changed files with 83 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.travis.yml
View File

@@ -5,4 +5,4 @@ python:
- "3.7.1"
script:
- pytest
- pycodestyle checkin.py --show-source --ignore=E501
- pycodestyle *.py --show-source --ignore=E501

13
checkin.py
View File

@@ -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)

26
checkin_test.py
View File

@@ -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")

1
openflights.py
View File

@@ -2,6 +2,7 @@ import requests
import json
import pytz
def timezone_for_airport(airport_code):
tzrequest = {'iata': airport_code,
'country': 'ALL',

112
southwest.py
View File

@@ -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)