Building a Flight Price Alert System with Python Scrapers

Building a Flight Price Alert System with Python Scrapers

Flight prices fluctuate constantly. Build your own price tracker that monitors routes, detects price drops, and sends you instant alerts — saving hundreds on every trip.

Architecture Overview

1. Scraper: Fetches prices from flight search engines
2. Database: Stores historical price data
3. Analyzer: Detects price drops and trends
4. Notifier: Sends alerts via email or Telegram

Setup

pip install requests beautifulsoup4 selenium pandas sqlite3 schedule

The Flight Price Scraper

import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
import json
import time

class FlightScraper:
    def __init__(self, proxy_api_key=None):
        self.session = requests.Session()
        self.proxy_key = proxy_api_key

    def fetch_page(self, url):
        """Fetch with optional proxy support."""
        if self.proxy_key:
            api_url = f"http://api.scraperapi.com?api_key={self.proxy_key}&url={url}&render=true"
            return self.session.get(api_url, timeout=60)
        return self.session.get(url, headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
        })

    def search_google_flights(self, origin, destination, date):
        """Scrape Google Flights results."""
        url = f"https://www.google.com/travel/flights?q=Flights+from+{origin}+to+{destination}+on+{date}"
        response = self.fetch_page(url)
        return self.parse_flight_results(response.text)

    def parse_flight_results(self, html):
        """Extract flight data from search results."""
        soup = BeautifulSoup(html, "html.parser")
        flights = []

        # Google Flights uses specific data attributes
        result_cards = soup.select("[data-resultid]")

        for card in result_cards:
            try:
                price_el = card.select_one("[data-gs]") or card.select_one(".price")
                airline_el = card.select_one(".airline-name") or card.select_one("[data-airline]")
                duration_el = card.select_one(".duration")

                flights.append({
                    "price": self.clean_price(price_el.text if price_el else "0"),
                    "airline": airline_el.text.strip() if airline_el else "Unknown",
                    "duration": duration_el.text.strip() if duration_el else "N/A",
                    "scraped_at": datetime.now().isoformat()
                })
            except (AttributeError, ValueError):
                continue

        return flights

    def clean_price(self, price_str):
        """Convert price string to float."""
        return float(price_str.replace("$", "").replace(",", "").strip())

Price Database

import sqlite3
from contextlib import contextmanager

class PriceDatabase:
    def __init__(self, db_path="flights.db"):
        self.db_path = db_path
        self.init_db()

    def init_db(self):
        with self.get_conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS prices (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    origin TEXT NOT NULL,
                    destination TEXT NOT NULL,
                    travel_date TEXT NOT NULL,
                    airline TEXT,
                    price REAL NOT NULL,
                    duration TEXT,
                    scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS alerts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    origin TEXT NOT NULL,
                    destination TEXT NOT NULL,
                    travel_date TEXT NOT NULL,
                    target_price REAL NOT NULL,
                    email TEXT NOT NULL,
                    active BOOLEAN DEFAULT 1
                )
            """)

    @contextmanager
    def get_conn(self):
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        finally:
            conn.close()

    def save_prices(self, origin, dest, date, flights):
        with self.get_conn() as conn:
            for flight in flights:
                conn.execute(
                    "INSERT INTO prices (origin, destination, travel_date, airline, price, duration) VALUES (?, ?, ?, ?, ?, ?)",
                    (origin, dest, date, flight["airline"], flight["price"], flight["duration"])
                )

    def get_price_history(self, origin, dest, date):
        with self.get_conn() as conn:
            cursor = conn.execute(
                "SELECT MIN(price), AVG(price), MAX(price), COUNT(*) FROM prices WHERE origin=? AND destination=? AND travel_date=?",
                (origin, dest, date)
            )
            row = cursor.fetchone()
            return {
                "min": row[0], "avg": row[1],
                "max": row[2], "samples": row[3]
            }

Price Drop Detection

class PriceAnalyzer:
    def __init__(self, db):
        self.db = db

    def detect_drops(self, origin, dest, date, threshold_pct=10):
        """Detect if current price is significantly lower than average."""
        history = self.db.get_price_history(origin, dest, date)

        if not history["avg"] or history["samples"] < 3:
            return None

        with self.db.get_conn() as conn:
            cursor = conn.execute(
                "SELECT price FROM prices WHERE origin=? AND destination=? AND travel_date=? ORDER BY scraped_at DESC LIMIT 1",
                (origin, dest, date)
            )
            current = cursor.fetchone()

        if not current:
            return None

        current_price = current[0]
        drop_pct = ((history["avg"] - current_price) / history["avg"]) * 100

        if drop_pct >= threshold_pct:
            return {
                "current": current_price,
                "average": round(history["avg"], 2),
                "drop_pct": round(drop_pct, 1),
                "all_time_low": history["min"]
            }
        return None

Alert System

import smtplib
from email.mime.text import MIMEText

def send_price_alert(route, price_data, recipient):
    """Send email alert for price drop."""
    subject = f"Flight Deal: {route} - ${price_data['current']} ({price_data['drop_pct']}% below avg)"
    body = f"""
    Price Alert for {route}!

    Current Price: ${price_data['current']}
    Average Price: ${price_data['average']}
    Drop: {price_data['drop_pct']}%
    All-Time Low: ${price_data['all_time_low']}

    Book now before prices go back up!
    """
    print(f"Alert sent to {recipient}: {subject}")

Running the Monitor

import schedule

def monitor_flights():
    scraper = FlightScraper(proxy_api_key="your_key")
    db = PriceDatabase()
    analyzer = PriceAnalyzer(db)

    routes = [
        ("NYC", "LAX", "2025-07-15"),
        ("SFO", "LHR", "2025-08-01"),
        ("CHI", "MIA", "2025-06-20"),
    ]

    for origin, dest, date in routes:
        flights = scraper.search_google_flights(origin, dest, date)
        if flights:
            db.save_prices(origin, dest, date, flights)
            drop = analyzer.detect_drops(origin, dest, date)
            if drop:
                send_price_alert(f"{origin}->{dest}", drop, "[e-mail]")
        time.sleep(5)  # Respectful delay

# Check prices every 6 hours
schedule.every(6).hours.do(monitor_flights)
print("Flight price monitor started...")
while True:
    schedule.run_pending()
    time.sleep(60)

Scaling with Proxy Services

Flight sites have aggressive anti-bot measures. Use:

• ScraperAPI — JS rendering and CAPTCHA solving for Google Flights
• ThorData — residential proxies to avoid detection
• ScrapeOps — monitor scraper health and success rates

Conclusion

Your personal flight deal finder is now running. Extend it with a web interface, support more routes, or add Telegram notifications for instant mobile alerts.

---

Follow for more Python automation projects!