Day 39 & 40 - Flight Deal Finder [BIG project]

learning from 100 Days of Code: The Complete Python Pro Bootcamp for 2022

Tequila Flight Search API Documentation

main.py

from data_manager import DataManager
from flight_search import FlightSearch
from datetime import datetime, timedelta
from notification_manager import NotificationManager

ORIGIN_CITY_IATA = "TPE"

datamanager = DataManager()
sheet_data = datamanager.get_data()
flight_search = FlightSearch()
notification_manager = NotificationManager()

# if needs to add new account
# data_manager.add_new_account()

tomorrow = datetime.today() + timedelta(days=1)
six_month_from_today = datetime.today() + timedelta(days=(6 * 30))


for destination in sheet_data:
    if destination['iataCode'] == '':
        flight_search.fill_iata_code(destination['id'], destination['city'])

    flight = flight_search.check_flight(
        origin_city_iata=ORIGIN_CITY_IATA,
        destination_iata=destination['iataCode'],
        date_from=tomorrow,
        date_to=six_month_from_today)

    if flight == None:
        continue

    if flight.price <= destination['lowestPrice']:
        msg = f"Low price alert! Only NTD {flight.price} to fly from {flight.departure_city}-" \
               f"{flight.departure_airport_code} to {flight.arrival_city}-" \
               f"{flight.arrival_airport_code}, from {flight.departure_date} to " \
               f"{flight.arrival_date}."

        if flight.via_city != "":
            msg += f"Flight has 1 stop over, via {flight.via_city}"

        link = f"{flight.deep_link}"

        notification_manager.send_alert(message=msg, link=link)

data_manager.py

import requests

SHEET_ENDPOINT_PRICE = YOUR_OWN_ENDPOINT
SHEET_END_POINT_USER = YOUR_OWN_ENDPOINT

class DataManager:
    def __init__(self):
        self.destination_data = []

    def get_data(self):
        response = requests.get(url=SHEET_ENDPOINT_PRICE)
        data = response.json()
        self.destination_data = data["prices"]
        return self.destination_data

    def update_sheet(self, index, column, new_data):
        edit_endpoint = f"{SHEET_ENDPOINT_PRICE}/{index}"
        edit_para = {
            "price": {
                column: new_data,
            }
        }
        edit_response = requests.put(url=edit_endpoint, json=edit_para)

        edit_response.raise_for_status()

    def get_email(self):
        '''return mail list'''
        response = requests.get(url=SHEET_END_POINT_USER)
        data = response.json()
        return [data['users'][index]['email'] for index in range(0, len(data['users']))]

    def save_info(self, f_name, l_name, email):
        save_para = {
            "user": {
                "firstName": f_name,
                "lastName": l_name,
                "email": email
            }
        }

        response = requests.post(url=SHEET_END_POINT_USER, json=save_para)
        response.raise_for_status()
        # print(response.text)
        print("Success! Your email has been added, look forwards!")

    def add_new_account(self):
        print("Welcome to Penny's Flight Club")
        print("We find the best flight deals and email you\n")
        f_name = input("What is your first name?\n")
        l_name = input("What is your last name\n")
        email = input("What is your email?\n")
        email_check = input("Type your email again.\n")

        if email == email_check:
            self.save_info(f_name, l_name, email)

        else:
            print("Failed! Please enter same email twice, retry again? (y/n)")

        again = input("This repl has exited, run again? (y/n)")

        if again == 'y':
            self.add_new_account()

flight_search.py

import requests
from flight_data import FlightData
from data_manager import DataManager
from datetime import datetime

IATA_SEARCH_ENDPOINT = "https://tequila-api.kiwi.com/"
IATA_API_KEY = YOUR_OWN_API_KEY

data_manager = DataManager()

header = {"apikey": IATA_API_KEY}


class FlightSearch:

    def fill_iata_code(self, id, city):

        code_search_para = {
            "term": city
        }

        iata_response = requests.get(url=f"{IATA_SEARCH_ENDPOINT}locations/query",
                                     headers=header,
                                     params=code_search_para)

        iata_response.raise_for_status()
        iata_code = iata_response.json()['locations'][0]['code']

        data_manager.update_sheet(id, "iataCode", iata_code)

    def check_flight(self, origin_city_iata, destination_iata, date_from, date_to, stop_over=1):
        price_search_para = {
            "fly_from": origin_city_iata,
            "fly_to": destination_iata,
            "date_from": date_from.strftime("%d/%m/%Y"),
            "date_to": date_to.strftime("%d/%m/%Y"),
            "max_stopovers": 0,
            "nights_in_dst_from": 7,
            "nights_in_dst_to": 28,
            "max_stopovers": stop_over,
            "flight_type": "round",
            "curr": "TWD",
        }
        flight_response = requests.get(url=f"{IATA_SEARCH_ENDPOINT}search",
                                       headers=header,
                                       params=price_search_para)

        try:
            flight_info = flight_response.json()['data'][0]
        except IndexError:
            if stop_over == 1:
                return self.check_flight(origin_city_iata, destination_iata, date_from, date_to, stop_over=2)
            if stop_over == 2:
                print(f"No flights found for {destination_iata}")
                return None

        if len(flight_info['route']) == 4:
            via_city = flight_info['route'][0]['cityTo']
            d_date = flight_info['route'][0]['dTimeUTC']
            a_date = flight_info['route'][-1]['aTimeUTC']
        else:
            via_city = ""
            d_date = flight_info['route'][0]['dTimeUTC']
            a_date = flight_info['route'][-1]['aTimeUTC']

        flight_data = FlightData(
            price=flight_info['price'],
            city_from=flight_info['cityFrom'],
            airport_from=flight_info['flyFrom'],
            city_to=flight_info['cityTo'],
            airport_to=flight_info['flyTo'],
            d_date=datetime.fromtimestamp(d_date).strftime('%Y-%m-%d'),
            a_date=datetime.fromtimestamp(a_date).strftime('%Y-%m-%d'),
            deep_link=flight_info['deep_link'],
            via_city=via_city
        )

        return flight_data

flight_data.py

class FlightData:
    def __init__(self, price, city_from, city_to, airport_from, airport_to, d_date, a_date, deep_link, via_city):
        self.price = price
        self.departure_city = city_from
        self.departure_airport_code = airport_from
        self.departure_date = d_date
        self.arrival_city = city_to
        self.arrival_airport_code = airport_to
        self.arrival_date = a_date
        self.deep_link = deep_link
        self.via_city = via_city

notification.py

from smtplib import SMTP

MY_MAIL = YOUR_OWN_EMAIL
PASSWORD = YOUR_OWN_PASSWORD

class NotificationManager:
    def send_alert(self, message, link):
        with SMTP('smtp.gmail.com') as connection:
            connection.starttls()
            connection.login(user=MY_MAIL, password=PASSWORD)
            for email in DataManager().get_email():
                connection.sendmail(from_addr=MY_MAIL,
                                    to_addrs=email,
                                    msg=f"SUBJECT: Flight alert\n\n{message}\n{link}")