ShoGinn/SouthwestCheckin

View on GitHub
src/swcheckin/southwest.py

Summary

Maintainability
A
1 hr
Test Coverage
"""Southwest Checkin."""
import json
import sys
import uuid

from time import sleep
from typing import Optional
from urllib.parse import urlencode, urljoin

import requests

from .config import config


class Reservation():
    """The Reservation Class."""

    def __init__(self, number: str, first: str, last: str, verbose: bool = False):
        """Require confirmation number, first and last name."""
        self.number = number
        self.first = first
        self.last = last
        self.verbose = verbose

    @staticmethod
    def generate_headers():
        """Generate the headers necessary for request.

        Returns:
            dict -- The header
        """
        config_js = requests.get('https://mobile.southwest.com/js/config.js')
        if config_js.status_code == requests.codes.ok:  # pylint: disable=no-member
            modded = config_js.text[config_js.text.index('API_KEY'):]
            api_key = modded[
                modded.index(':') +
                1:modded.index(',')
            ].strip('"')
        else:
            print("Couldn't get API_KEY")
            sys.exit(1)

        user_experience_key = str(uuid.uuid1()).upper()
        # Pulled from proxying the Southwest iOS App
        return {
            'Host': 'mobile.southwest.com',
            'Content-Type': 'application/json',
            'X-API-Key': api_key,
            'X-User-Experience-Id': user_experience_key,
            'Accept': '*/*',
            'X-Channel-ID': 'MWEB',
        }

    # 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: str, body: Optional[str] = None):
        """Loops the request a set number of times [Default: 40].

        Arguments:
            url {str} -- southwests url for mobile

        Keyword Arguments:
            body {Optional[str]} -- extra info (default: {None})

        Returns:
            {dict} -- The json from southwest
        """
        try:
            headers = Reservation.generate_headers()
            for _ in range(0, config.max_attempts):
                if body is not None:
                    response = requests.post(url, headers=headers, json=body)
                else:
                    response = requests.get(url, headers=headers)
                data = response.json()
                if 'httpStatusCode' in data and data['httpStatusCode'] in [
                        'NOT_FOUND',
                        'BAD_REQUEST',
                        'FORBIDDEN',
                ]:
                    if not self.verbose:
                        print(data['message'])
                    else:
                        print(response.headers)
                        print(json.dumps(data, indent=2))
                    sleep(config.checkin_interval_seconds)
                    continue
                if self.verbose:
                    print(response.headers)
                    print(json.dumps(data, indent=2))
                return data
            sys.exit('Unable to get data, killing self')
        except ValueError:  # pragma: no cover
            # Ignore responses with no json data in body
            pass

    def load_json_page(self, url: str, body: Optional[str] = None):
        """Load the page with the info we need.

        Arguments:
            url {str} -- southwests url for mobile

        Keyword Arguments:
            body {Optional[str]} -- extra info (default: {None})

        Returns:
            [type] -- [description]
        """
        # pylint: disable=invalid-name
        data = self.safe_request(url, body)
        for k, v in list(data.items()):
            if k.endswith('Page'):
                return v
        return None  # pragma: no cover

    def with_suffix(self, uri):
        """Add the sw suffix to the url.

        Arguments:
            uri {str} -- The URL

        Returns:
            str -- The Url with the suffix added.
        """
        combined_url = urljoin(config.base_url, uri) + self.number + '?'
        suffix_dict = {'first-name': self.first, 'last-name': self.last}
        return combined_url + urlencode(suffix_dict)

    def lookup_existing_reservation(self):
        """Find our existing record."""
        return self.load_json_page(
            self.with_suffix(
                'mobile-air-booking/v1/mobile-air-booking/page/view-reservation/',
            ), )

    def get_checkin_data(self):
        """Get the Checkin Data."""
        return self.load_json_page(
            self.with_suffix(
                'mobile-air-operations/v1/mobile-air-operations/page/check-in/',
            ), )

    def checkin(self):
        """Check in function.

        Returns:
            [str] -- returns the confirmation
        """
        data = self.get_checkin_data()
        info_needed = data['_links']['checkIn']
        base_url = urljoin(config.base_url, 'mobile-air-operations')
        url = base_url + info_needed['href']
        print('Attempting check-in...')
        confirmation = self.load_json_page(url, info_needed['body'])
        return confirmation