import os import json import requests from datetime import datetime, timedelta from functools import lru_cache import gradio as gr import openai import random # -------------------- CONFIGURATION -------------------- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY", "") SERP_API_KEY = os.getenv("SERPAPI_API_KEY", "") if not OPENAI_API_KEY: print("⚠️ OPENAI_API_KEY not set. AI itinerary generation will fall back to manual mode.") client = None else: client = openai.OpenAI(api_key=OPENAI_API_KEY) # Image cache to avoid repeated API calls IMAGE_CACHE = {} # -------------------- IMAGE FUNCTION -------------------- def get_destination_image(city: str) -> str: """Fetch a real image of the destination using SerpAPI.""" cache_key = city.lower() if cache_key in IMAGE_CACHE: return IMAGE_CACHE[cache_key] if SERP_API_KEY: try: params = { "q": f"{city} city landmark beautiful view", "api_key": SERP_API_KEY, "engine": "google", "tbm": "isch", "num": 1 } response = requests.get("https://serpapi.com/search", params=params, timeout=10) data = response.json() images = data.get("images_results", []) if images and len(images) > 0: img_url = images[0].get("original", images[0].get("thumbnail")) if img_url: IMAGE_CACHE[cache_key] = img_url return img_url except Exception as e: print(f"Image fetch error: {e}") return None # -------------------- WEATHER FUNCTION -------------------- def get_weather(city: str) -> dict: """Fetch real-time weather data for any city.""" if not OPENWEATHER_API_KEY: return { "city": city, "temperature": 22, "condition": "sunny", "humidity": 60, "wind_speed": 10, "note": "Add OpenWeather API key for real data" } try: url = "https://api.openweathermap.org/data/2.5/weather" params = {'q': city, 'appid': OPENWEATHER_API_KEY, 'units': 'metric'} response = requests.get(url, params=params, timeout=10) data = response.json() if response.status_code != 200: return {"error": f"Weather API error: {data.get('message', 'unknown')}"} return { "city": city, "temperature": data['main']['temp'], "condition": data['weather'][0]['description'], "humidity": data['main']['humidity'], "wind_speed": data['wind']['speed'], "precipitation": data.get('rain', {}).get('1h', 0) } except Exception as e: return {"error": f"Weather error: {str(e)}"} # -------------------- REAL ATTRACTIONS FUNCTION -------------------- def get_real_attractions(city: str) -> list: """Get real attractions using OpenAI first for better quality.""" # Try OpenAI first for accurate attraction names and activities if client: try: prompt = f"""List the top 6 REAL tourist attractions in {city}. For each attraction, provide: - Name (actual famous attraction name) - A brief description including specific ACTIVITIES you can do there (use action words like: swimming, hiking, skating, climbing, boating, cycling, shopping, dining, photography, sightseeing, museum visiting, etc.) - Typical entry fee in USD (use numbers, 0 if free) - Approximate visit duration in hours Return ONLY a valid JSON array with keys: name, description, entry_fee, duration_hours. Example for Paris: [ {{ "name": "Eiffel Tower", "description": "Climb to the top for panoramic views, dine at the restaurant, and enjoy photography at sunset.", "entry_fee": 25, "duration_hours": 2.5 }} ] Important: Use REAL attractions specific to {city}. Make descriptions include ACTION WORDS like hiking, swimming, exploring, climbing.""" response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], temperature=0.7, max_tokens=1000 ) content = response.choices[0].message.content # Clean response content = content.strip() if content.startswith('```json'): content = content[7:] if content.startswith('```'): content = content[3:] if content.endswith('```'): content = content[:-3] attractions = json.loads(content) if isinstance(attractions, dict) and "attractions" in attractions: attractions = attractions["attractions"] elif not isinstance(attractions, list): attractions = [attractions] result = [] for a in attractions[:6]: fee = a.get("entry_fee", 0) if isinstance(fee, (int, float)): if fee == 0: fee_display = "Free" else: fee_display = f"${fee}" else: fee_display = str(fee) result.append({ "name": a.get("name", "Unknown"), "description": a.get("description", f"A must-visit attraction in {city}"), "entry_fee": fee_display, "duration_hours": a.get("duration_hours", 2) }) if result: return result except Exception as e: print(f"OpenAI attractions error: {e}") # Fallback to SerpAPI if SERP_API_KEY: try: params = { "q": f"top tourist attractions in {city}", "api_key": SERP_API_KEY, "engine": "google", "num": 6 } response = requests.get("https://serpapi.com/search", params=params, timeout=10) data = response.json() attractions = [] for result in data.get("organic_results", [])[:6]: title = result.get("title", "") if "wikipedia" not in title.lower() and "tripadvisor" not in title.lower(): name = title.split(" - ")[0].split(" | ")[0].split(":")[0].strip() if len(name) > 5 and len(name) < 100: attractions.append({ "name": name, "description": result.get("snippet", f"Explore this popular attraction in {city}"), "entry_fee": "Check website", "duration_hours": 2 }) if attractions and len(attractions) >= 3: return attractions[:6] except Exception as e: print(f"SerpAPI error: {e}") # Ultimate fallback return [ {"name": f"Historic {city} City Center", "description": f"Explore the historic heart of {city} with walking tours, shopping, and photography.", "entry_fee": "Free", "duration_hours": 2}, {"name": f"{city} Cultural Museum", "description": f"Discover art and history through guided tours and interactive exhibits.", "entry_fee": "$10", "duration_hours": 2}, {"name": f"{city} Central Park", "description": f"Enjoy hiking, picnicking, and outdoor activities in this beautiful green space.", "entry_fee": "Free", "duration_hours": 1.5} ] # -------------------- BUDGET CALCULATION -------------------- def calculate_budget(num_days: int, budget_amount: float = None) -> dict: """Calculate budget based on days and user input.""" if budget_amount: daily_budget = budget_amount / num_days if daily_budget < 100: level = "Budget" daily_rates = {"accommodation": 50, "food": 35, "transport": 15, "activities": 10} elif daily_budget < 200: level = "Moderate" daily_rates = {"accommodation": 100, "food": 60, "transport": 25, "activities": 20} elif daily_budget < 400: level = "Comfortable" daily_rates = {"accommodation": 180, "food": 100, "transport": 35, "activities": 35} else: level = "Luxury" daily_rates = {"accommodation": 300, "food": 150, "transport": 50, "activities": 60} else: level = "Moderate" daily_rates = {"accommodation": 100, "food": 60, "transport": 25, "activities": 20} accommodation = daily_rates["accommodation"] * num_days food = daily_rates["food"] * num_days transport = daily_rates["transport"] * num_days activities = daily_rates["activities"] * num_days total = accommodation + food + transport + activities return { "level": level, "accommodation": accommodation, "food": food, "transport": transport, "activities": activities, "total": total, "daily": daily_rates } # -------------------- ITINERARY GENERATION -------------------- def generate_itinerary(destination: str, start_date: str, num_days: int, budget_amount: float, budget_currency: str, departure_city: str = ""): """Main itinerary generation function.""" try: if not destination: return "❌ Please enter a destination city." if num_days < 1 or num_days > 14: return "❌ Number of days must be between 1 and 14." weather = get_weather(destination) if "error" in weather: return f"❌ Weather error: {weather['error']}" attractions = get_real_attractions(destination) destination_image = get_destination_image(destination) budget_data = calculate_budget(num_days, budget_amount) start = datetime.strptime(start_date, "%Y-%m-%d") end = start + timedelta(days=int(num_days) - 1) html = generate_html_itinerary( destination, weather, attractions, budget_data, num_days, start, end, budget_amount, departure_city, destination_image ) return html except Exception as e: return f"❌ An unexpected error occurred: {str(e)}" def generate_html_itinerary(destination, weather, attractions, budget_data, num_days, start_date, end_date, budget_amount, departure_city, destination_image): """Create beautiful HTML itinerary with max 3 attractions per day.""" weather_temp = f"{weather['temperature']:.1f}°C" if isinstance(weather['temperature'], (int, float)) else str(weather['temperature']) weather_condition = weather['condition'].capitalize() # Budget warning budget_warning = "" if budget_amount and budget_data['total'] > budget_amount: budget_warning = f"""
Intelligent Itinerary Generation • Real-time Data • Smart Recommendations
{num_days} Days of Adventure • {start_date.strftime('%B %d')} - {end_date.strftime('%B %d, %Y')}
{f"✈️ From: {departure_city}
" if departure_city else ""}🛏️ Search Hotels | ✈️ Search Flights | 🎟️ Read Reviews {currency_link}
🤖 Agentic AI System • Powered by OpenAI, OpenWeather & SerpAPI
Multi-Agent System • Intelligent Itinerary Generation • Real-time Data Processing