[✈️] 항공권 가격 검색 앱 개발해보기 <6> 크롤링

이제 실제 항공권 데이터를 받아오기 위한 작업을 해볼 것이다.

 

먼저 항공권이 판매되는 시스템부터 알아보면, 항공사 자체 홈페이지에서 판매하는 항공권이 있고 스카이스캐너같은 사이트에 제공하는 항공권등이 따로 있어서 어디에는 올라와있는데 어디에는 없거나, 같은 항공편이라도 가격이 다른 상황이 발생한다.

 

그래서 최대한 많은 곳에서 데이터를 받아온 뒤, 판매처별로 가격을 보여주는 식으로 구조를 변경하는 게 좋을 것 같다.

 

먼저 서버측의 요청, 반환 데이터 구조를 수정해준다.

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
from search_naver import search_naver_flights

app = FastAPI()

# 반환할 데이터 구조
class PriceItem(BaseModel):
    provider : str
    price : int

class Ticket(BaseModel):
    airline : str
    departTime : str
    arrivalTime : str
    prices : List[PriceItem]

# Request 구조
class SearchRequest(BaseModel):
    from_: str
    to: str
    date: str # YYYY-MM-DD

# Response 구조
class SearchResponse(BaseModel):
    results: List[Ticket]

@app.post("/search", response_model=SearchResponse)
def search(req: SearchRequest):
    print("[REQUEST]", req.from_, "→", req.to, req.date) # Log
    results_json = search_naver_flights(req.from_, req.to, req.date)
    results = [Ticket(**r) for r in results_json] # Pydantic 모델로 변환
    return {"results": results}

 

플러터측도 구조에 맞게 수정해준다.

import 'dart:convert';
import 'package:http/http.dart' as http;
import 'dart:io';
import 'package:flutter/foundation.dart';

class TicketApi {
  static String get baseUrl {
    if (kIsWeb) return "http://127.0.0.1:8000";
    if (Platform.isAndroid) return "http://10.0.2.2:8000";

    return "http://127.0.0.1:8000";
  }

  static Future<List<dynamic>> searchFlights({required String from, required String to, required String date,}) async {
    final url = Uri.parse("$baseUrl/search");

    final response = await http.post(
      url,
      headers: {"Content-Type": "application/json"},
      body: jsonEncode({
        "from_": from,
        "to": to,
        "date": date,
      }),
    );

    if (response.statusCode == 200) {
      final data = jsonDecode(response.body);
      return data["results"];
    }
    else{ throw Exception("서버 오류: ${response.statusCode}"); }
  }
}

 

구조를 수정했으면 이제 크롤링을 구현해보자.

크롤링은 Playwright 라이브러리를 사용하여 구현할것이다.

Playwright는 브라우저를 코드로 조작할 수 있는 자동화 엔진이다.

 

pip install playwright
python -m playwright install
로 라이브러리를 설치해주고

 

이번에는 네이버 항공권을 예로 들면, 항공권 검색 시

https://flight.naver.com/flights/international/ICN:airport-TYO:city-20260113?adult=1&fareType=Y

이런 식으로 검색 조건(출발지, 도착지, 날짜, 인원 수 등)이 url에 포함되어있는 것을 볼 수 있다.

 

이를 이용해서 검색 자체는 사용자의 입력을 url에 포함시키는 식으로 간단하게 구현할 수 있다!

(입력받은 변수를 각 사이트에 맞는 형식으로 변환 필요)

 

그렇다면 남은 건 검색한 결과를 데이터로 정리하여 보여주는 것이다.

 

검색 결과창에서 브라우저(크롬)의 개발자 도구(F12)로 확인해보면

사진처럼 원하는 데이터의 클래스명을 확인할 수 있다.

 

필요한 클래스명들을 찾았으면 크롤링용 파이썬 파일을 하나 생성한 뒤 코드를 작성해준다.

from playwright.sync_api import sync_playwright
import time

NAVER_CODE_MAP = {
    "INCHEON": "ICN",
    "TOKYO": "TYO",
}

def search_naver_flights(from_: str, to: str, date: str):
    date_str = date.replace("-", "")
    naver_from = NAVER_CODE_MAP.get(from_, from_)
    naver_to = NAVER_CODE_MAP.get(to, to)

    url = f"https://flight.naver.com/flights/international/{naver_from}:airport-{naver_to}:city-{date_str}?adult=1&fareType=Y"

    results = []

    try:
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=False)
            page = browser.new_page()
            page.goto(url, wait_until="networkidle", timeout=10000)
            
            # 항공사명 요소를 찾을때까지 대기 (최대 10초)
            airline_found = False
            for i in range(10):
                time.sleep(1)
                airline_count = page.locator('[class*="airline_name__"]').count()
                if airline_count > 0:
                    airline_found = True
                    break
            
            if not airline_found:
                print("[WARNING] 항공편을 찾을 수 없습니다")
                browser.close()
                return results
            
            flight_items = page.locator('[class*="combination_ConcurrentItemContainer__"]')
            flight_count = flight_items.count()
            print(f"{flight_count}개 항공권 발견")
            
            if flight_count == 0:
                print("[WARNING] flight_items을 찾을 수 없습니다.")
                browser.close()
                return results
            
            # 각 항공편 컨테이너 내에서 데이터 추출
            for i in range(flight_count):
                try:
                    flight_item = flight_items.nth(i)
                    
                    # 항공사명
                    airline = flight_item.locator('[class*="airline_name__"]').inner_text()
                    
                    # 출발시간, 도착시간
                    route_times = flight_item.locator('[class*="route_time__"]')
                    depart_time = route_times.nth(0).inner_text()
                    arrival_time = route_times.nth(1).inner_text()
                    
                    # 가격
                    price_text = flight_item.locator('[class*="item_num__"]').first.inner_text()
                    price = int(price_text.replace(",", "").replace("원", "").strip())
                    
                    results.append({
                        "airline": airline.strip(),
                        "departTime": depart_time.strip(),
                        "arrivalTime": arrival_time.strip(),
                        "prices": [{"provider": "네이버항공권", "price": price}]
                    })
                    
                    # print(f"[{i+1:2d}] {airline.strip():15s} {depart_time.strip()}→{arrival_time.strip():8s} - {price:,}원")
                    
                except Exception as e:
                    print(f"[ERROR-{i}] 항공권 {i} 파싱 실패: {e}")
                    continue
            
            # 중복 제거
            seen = set()
            unique_results = []
            for flight in results:
                key = (flight['airline'], flight['departTime'], flight['arrivalTime'], flight['prices'][0]['price'])
                if key not in seen:
                    seen.add(key)
                    unique_results.append(flight)
            
            results = unique_results
            browser.close()
            print("크롤링 완료")
            
    except Exception as e:
        print(f"[FATAL ERROR] {type(e).__name__}: {e}")
        import traceback
        traceback.print_exc()
        raise

    return results

 

위에서부터 살펴보면 먼저 날짜, 출발지, 도착지 등을 네이버항공권 사이트에 맞게 변환해주고,

해당 url로 브라우저를 열고 스크립트가 렌더링이 된 것을 확인했으면 각 항공권 컨테이너(묶음)에서 항공사명, 출발시간, 도착시간, 가격등을 읽어와 results로 정리한 뒤 반환해준다.

 

플러터의 기존 UI도 바뀐 형식에 맞게 수정해주면

실제 항공권 데이터가 잘 불러와진 모습이다!

이제 다른 항공사나 항공권 사이트들을 추가해주면 된다.