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

JOOYEUN SEO·2024년 9월 23일
pythonudemy

100 Days of Python

목록 보기
40/76
post-thumbnail

🗂️ Day40 프로젝트 : 항공 클럽

🗂️ Day39 프로젝트 : 항공권 특가 검색기를 업그레이드 한 프로그램

  • 항공 클럽에 등록된 모든 사용자들에게 특가 항공권 알림 이메일을 보냄
  • jacksflightclub를 유사하게 구현 (등록한 이메일로 특가 알림을 보내는 서비스)

1. 고객 등록 코드 만들기

🔍 유의 사항

  • 사용자가 고객 등록 프로그램에 접근할 수 있도록 레플릿으로 코드를 호스팅
    • 레플릿 프로젝트에서 사용자에게 성, 이름, 이메일을 묻는 코드 작성
    • 이메일 주소 확인을 위해 한 번 더 입력하게 하기
    • 두 번의 이메일 입력이 일치하면, 사용자가 클럽에 가입되었다고 안내하기
  • 원래 구글 시트에 users라는 이름으로 새 시트(탭) 생성
  • Sheety에서 원래 프로젝트를 Refresh하여 업데이트

다음 달에 다시 시도하기🥲

2. 항공편이 없는 도착지 예외처리

🔍 유의 사항

  • 일부 도착지나 특정 기간에는 항공편이 없는 경우 존재
  • 데이터가 비어 있어도 오류가 발생하지 않도록 하기

⌨️ flight_search.py

…

class FlightSearch:

…

    def check_flights(self, origin_city_code, destination_city_code, from_time, to_time):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            "originLocationCode": origin_city_code,
            "destinationLocationCode": destination_city_code,
            "departureDate": from_time.strftime("%Y-%m-%d"),
            "returnDate": to_time.strftime("%Y-%m-%d"),
            "adults": "1",
            "nonStop": "true",
            "currencyCode": "GBP",
            "max": "5"
        }

        response = requests.get(url=FLIGHT_ENDPOINT,
                                headers=headers,
                                params=query)

        # 현재 기준 가장 먼저 나오는 데이터가 가장 저렴하기 때문에 [0]만 체크
        try:
            data = response.json()["data"][0]
        except IndexError:
            print(f"No flights found for {destination_city_code}.")
            return None
        else:
            flight_data = FlightData(
                price=data["price"]["grandTotal"],
                origin_airport=data["itineraries"][0]["segments"][0]["departure"]["iataCode"],
                destination_airport=data["itineraries"][0]["segments"][0]["arrival"]["iataCode"],
                out_date=data["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0],
                return_date=data["itineraries"][0]["segments"][0]["arrival"]["at"].split("T")[0],
            )
            return flight_data

⌨️ main.py

…

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

for destination in sheet_data:
    flight = flight_search.check_flights(
        ORIGIN_CITY_IATA,
        destination["iataCode"],
        from_time=tomorrow,
        to_time=return_day
    )

    if flight is None:
        continue
        
    if flight.price < destination["lowestPrice"]:
        …

3. 직항편이 없는 도착지

🔍 유의 사항

  • 항공권이 검색되지 않으면 경유하는 항공권이 있는지 확인하기
    (예를 들어 런던 → 발리의 경우 직항편이 없음)
  • nonstop 옵션을 false로 바꾸면 경유하는 옵션을 넣어서 검색 가능

⌨️ flight_search.py

…

class FlightSearch:

…

    def check_flights(self, origin_city_code, destination_city_code, from_time, to_time):
        headers = {"Authorization": f"Bearer {self.token}"}
        query = {
            …
        }

        response = requests.get(url=FLIGHT_ENDPOINT,
                                headers=headers,
                                params=query)

        try:
            data = response.json()["data"][0]
        except IndexError:
            query["nonStop"] = "false"
            response = requests.get(url=FLIGHT_ENDPOINT,
                                    headers=headers,
                                    params=query)
            data = response.json()["data"][0]

            flight_data = FlightData(
                price=data["price"]["grandTotal"],
                origin_airport=data["itineraries"][0]["segments"][0]["departure"]["iataCode"],
                destination_airport=data["itineraries"][0]["segments"][1]["arrival"]["iataCode"],
                out_date=data["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0],
                return_date=data["itineraries"][0]["segments"][1]["arrival"]["at"].split("T")[0],
                via_city=data["itineraries"][0]["segments"][0]["arrival"]["iataCode"]
            )
            return flight_data
        else:
            flight_data = FlightData(
                price=data["price"]["grandTotal"],
                origin_airport=data["itineraries"][0]["segments"][0]["departure"]["iataCode"],
                destination_airport=data["itineraries"][0]["segments"][0]["arrival"]["iataCode"],
                out_date=data["itineraries"][0]["segments"][0]["departure"]["at"].split("T")[0],
                return_date=data["itineraries"][0]["segments"][0]["arrival"]["at"].split("T")[0],
            )
            return flight_data

⌨️ main.py

…
    if flight is None:
        continue
        
    if flight.price < destination["lowestPrice"]:
        message = (f"Low price alert! Only £{flight.price} to fly"
                   f"from {flight.origin_airport} to {flight.destination_airport},"
                   f"from {flight.out_date} to {flight.return_date}.")

        if flight.via_city != "":
            message += f"\nFlight has a stop over, via {flight.via_city}."
…

4. 전체 고객에게 이메일 보내기

🔍 유의 사항

  • 시트에 있는 고객 모두에게 최저가 항공권 정보가 포함된 이메일을 발송
  • 이메일을 보낼때 통화 기호 등의 기호를 넣으면 오류 발생 가능
    • 메시지를 UTF-8로 인코딩하여 해결
  • 구글 항공편 링크를 첨부하여 바로 예매할 수 있도록 하기
    • URL에서 필요한 부분 변경하기

⌨️ data_manager.py

import os
import requests
from dotenv import load_dotenv

load_dotenv()

SHEETY_PRICES_ENDPOINT = os.getenv("SHEETY_PRICES_ENDPOINT")
SHEETY_USERS_ENDPOINT = os.getenv("SHEETY_USERS_ENDPOINT")
BEARER_TOKEN = os.getenv("BEARER_TOKEN")

class DataManager:

    def __init__(self):
        self.headers = {"Authorization": f"Bearer {BEARER_TOKEN}"}
        self.destination_data = {}

    def get_destination_data(self):
        response = requests.get(url=SHEETY_PRICES_ENDPOINT, headers=self.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"{SHEETY_PRICES_ENDPOINT}/{city['id']}",
                json=new_data
            )
            print(response.text)

    def get_customer_emails(self):
        customers_endpoint = SHEETY_USERS_ENDPOINT
        response = requests.get(url=customers_endpoint, headers=self.headers)
        data = response.json()
        self.customer_data = data["users"]
        return self.customer_data

⌨️ flight_data.py

class FlightData:

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

⌨️ flight_search.py

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

ORIGIN_CITY_IATA = "LON"

data_manager = DataManager()
sheet_data = data_manager.get_destination_data()
flight_search = FlightSearch()
notification_manager = NotificationManager()

if sheet_data[0]["iataCode"] == "":
    for row in sheet_data:
        row["iataCode"] = flight_search.get_destination_code(row["city"])
    data_manager.destination_data = sheet_data
    data_manager.update_destination_codes()

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

for destination in sheet_data:
    flight = flight_search.check_flights(
        ORIGIN_CITY_IATA,
        destination["iataCode"],
        from_time=tomorrow,
        to_time=return_day
    )

    if flight is None:
        continue

    if flight.price < destination["lowestPrice"]:
        users = data_manager.get_customer_emails()
        emails = [row["email"] for row in users]
        names = [row["firstName"] for row in users]

        message = (f"Low price alert! Only £{flight.price} to fly"
                   f"from {flight.origin_airport} to {flight.destination_airport},"
                   f"from {flight.out_date} to {flight.return_date}.")

        if flight.via_city != "":
            message += f"\nFlight has a stop over, via {flight.via_city}."

        notification_manager.send_emails(emails, message)

⌨️ notification_manager.py

import os
from dotenv import load_dotenv
import smtplib

load_dotenv()

class NotificationManager:

    def __init__(self):
        self.my_email = os.getenv("MY_EMAIL")
        self.my_password = os.getenv("MY_PASSWORD")

    def send_emails(self, emails, message):
        message += "\nhttps://www.google.com/travel/flights?authuser=9&hl=en"
        with smtplib.SMTP("smtp.gmail.com") as connection:
            connection.starttls()
            connection.login(self.my_email, self.my_password)
            for email in emails:
                connection.sendmail(
                    from_addr=self.my_email,
                    to_addrs=email,
                    msg=f"Subject:New Low Price Flight!\n\n{message}".encode('utf-8')
                )

⌨️ main.py

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

ORIGIN_CITY_IATA = "LON"

data_manager = DataManager()
sheet_data = data_manager.get_destination_data()
flight_search = FlightSearch()
notification_manager = NotificationManager()

if sheet_data[0]["iataCode"] == "":
    for row in sheet_data:
        row["iataCode"] = flight_search.get_destination_code(row["city"])
    data_manager.destination_data = sheet_data
    data_manager.update_destination_codes()

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

for destination in sheet_data:
    flight = flight_search.check_flights(
        ORIGIN_CITY_IATA,
        destination["iataCode"],
        from_time=tomorrow,
        to_time=return_day
    )

    if flight is None:
        continue

    if flight.price < destination["lowestPrice"]:
        users = data_manager.get_customer_emails()
        emails = [row["email"] for row in users]
        names = [row["firstName"] for row in users]

        message = (f"Low price alert! Only £{flight.price} to fly"
                   f"from {flight.origin_airport} to {flight.destination_airport},"
                   f"from {flight.out_date} to {flight.return_date}.")

        if flight.via_city != "":
            message += f"\nFlight has a stop over, via {flight.via_city}."

        notification_manager.send_emails(emails, message)




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

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

다음 포스트

HTML 소개

0개의 댓글