Day 040 - 항공권 가격 추적 프로젝트 (심화)
Reo
Day 040 - 항공권 가격 추적 프로젝트 (심화)◎ Python/Udemy Python2024. 3. 8. 20:27@Reo
Table of Contents
반응형

🖥️ 시작하며

이전 포스팅에서는 개인 사용자들을 위한, 로컬 사용자들만을 위한 프로그램이었습니다. 그렇다면 이번에는 여러 사용자들을 가입시켜 이메일로 할인 정보를 보내주도록 코드를 업데이트 해보도록 하겠습니다. 이 사이트와 비슷한 기능을 수행합니다!

💡 로직 순서

  • 사용자의 성, 이름, 이메일을 받아와서 스프레드시트에 저장합니다.
  • 사실 간단합니다. 해당 정보들만 가져오면, API 서비스를 이용해 사용자들에게 이메일만 보내면 됩니다.

주의사항!

Sheety API를 사용할 때, 열의 제목을 JSON에서 취급할 때는 아래 사진과 같게 해야 합니다..
요즈음 API를 건들 때마다 Docs를 잘 봐야겠다는 생각이 항상 드네요.

 

또한 이전 코드에서 고쳐야 할 점은 다음과 같습니다.

  1. 항공편이 아예 없는 경우, 오류가 생길 수 있습니다.
    • IndexError를 활용해 예외처리를 하도록 합니다.
  2. 직항편이 없는 경우
    • 1회 경유하는 항공권이 있는지 확인합니다. 만약 존재하면 이를 출력합니다.

 

항공편이 아예 없는 경우

항공편이 없으면 flight_search.py파일에서 API를 받아올 때 인덱스 에러가 날 수 있으므로 try/except구문을 추가해 줍니다.

직항편이 없는 경우

직항편이 없는 경우, 우선 항공편이 없는 경우에 직항편이 있을 수도 있으므로 위 경우와 합쳐 try/except구문을 두 번 활용해야 합니다.

# 인덱스 오류가 나는지 확인
# 항공편이 아예 없는 경우와 직항편이 없는 경우가 있을 수 있음
try:
    data = r.json()["data"][0]
except IndexError:
    # 만약 경유편이라도 있는지 확인해야 하므로, Query를 업데이트 해야 함
    param["max_stopovers"] = 1
    r = requests.get(url=local_endpoint, params=param, headers=cls.headers)

    # 경유편도 없으면
    try:
        data = r.json()["data"][0]
    except IndexError:
        print(f"{IATA_Code}로 가는 항공편이 없습니다.")
        return None
    else:
        flight_data = FlightData(
            price=data["price"],
            origin_city=data["route"][0]["cityFrom"],
            origin_airport=data["route"][0]["flyFrom"],
            destination_city=data["route"][1]["cityTo"],
            destination_airport=data["route"][1]["flyTo"],
            out_date=data["route"][0]["local_departure"].split("T")[0],
            return_date=data["route"][2]["local_departure"].split("T")[0],
            stop_overs=1,
            via_city=data["route"][0]["cityTo"]
        )
        return flight_data
# 직항편이 있으면
else:

 

⚙️ 코드 전문

main.py

# main.py

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

data = DataManager()
sheet_data = data.get_data()

# iataCode가 비어있을 시
for i in sheet_data:
    if i['iataCode'] == '':
        i['iataCode'] = FlightSearch.get_iatacode(city=i['city'])
        data.update_data(sheet_data=sheet_data)

from_date = datetime.now() + timedelta(days = 1)
to_date = datetime.now() + timedelta(days = 180)

# 최저가 검색
for i in sheet_data:
    # 최저가를 찾기 위해 데이터 입력
    flight = FlightSearch.search_flight_price(
        i['iataCode'],
        from_date.strftime("%d/%m/%Y"),
        to_date.strftime("%d/%m/%Y")
    )

    # 만약 설정한 최저가보다 가격이 낮다면 메세지 전송
    # 직항편이 없다면 flight가 None이므로 스킵하도록 함

    # 유저 이메일 가져오기
    user_email = data.get_user_email()

    if flight is not None and flight.price < i['lowestPrice']:
        message=f"최저가가 발견되었습니다! {flight.from_city}-{flight.from_airport} 에서 {flight.to_city}-{flight.to_airport}로 가는 항공편이 ₩{flight.price}입니다. \n출발시각:{flight.out_date}\n도착시각{flight.return_date}"

        # 만약 경유하면 메세지 추가
        if flight.stop_overs > 0:
            message += f"\n이 비행은 {flight.stop_overs}번 {flight.via_city}를 경유합니다."

        NotificationManager.send_message(emails=user_email, message=message)

data_manager.py

# data_manager.py

from dotenv import load_dotenv
import os
import requests
from pprint import pprint

load_dotenv()

class DataManager:
    def __init__(self) -> None:
        self.endpoint = os.getenv('SHEETY_API_FLIGHT')
        self.endpoint_user = os.getenv('SHEETY_API_USERS')

    """구글 시트 데이터 가져오기"""
    def get_data(self):
        r = requests.get(url=self.endpoint)
        return r.json()['prices']

    """구글 시트에 IATA Code 업데이트"""
    def update_data(self, sheet_data):
        for city in sheet_data:
            new_data = {
                "price": {
                    "iataCode": city["iataCode"]
                }
            }
            r = requests.put(
                url=f"{self.endpoint}/{city['id']}",
                json=new_data
            )
            # print(r.text)

    """유저들 이메일 가져오기"""    
    def get_user_email(self):
        r = requests.get(url=self.endpoint)
        return r.json()['users']

flight_search.py

# flight_search.py

from dotenv import load_dotenv
import os
import requests
from flight_data import FlightData

load_dotenv()

class FlightSearch:
    # 알게된 점 : classmethod를 사용할 거면 cls를 사용해야 한다.
    KEY = os.getenv("KIWI_API_KEY")
    IATA_endpoint = "https://api.tequila.kiwi.com"
    PRICE_endpoint = "https://tequila-api.kiwi.com/v2"
    headers = {
        "apikey": KEY
    }    

    """API에서 IATA Code 가져오기"""
    @classmethod
    def get_iatacode(cls, city):
        local_endpoint = f"{cls.IATA_endpoint}/locations/query"
        param = {
            "term": city,
            "location_types": "city",
        }

        r = requests.get(url=local_endpoint, params=param, headers=cls.headers)
        res = r.json()['locations']
        IATA_Code = res[0]['code']

        return IATA_Code

    """최저가 항공권 검색"""
    @classmethod
    def search_flight_price(cls, IATA_Code, from_date, to_date):
        local_endpoint = f"{cls.PRICE_endpoint}/search"
        param = {
            'fly_from': "ICN",
            'fly_to': IATA_Code,
            'date_from': from_date,
            'date_to': to_date,
            # 여기 아래 항목들 안넣으면 오류남
            "nights_in_dst_from": 7,
            "nights_in_dst_to": 28,
            "one_for_city": 1,
            "max_stopovers": 0,
            'curr': "KRW",
        }

        r = requests.get(url=local_endpoint, params=param, headers=cls.headers)

        # 인덱스 오류가 나는지 확인
        # 항공편이 아예 없는 경우와 직항편이 없는 경우가 있을 수 있음
        try:
            data = r.json()["data"][0]
        except IndexError:
            # 만약 경유편이라도 있는지 확인해야 하므로, Query를 업데이트 해야 함
            param["max_stopovers"] = 1
            r = requests.get(url=local_endpoint, params=param, headers=cls.headers)

            # 경유편도 없으면
            try:
                data = r.json()["data"][0]
            except IndexError:
                print(f"{IATA_Code}로 가는 항공편이 없습니다.")
                return None
            else:
                flight_data = FlightData(
                    price=data["price"],
                    origin_city=data["route"][0]["cityFrom"],
                    origin_airport=data["route"][0]["flyFrom"],
                    destination_city=data["route"][1]["cityTo"],
                    destination_airport=data["route"][1]["flyTo"],
                    out_date=data["route"][0]["local_departure"].split("T")[0],
                    return_date=data["route"][2]["local_departure"].split("T")[0],
                    stop_overs=1,
                    via_city=data["route"][0]["cityTo"]
                )
                return flight_data
        # 직항편이 있으면
        else:
            flight_data = FlightData(
                price=data["price"],
                from_city=data["route"][0]["cityFrom"],
                from_airport=data["route"][0]["flyFrom"],
                to_city=data["route"][0]["cityTo"],
                to_airport=data["route"][0]["flyTo"],
                out_date=data["route"][0]["local_departure"].split("T")[0],
                return_date=data["route"][1]["local_departure"].split("T")[0]
            )

            print(f"{flight_data.to_city}: ₩{flight_data.price}")
            return flight_data

flight_data.py

# flight_data.py

class FlightData:

    def __init__(self, price, from_city, from_airport, to_city, to_airport, out_date, return_date, stop_overs=0, via_city=""):
        self.price = price
        self.from_city = from_city
        self.from_airport = from_airport
        self.to_city = to_city
        self.to_airport = to_airport
        self.out_date = out_date
        self.return_date = return_date

        # 직항편이 없을 경우
        self.stop_overs = stop_overs
        self.via_city = via_city

notification_manager.py

# notification_manager.py

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

load_dotenv()

class NotificationManager:
    account_sid = os.getenv('ACCOUNT_SID')
    auth_token = os.getenv('AUTH_TOKEN')
    client = Client(account_sid, auth_token)
    MY_EMAIL = os.getenv('MY_EMAIL')
    MY_PASSWORD = os.getenv('MY_PASSWORD')

    @classmethod
    def send_message(cls, emails, message):
        with smtplib.SMTP("smtp.gmail.com") as connection:
            connection.starttls()
            connection.login(cls.MY_EMAIL, cls.MY_PASSWORD)
            for email in emails:
                connection.sendmail(
                    from_addr=cls.MY_EMAIL,
                    to_addrs=email,
                    msg=f"Subject:New Low Price Flight!\n\n{message}".encode('utf-8')
                )


        # 메세지 버전
        # message = cls.client.messages.create(
        #     to=os.getenv("MY_PHONE_NUMBER"),
        #     from_=os.getenv("MY_TO_NUMBER"),
        #     body=message)

        # print(message.sid)

 

유저를 받기 위한 replit..?

일단 연동은 시켜놓지 않았는데.. 만들어 보긴 했습니다.
https://replit.com/@reo91004/FlightInputData

 

부록

참고문헌

반응형
저작자표시

'◎ Python > Udemy Python' 카테고리의 다른 글

Day 039 - 항공권 가격 추적 프로젝트  (0) 2024.03.08
Day 038 - 구글 시트에 운동 기록  (1) 2024.03.08
Day 037 - 습관 추적기 프로젝트  (0) 2024.03.08
Day 036 - 주식시장 알림 프로젝트  (1) 2024.03.08
Day 033 - API 활용  (1) 2024.03.08
Reo
@Reo :: 코드 아카이브

자기계발 블로그

◎ Python/Udemy Python 추천 글
more
image

티스토리툴바