# 🚀 1. IMPORTS & SETUP import asyncio import datetime import logging import os import random import gradio as gr from cachetools import TTLCache from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI from playwright.async_api import async_playwright # 🥳 Logging config logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # 🧠 Simple in-memory cache (1h TTL) ttl_cache = TTLCache(maxsize=100, ttl=3600) # User agents for stealth USER_AGENTS = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' ] # 📦 2. DATA MODEL class TravelRequest(BaseModel): origin_city: str = Field(description="Starting city or airport code (e.g., YHZ)") destination_city: str = Field(description="Destination city or airport code (e.g., YYZ)") start_date: str = Field(description="Trip start date in YYYY-MM-DD format") end_date: str = Field(description="Trip end date in YYYY-MM-DD format") budget: int = Field(description="Budget in USD (informational only)") interests: list[str] = Field(description="List of travel interests") # ✨ 3. EXTRACT USER REQUEST async def _extract_user_request(question: str) -> dict: llm = ChatOpenAI(model="gpt-4o", temperature=0) parser = JsonOutputParser(pydantic_object=TravelRequest) prompt = ChatPromptTemplate.from_messages([ ("system", "You are a travel assistant. Extract flight details from the user's request. Assume the current year is 2025 unless specified otherwise. Today's date is July 8, 2025."), ("human", "{format_instructions}\n{request}") ]).partial(format_instructions=parser.get_format_instructions()) logging.info(f"🔍 Extracting details from: {question}") return await (prompt | llm | parser).ainvoke({"request": question}) # 🛠 4. CACHE WRAPPER async def _cached_search(func, details): key = (func.__name__, details['origin_city'], details['destination_city'], details['start_date']) if key in ttl_cache: logging.info(f"🔁 Cache hit for {func.__name__}") return ttl_cache[key] logging.info(f"▶️ Cache miss for {func.__name__}") res = await func(details) ttl_cache[key] = res return res # 🌐 5. PLAYWRIGHT SCRAPER FUNCTIONS async def _search_google_flights(details): results = [] try: async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) context = await browser.new_context(user_agent=random.choice(USER_AGENTS)) page = await context.new_page() # Add random delay await asyncio.sleep(random.uniform(1, 3)) o = details["origin_city"] d = details["destination_city"] sd = details["start_date"] ed = details["end_date"] # More reliable Google Flights URL url = f"https://www.google.com/travel/flights/search?tfs=CBwQAhokEgoyMDI1LTA4LTAxagwIAhIIL20vMDVxdGwyDAg" await page.goto(url, timeout=60000) # Wait for any flight results to load try: await page.wait_for_selector('[data-testid="flight-offer"]', timeout=20000) cards = await page.query_selector_all('[data-testid="flight-offer"]') for card in cards[:3]: try: price_el = await card.query_selector('[data-testid="price-text"]') if not price_el: price_el = await card.query_selector('span[aria-label*="dollars"]') if price_el: price_text = await price_el.inner_text() price_clean = ''.join(filter(str.isdigit, price_text)) price = float(price_clean) if price_clean else 0.0 results.append({ "source": "Google Flights", "type": "flight", "details": price_text.strip(), "price": price, "link": url }) except Exception as e: logging.warning(f"Error parsing Google Flights card: {e}") continue except Exception as e: logging.warning(f"Google Flights selector timeout: {e}") # Fallback: try to find any price-like elements price_elements = await page.query_selector_all('span[aria-label*="dollar"], span[aria-label*="price"]') for elem in price_elements[:3]: try: text = await elem.inner_text() if '$' in text: results.append({ "source": "Google Flights", "type": "flight", "details": text.strip(), "price": 0.0, "link": url }) except: continue await browser.close() except Exception as e: logging.error(f"Google Flights scraping failed: {e}") return results async def _search_kayak(details): results = [] try: async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) context = await browser.new_context(user_agent=random.choice(USER_AGENTS)) page = await context.new_page() await asyncio.sleep(random.uniform(1, 3)) o = details["origin_city"].upper() d = details["destination_city"].upper() sd = details["start_date"] ed = details["end_date"] url = f"https://www.kayak.com/flights/{o}-{d}/{sd}/{ed}" await page.goto(url, timeout=60000) # Multiple selector attempts selectors_to_try = [ ".resultWrapper", "[data-testid='result-item']", ".Common-Booking-MultiBookProvider", ".result-column" ] cards = [] for selector in selectors_to_try: try: await page.wait_for_selector(selector, timeout=10000) cards = await page.query_selector_all(selector) if cards: break except: continue for card in cards[:3]: try: # Try multiple price selectors price_selectors = [ ".price-text", "[data-testid='price']", ".Common-Booking-MultiBookProvider-price", "span[aria-label*='price']" ] price_text = None for price_sel in price_selectors: try: price_el = await card.query_selector(price_sel) if price_el: price_text = await price_el.inner_text() break except: continue if price_text: price_clean = ''.join(filter(str.isdigit, price_text)) price = float(price_clean) if price_clean else 0.0 results.append({ "source": "Kayak", "type": "flight", "details": price_text.strip(), "price": price, "link": url }) except Exception as e: logging.warning(f"Error parsing Kayak card: {e}") continue await browser.close() except Exception as e: logging.error(f"Kayak scraping failed: {e}") return results async def _search_trabber(details): results = [] try: async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) context = await browser.new_context(user_agent=random.choice(USER_AGENTS)) page = await context.new_page() await asyncio.sleep(random.uniform(1, 3)) o = details["origin_city"].upper() d = details["destination_city"].upper() sd_parts = details["start_date"].split('-') sd = f"{sd_parts[2]}{sd_parts[1]}{sd_parts[0][2:]}" # DDMMYY format url = f"https://www.trabber.ca/flights-from-{o}-to-{d}-on-{sd}" await page.goto(url, timeout=60000) try: await page.wait_for_selector("#results_list_det tr", timeout=20000) rows = await page.query_selector_all("#results_list_det tr") for row in rows[:3]: try: price_el = await row.query_selector("td.results_price a, .results_price") if price_el: price_text = await price_el.inner_text() price_clean = ''.join(filter(str.isdigit, price_text)) price = float(price_clean) if price_clean else 0.0 results.append({ "source": "Trabber", "type": "flight", "details": price_text.strip(), "price": price, "link": url }) except Exception as e: logging.warning(f"Error parsing Trabber row: {e}") continue except Exception as e: logging.warning(f"Trabber selector timeout: {e}") await browser.close() except Exception as e: logging.error(f"Trabber scraping failed: {e}") return results async def _search_travala(details): results = [] try: async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) context = await browser.new_context(user_agent=random.choice(USER_AGENTS)) page = await context.new_page() await asyncio.sleep(random.uniform(1, 3)) url = "https://www.travala.com/deals" await page.goto(url, timeout=60000) try: # Try multiple selectors for deals selectors = [ "div[class*='DealCard_container__']", ".deal-card", "[data-testid='deal-card']" ] cards = [] for selector in selectors: try: await page.wait_for_selector(selector, timeout=10000) cards = await page.query_selector_all(selector) if cards: break except: continue for card in cards[:5]: try: title_selectors = [ "p[class*='DealCard_title__']", ".deal-title", "h3, h4" ] title = "Travala Deal" for title_sel in title_selectors: try: title_el = await card.query_selector(title_sel) if title_el: title = await title_el.inner_text() break except: continue results.append({ "source": "Travala", "type": "deal", "details": title[:100], "price": 0, "link": url }) except Exception as e: logging.warning(f"Error parsing Travala card: {e}") continue except Exception as e: logging.warning(f"Travala selector timeout: {e}") await browser.close() except Exception as e: logging.error(f"Travala scraping failed: {e}") return results async def _search_secretflying(details): results = [] try: async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True) context = await browser.new_context(user_agent=random.choice(USER_AGENTS)) page = await context.new_page() await asyncio.sleep(random.uniform(1, 3)) url = "https://www.secretflying.com/canada-deals/" await page.goto(url, timeout=60000) try: # Try multiple selectors for posts selectors = [ ".post-item", "article", ".entry" ] posts = [] for selector in selectors: try: await page.wait_for_selector(selector, timeout=10000) posts = await page.query_selector_all(selector) if posts: break except: continue for post in posts[:5]: try: title_selectors = [ "h2.entry-title a", ".entry-title a", "h3 a", "h2 a" ] title = "Deal" link = url for title_sel in title_selectors: try: title_el = await post.query_selector(title_sel) if title_el: title = await title_el.inner_text() href = await title_el.get_attribute("href") if href: link = href if href.startswith('http') else f"https://www.secretflying.com{href}" break except: continue results.append({ "source": "SecretFlying", "type": "deal", "details": title[:100], "price": 0, "link": link }) except Exception as e: logging.warning(f"Error parsing SecretFlying post: {e}") continue except Exception as e: logging.warning(f"SecretFlying selector timeout: {e}") await browser.close() except Exception as e: logging.error(f"SecretFlying scraping failed: {e}") return results # 🔄 6. GATHER ALL DATA async def _gather_travel_data(req): tasks = [ _cached_search(_search_google_flights, req), _cached_search(_search_kayak, req), _cached_search(_search_trabber, req), _cached_search(_search_travala, req), _cached_search(_search_secretflying, req), ] results = await asyncio.gather(*tasks, return_exceptions=True) combined = [] for r in results: if isinstance(r, list): combined.extend(r) else: logging.error(f"A scraper task failed: {r}") return combined # ✍️ 7. FORMAT RESULTS def _format_response(req, data): dest = req["destination_city"] md = f"## 🌏 Travel Plan to **{dest}**\n\n" flights = sorted([i for i in data if i["type"]=="flight" and i.get("price", 0) > 0], key=lambda x: x["price"]) if flights: md += "### ✈️ Flights\n" for item in flights[:8]: md += f"- **{item['source']}**: `{item['details']}` — **${item['price']:.2f}** ([Book]({item['link']}))\n" deals = [i for i in data if i["type"]=="deal"] if deals: md += "\n### 💸 Hot Deals\n" for item in deals[:8]: md += f"- **{item['source']}**: {item['details']} ([Link]({item['link']}))\n" if not flights and not deals: md += "No results found. The websites may be blocking our scrapers or there are no flights available for those dates.\n\n" md += "**Debug Info:**\n" md += f"- Searched for: {req['origin_city']} → {req['destination_city']}\n" md += f"- Dates: {req['start_date']} to {req['end_date']}\n" md += f"- Total scrapers attempted: 5\n" return md # 💬 8. MAIN BOT & UI async def ask_bot(question): if not question: return "❓ Tell me where you want to go!" try: req = await _extract_user_request(question) all_data = await _gather_travel_data(req) return _format_response(req, all_data) except Exception as e: logging.error(f"An error occurred in the main process: {e}") return f"😵 Oh no! Something went wrong: {str(e)}\n\nPlease try asking in a different way or check if your OpenAI API key is set correctly." iface = gr.Interface( fn=ask_bot, inputs=gr.Textbox(lines=4, label="🌍 Your dream trip?", placeholder="e.g., I want to fly from Halifax to Toronto for a week starting August 1st"), outputs=gr.Markdown(label="🚀 Your Travel Plan"), title="🛫 High Flyer AI Bot", description="Your personal AI travel agent. Uses Playwright to scrape multiple sources in real-time." ) if __name__ == "__main__": iface.launch(server_name="0.0.0.0", server_port=7860)