캡스톤 프로젝트 1(항공권 특가 검색기)

JOOYEUN SEO·2024년 9월 22일

python udemy

100 Days of Python

목록 보기

39/76

🗂️ Day39 프로젝트 : 항공권 특가 검색기

가고 싶은 장소들과 상한가를 정해두면, 더 저렴한 항공권이 나왔을 때 알리는 프로그램

1. 구글 시트 준비

2. Sheety API로 구글시트의 데이터 읽고 쓰기

🔍 유의 사항

Sheety API의 무료 멤버십 이용 시 불필요한 요청 줄이기

pprint.pprint( object, ... )

구조가 복잡한 데이터를 pprint 모듈로 보기 좋게 출력 가능

pprint = pretty print

pprint()로 출력한 결과

사용자의 환경변수를 📄.env에 저장

📄.env

SHEETY_ENDPOINT="https://api.sheety.co/<유저이름>/citiesToFly/prices"
BEARER_TOKEN="개인 토큰"

⌨️ data_manager.py

import os
import requests
from dotenv import load_dotenv

# 환경변수 불러오기
load_dotenv()

class DataManager:

    def __init__(self):
        self.sheety_endpoint = os.getenv("SHEETY_ENDPOINT")
        self.bearer_token = os.getenv("BEARER_TOKEN")
        self.bearer_headers = {"Authorization": f"Bearer {self.bearer_token}"}
		self.destination_data = {}
        
    def get_destination_data(self):
        response = requests.get(url=self.sheety_endpoint,
                                headers=self.bearer_headers)
        data = response.json()
        self.destination_data = data["prices"]
        return self.destination_data

    def update_destination_code(self):
        for city in self.destination_data:
            new_data = {
                "price": {"iataCode": city["iataCode"]}
            }
            response = requests.put(url=f"{self.sheety_endpoint}/{city['id']}",
                                    json=new_data,
                                    headers=self.bearer_headers)

⌨️ flight_search.py

class FlightSearch:

    def get_destination_code(self, city_name):
        # 우선 시드에 코드가 제대로 입력되는지 보기 위해 테스트 문구 넣기
        code = "TESTING"
        return code

⌨️ main.py

from data_manager import DataManager
from flight_search import FlightSearch

data_manager = DataManager()
sheet_data = data_manager.get_destination_data()

# 구글 시트의 공항 코드가 비어 있는지 확인
if sheet_data[0]["iataCode"] == "":
    from flight_search import FlightSearch
    flight_search = FlightSearch()
    for row in sheet_data:
        row["iataCode"] = flight_search.get_destination_code(row["city"])

    data_manager.destination_data = sheet_data
    data_manager.update_destination_code()

출력되는 것을 체크한 뒤에는 TESTING을 모두 삭제하기

3. Amadeus로 IATA 코드 가져오기

🔍 유의 사항
Amadeus Flight Search API (무료 가입, 신용카드 정보 불필요)
My Self-Service Workspace → My apps → 새 앱 생성 후 API 받기
Amadeus API 키&토큰 사용법

API id와 secret으로 토큰을 요청해야 한다

응답 결과
{
    'type': 'amadeusOAuth2Token',
    'username': '개인 이메일 주소',
    'application_name': 'Flight Deals',
    'client_id': '개인 API key',
    'token_type': 'Bearer',
    'access_token': '발급받은 토큰',
    'expires_in': 1799,
    'state': 'approved',
    'scope': ''
 }
키 expires_in 는 1799초(약 30분) 후 토큰이 만료된다는 뜻
도시별 IATA(국제항공운송협회) 코드를 구글 시트에 붙이기

Amadeus Search for Airport Codes by City name

여러 개의 공항이 있는 도시는 공항 코드와는 별도로 도시 IATA 코드 존재
(Cities with multiple commercial airports)

공항 코드가 출력되지 않는 상황을 예외처리하기

하루에 제한된 요청 수를 넘었을 때

아마데우스 테스트 API에 없는 공항 코드를 요청했을 때
(인기 있는 도시들 위주로 넣는 것이 좋음)

무료 API로 너무 빠르게 여러 번 요청할 경우 차단되거나 오류가 발생할 수 있음

time.sleep()으로 각 요청 사이에 간격 두기

⌨️ flight_search.py

import os
import requests
from dotenv import load_dotenv

load_dotenv()

IATA_ENDPOINT = "https://test.api.amadeus.com/v1/reference-data/locations/cities"
TOKEN_ENDPOINT = "https://test.api.amadeus.com/v1/security/oauth2/token"

class FlightSearch:

    def __init__(self):
        self.api_key = os.getenv("API_KEY")
        self.api_secret = os.getenv("API_SECRET")
        # 프로그램을 시작할 때마다 토큰을 새로 발급
        self.token = self.get_new_token()

    def get_new_token(self):
        header = {'Content-Type': 'application/x-www-form-urlencoded'}
        body = {
            'grant_type': 'client_credentials',
            'client_id': self.api_key,
            'client_secret': self.api_secret
        }
        response = requests.post(url=TOKEN_ENDPOINT, headers=header, data=body)
        print(f"Your token is {response.json()['access_token']}")
        return response.json()["access_token"]

    def get_destination_code(self, city_name):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            "keyword": city_name,
            "max": "2",
            "include": "AIRPORTS"
        }
        response = requests.get(url=IATA_ENDPOINT, headers=headers, params=query)
        try:
            code = response.json()["data"][0]['iataCode']
        except IndexError:
            print(f"IndexError: No airport code found for {city_name}.")
            return "N/A"
        except KeyError:
            print(f"KeyError: No airport code found for {city_name}.")
            return "Not Found"

        return code

⌨️ main.py

import time
from data_manager import DataManager
from flight_search import FlightSearch

# ==================== Set up the Flight Search ====================
data_manager = DataManager()
sheet_data = data_manager.get_destination_data()
flight_search = FlightSearch()

# ==================== Update the Airport Codes in Google Sheet ====================
for row in sheet_data:
    if row["iataCode"] == "":
        row["iataCode"] = flight_search.get_destination_code(row["city"])
        # 각 요청 간 2초 정도 텀 주기
        time.sleep(2)

data_manager.destination_data = sheet_data
data_manager.update_destination_codes()

4. 저렴한 항공권 검색

🔍 유의 사항

Amadeus Flight Offer

❗️강의에서는 내일부터 6개월 이내에 출발하는 저렴한 항공권을 검색 가능했으나,
현재는 설정한 출발일과 리턴일(설정할 경우 왕복, 미설정할 경우 편도)의 티켓만 검색하므로
내일 출발하는 편도 항공권 중 가장 저렴한 것을 찾는 방식으로 변경

조건

출발 도시 : 런던(LON)

도착 도시 : 구글 시트에 있는 도시들

출발일 : 내일

datetime.timedelta() 사용 (Day32 ❖ datetime 모듈)

split() 함수로 json 데이터에서 날짜의 첫 부분만 가져오기(T 이전)

성인 1명의 편도 항공권

통화 : GBP

🐞 디버깅 → 특정 목적지로 향하는 항공권이 없을 때

출발 도시를 큰 공항이 있는 곳으로 변경

도착 도시를 큰 공항이 있는 곳으로 변경

통화를 USD로 변경

시트의 첫 번째 도시에 대해 check_flight를 한 결과
→ 이 데이터를 파싱해서 가장 저렴한 항공편을 추출해야 한다

⌨️ flight_data.py

class FlightData:

    def __init__(self, price, origin_airport, destination_airport, out_date):
        self.price = price
        self.origin_airport = origin_airport
        self.destination_airport = destination_airport
        self.out_date = out_date

    def find_cheapest_flight(data):
        # 오류 처리
        if data is None or not data["data"]:
            print("No flight data")
            return FlightData("N/A", "N/A", "N/A", "N/A")

        # 먼저 data["data"]의 1번째 원소를 가장 싼 항공편으로 지정
        first_flight = data["data"][0]
        lowest_price = float(first_flight["price"]["grandTotal"])
        origin = first_flight["itineraries"][0]["segments"][0]["departure"]["iataCode"]
        destination = first_flight["itineraries"][0]["segments"][0]["arrival"]["iataCode"]
        out_date = first_flight["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0]
        cheapest_flight = FlightData(lowest_price, origin, destination, out_date)

        for flight in data["data"]:
            price = float(flight["price"]["grandTotal"])
            if price < lowest_price:
                lowest_price = price
                origin = flight["itineraries"][0]["segments"][0]["departure"]["iataCode"]
                destination = flight["itineraries"][0]["segments"][0]["arrival"]["iataCode"]
                out_date = flight["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0]
                cheapest_flight = FlightData(lowest_price, origin, destination, out_date)

        return cheapest_flight

⌨️ flight_search.py

import os
import requests
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

IATA_ENDPOINT = "https://test.api.amadeus.com/v1/reference-data/locations/cities"
FLIGHT_ENDPOINT = "https://test.api.amadeus.com/v2/shopping/flight-offers"
TOKEN_ENDPOINT = "https://test.api.amadeus.com/v1/security/oauth2/token"

class FlightSearch:

    def __init__(self):
        …
        
    def get_new_token(self):
        …

    def get_destination_code(self, city_name):
        …

    def check_flight(self, origin_city_code, destination_city_code, from_time):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            "originLocationCode": origin_city_code,
            "destinationLocationCode": destination_city_code,
            "departureDate": from_time.strftime("%Y-%m-%d"),
            "adults": "1",
            "nonStop": "true",
            "currencyCode": "GBP",
            "max": "5"
        }
        response = requests.get(url=FLIGHT_ENDPOINT, headers=headers, params=query)

        if response.status_code != 200:
            print(f"check_flights() response code: {response.status_code}")
            print("Response body:", response.text)
            return None

        return response.json()

⌨️ main.py

import time
from datetime import datetime, timedelta
from data_manager import DataManager
from flight_search import FlightSearch
from flight_data import FlightData


# ==================== Set up the Flight Search ====================
…

ORIGIN_CITY_IATA = "LON"

# ==================== Update the Airport Codes in Google Sheet ====================
…

# ==================== Search for Flights ====================

tomorrow = datetime.now() + timedelta(days=1)

for destination in sheet_data:
    print(f"Getting flights for {destination['city']}...")
    flights = flight_search.check_flight(
        origin_city_code=ORIGIN_CITY_IATA,
        destination_city_code=destination["iataCode"],
        from_time=tomorrow
    )
    cheapest_flight = FlightData.find_cheapest_flight(flights)
    print(f"Lowest price to {destination['city']} is £{cheapest_flight.price}")
    time.sleep(2)

5. 항공권 가격이 구글시트에 있는 가격보다 저렴하면 WhatsApp 알림 전송하기

🔍 유의 사항

Programmable Messaging for WhatsApp and Python Quickstart

체험 계정의 요금 한도를 넘지 않기 위해 시트에 목적지 공항은 최대 10개 이내로 제한

검색한 최저가가 구글 시트의 최저가보다 저렴하다면, WhatsApp 메시지 전송하기

메시지에 포함될 정보
Low price alert!
Only "항공권 가격" to fly
from "출발 공항(도시) IATA 코드" to "도착 공항(도시) IATA 코드",
on "출발 날짜" until "리턴 날짜"
출력을 위해 시트에서 파리의 최저가를 변경

📄.env

SHEETY_ENDPOINT="https://api.sheety.co/<유저이름>/citiesToFly/prices"
BEARER_TOKEN="개인 토큰"
AMADEUS_API_KEY="개인 키"
AMADEUS_API_SECRET="개인 토큰"
TWILIO_SID="개인 시드"
TWILIO_AUTH_TOKEN="개인 토큰"
TWILIO_WHATSAPP_NUMBER="개인 가상 왓츠앱 전화번호"
TWILIO_VERIFIED_NUMBER="메시지를 받을 전화번호"

⌨️ data_manager.py

import os
import requests
from dotenv import load_dotenv

load_dotenv()

class DataManager:

    def __init__(self):
        self.sheety_endpoint = os.getenv("SHEETY_ENDPOINT")
        self.bearer_token = os.getenv("BEARER_TOKEN")
        self.bearer_headers = {"Authorization": f"Bearer {self.bearer_token}"}
        self.destination_data = {}

    def get_destination_data(self):
        response = requests.get(url=self.sheety_endpoint,
                                headers=self.bearer_headers)
        data = response.json()
        self.destination_data = data["prices"]
        return self.destination_data

    def update_destination_codes(self):
        for city in self.destination_data:
            new_data = {
                "price": {"iataCode": city["iataCode"]}
            }
            response = requests.put(url=f"{self.sheety_endpoint}/{city['id']}",
                                    json=new_data,
                                    headers=self.bearer_headers)

⌨️ flight_data.py

class FlightData:

    def __init__(self, price, origin_airport, destination_airport, out_date):
        self.price = price
        self.origin_airport = origin_airport
        self.destination_airport = destination_airport
        self.out_date = out_date

    def find_cheapest_flight(data):
        # 오류 처리
        if data is None or not data["data"]:
            print("No flight data")
            return FlightData("N/A", "N/A", "N/A", "N/A")

        # 먼저 data["data"]의 1번째 원소를 가장 싼 항공편으로 지정
        first_flight = data["data"][0]
        lowest_price = float(first_flight["price"]["grandTotal"])
        origin = first_flight["itineraries"][0]["segments"][0]["departure"]["iataCode"]
        destination = first_flight["itineraries"][0]["segments"][0]["arrival"]["iataCode"]
        out_date = first_flight["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0]
        cheapest_flight = FlightData(lowest_price, origin, destination, out_date)

        for flight in data["data"]:
            price = float(flight["price"]["grandTotal"])
            if price < lowest_price:
                lowest_price = price
                origin = flight["itineraries"][0]["segments"][0]["departure"]["iataCode"]
                destination = flight["itineraries"][0]["segments"][0]["arrival"]["iataCode"]
                out_date = flight["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0]
                cheapest_flight = FlightData(lowest_price, origin, destination, out_date)

        return cheapest_flight

⌨️ flight_search.py

import os
import requests
from dotenv import load_dotenv
from datetime import datetime

load_dotenv()

IATA_ENDPOINT = "https://test.api.amadeus.com/v1/reference-data/locations/cities"
FLIGHT_ENDPOINT = "https://test.api.amadeus.com/v2/shopping/flight-offers"
TOKEN_ENDPOINT = "https://test.api.amadeus.com/v1/security/oauth2/token"

class FlightSearch:

    def __init__(self):
        self.api_key = os.getenv("AMADEUS_API_KEY")
        self.api_secret = os.getenv("AMADEUS_API_SECRET")
        # 프로그램을 시작할 때마다 토큰을 새로 발급
        self.token = self.get_new_token()

    def get_new_token(self):
        header = {'Content-Type': 'application/x-www-form-urlencoded'}
        body = {
            'grant_type': 'client_credentials',
            'client_id': self.api_key,
            'client_secret': self.api_secret
        }
        response = requests.post(url=TOKEN_ENDPOINT, headers=header, data=body)
        print(f"Your token is {response.json()['access_token']}")
        return response.json()["access_token"]

    def get_destination_code(self, city_name):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            "keyword": city_name,
            "max": "2",
            "include": "AIRPORTS"
        }
        response = requests.get(url=IATA_ENDPOINT, headers=headers, params=query)
        try:
            code = response.json()["data"][0]['iataCode']
        except IndexError:
            print(f"IndexError: No airport code found for {city_name}.")
            return "N/A"
        except KeyError:
            print(f"KeyError: No airport code found for {city_name}.")
            return "Not Found"

        return code

    def check_flight(self, origin_city_code, destination_city_code, from_time):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            "originLocationCode": origin_city_code,
            "destinationLocationCode": destination_city_code,
            "departureDate": from_time.strftime("%Y-%m-%d"),
            "adults": "1",
            "nonStop": "true",
            "currencyCode": "GBP",
            "max": "5"
        }
        response = requests.get(url=FLIGHT_ENDPOINT, headers=headers, params=query)

        if response.status_code != 200:
            print(f"check_flights() response code: {response.status_code}")
            print("Response body:", response.text)
            return None

        return response.json()

⌨️ notification_manager.py

import os
from twilio.rest import Client
from dotenv import load_dotenv

load_dotenv()

class NotificationManager:

    def __init__(self):
        self.client = client = Client(os.getenv("TWILIO_SID"), os.getenv("TWILIO_AUTH_TOKEN"))

    def send_whatsapp(self, message_body):
        message = self.client.messages.create(
            body=message_body,
            from_=f"whatsapp:{os.getenv('TWILIO_WHATSAPP_NUMBER')}",
            to=f"whatsapp:{os.getenv('TWILIO_VERIFIED_NUMBER')}"
        )

⌨️ main.py

import time
from datetime import datetime, timedelta
from data_manager import DataManager
from flight_search import FlightSearch
from flight_data import FlightData
from notification_manager import NotificationManager

# ==================== Set up the Flight Search ====================
data_manager = DataManager()
sheet_data = data_manager.get_destination_data()
flight_search = FlightSearch()
notification_manager = NotificationManager()

ORIGIN_CITY_IATA = "LON"

# ==================== Update the Airport Codes in Google Sheet ====================
for row in sheet_data:
    if row["iataCode"] == "":
        row["iataCode"] = flight_search.get_destination_code(row["city"])
        time.sleep(2)

data_manager.destination_data = sheet_data
data_manager.update_destination_codes()

# ==================== Search for Flights ====================

tomorrow = datetime.now() + timedelta(days=1)

for destination in sheet_data:
    print(f"Getting flights for {destination['city']}...")
    flights = flight_search.check_flight(
        origin_city_code=ORIGIN_CITY_IATA,
        destination_city_code=destination["iataCode"],
        from_time=tomorrow
    )
    cheapest_flight = FlightData.find_cheapest_flight(flights)
    print(f"Lowest price to {destination['city']} is £{cheapest_flight.price}")
    time.sleep(2)

    if cheapest_flight.price != "N/A" and cheapest_flight.price < destination["lowestPrice"]:
        print(f"Lower price flight found to {destination['city']}!")
        notification_manager.send_whatsapp(
            message_body=f"Low price alert! Only £{cheapest_flight.price} to fly "
                         f"from {cheapest_flight.origin_airport} "
                         f"to {cheapest_flight.destination_airport}, "
                         f"on {cheapest_flight.out_date}."
        )

▷ Angela Yu, [Python 부트캠프 : 100개의 프로젝트로 Python 개발 완전 정복], Udemy, https://www.udemy.com/course/best-100-days-python/?couponCode=ST3MT72524

JOOYEUN SEO

이전 포스트

API와 구글 시트

다음 포스트

캡스톤 프로젝트 1(항공권 특가 검색기)

100 Days of Python

🗂️ Day39 프로젝트 : 항공권 특가 검색기

1. 구글 시트 준비

2. Sheety API로 구글시트의 데이터 읽고 쓰기

3. Amadeus로 IATA 코드 가져오기

4. 저렴한 항공권 검색

5. 항공권 가격이 구글시트에 있는 가격보다 저렴하면 WhatsApp 알림 전송하기

▷ Angela Yu, [Python 부트캠프 : 100개의 프로젝트로 Python 개발 완전 정복], Udemy, https://www.udemy.com/course/best-100-days-python/?couponCode=ST3MT72524

API와 구글 시트

캡스톤 프로젝트 2(항공 클럽)

0개의 댓글