Spaces:
Sleeping
Sleeping
| import os | |
| import dateparser | |
| from amadeus import Client, ResponseError | |
| from langchain.tools import Tool | |
| # Initialize Amadeus API Client | |
| AMADEUS_API_KEY = os.getenv("AMADEUS_API_KEY") | |
| AMADEUS_API_SECRET = os.getenv("AMADEUS_API_SECRET") | |
| amadeus = Client( | |
| client_id=AMADEUS_API_KEY, | |
| client_secret=AMADEUS_API_SECRET | |
| ) | |
| # π Convert City Name to IATA Code | |
| def get_airport_code(city_name: str): | |
| """Find the IATA airport code for a given city.""" | |
| try: | |
| response = amadeus.reference_data.locations.get( | |
| keyword=city_name, | |
| subType="AIRPORT,CITY" | |
| ) | |
| if response.data: | |
| return response.data[0]["iataCode"] | |
| return None | |
| except ResponseError: | |
| return None | |
| # βοΈ Get Full Airline Name from Code | |
| def get_airline_name(airline_code: str): | |
| """Get full airline name using Amadeus API.""" | |
| try: | |
| response = amadeus.reference_data.airlines.get(airlineCodes=airline_code) | |
| if response.data: | |
| return response.data[0]["businessName"] | |
| return airline_code # Fallback to code if not found | |
| except ResponseError: | |
| return airline_code | |
| # π Format Flight Duration | |
| def format_duration(duration: str): | |
| """Convert ISO 8601 duration (PT20H25M) into readable format (20h 25m).""" | |
| duration = duration.replace("PT", "").replace("H", "h ").replace("M", "m") | |
| return duration.strip() | |
| # π Generate Booking Link | |
| def generate_booking_link(from_iata: str, to_iata: str, departure_date: str): | |
| """Generate a booking link using Google Flights.""" | |
| return f"https://www.google.com/flights?hl=en#flt={from_iata}.{to_iata}.{departure_date};c:USD;e:1;s:0;sd:1;t:f" | |
| # π« Flight Search Tool (Now Gradio-Compatible with Simple Ticket Style) | |
| def search_flights(query: str): | |
| """Search for flights using Amadeus API and return plain Markdown output for Gradio.""" | |
| try: | |
| words = query.lower().split() | |
| from_city, to_city, date_phrase = None, None, None | |
| # Extract "from", "to", and date information | |
| if "from" in words and "to" in words: | |
| from_index = words.index("from") + 1 | |
| to_index = words.index("to") + 1 | |
| from_city = " ".join(words[from_index:to_index - 1]).title() | |
| to_city = " ".join(words[to_index:words.index("in")]) if "in" in words else " ".join(words[to_index:]).title() | |
| date_phrase = " ".join(words[words.index("in") + 1:]) if "in" in words else None | |
| # Validate extracted details | |
| if not from_city or not to_city: | |
| return "β Could not detect valid departure and destination cities. Please use 'from <city> to <city>'." | |
| # Convert city names to IATA codes | |
| from_iata = get_airport_code(from_city) | |
| to_iata = get_airport_code(to_city) | |
| if not from_iata or not to_iata: | |
| return f"β Could not find airport codes for {from_city} or {to_city}. Please check spelling." | |
| # Convert date phrase to YYYY-MM-DD | |
| departure_date = dateparser.parse(date_phrase) if date_phrase else None | |
| if not departure_date: | |
| return "β Could not understand the travel date. Use formats like 'next week' or 'on May 15'." | |
| departure_date_str = departure_date.strftime("%Y-%m-%d") | |
| # Fetch flight offers from Amadeus | |
| response = amadeus.shopping.flight_offers_search.get( | |
| originLocationCode=from_iata, | |
| destinationLocationCode=to_iata, | |
| departureDate=departure_date_str, | |
| adults=1, | |
| max=5 | |
| ) | |
| flights = response.data | |
| if not flights: | |
| return f"β No flights found from {from_city} to {to_city} on {departure_date_str}." | |
| # π PLAIN TICKET OUTPUT FOR GRADIO MARKDOWN | |
| result = f"### Flight Information from {from_city} ({from_iata}) to {to_city} ({to_iata})\n" | |
| for flight in flights[:3]: # Show up to 5 results for simplicity | |
| airline_code = flight["validatingAirlineCodes"][0] | |
| airline_name = get_airline_name(airline_code) | |
| price = flight["price"]["total"] | |
| duration = format_duration(flight["itineraries"][0]["duration"]) | |
| departure_time = flight["itineraries"][0]["segments"][0]["departure"]["at"] | |
| arrival_time = flight["itineraries"][0]["segments"][-1]["arrival"]["at"] | |
| stops = len(flight["itineraries"][0]["segments"]) - 1 | |
| booking_link = generate_booking_link(from_iata, to_iata, departure_date_str) | |
| # Plain ticket format without fancy styling | |
| result += f""" | |
| - **Airline:** {airline_name} | |
| - **Flight No:** {airline_code}123 | |
| - **Departure Date and Time:** {departure_time} ({from_iata}) | |
| - **Arrival Date and Time:** {arrival_time} ({to_iata}) | |
| - **Duration:** {duration} | |
| - **Stops:** {stops} | |
| - **Price:** ${price} | |
| - [Book Now]({booking_link}) | [Show More Flights]({booking_link}) | |
| --- | |
| """ | |
| print(f"Debug - Flight Data: {flights}") # Debugging output to verify data | |
| return result | |
| except ResponseError as error: | |
| print(f"Debug - Amadeus API Error: {str(error)}") # Debugging output for API errors | |
| return f"β Error: {str(error)}" | |
| # β Register as a Tool | |
| tools = [ | |
| Tool( | |
| name="Flight Booking", | |
| func=search_flights, | |
| description="Find and book flights using natural language. Examples: 'Flight from Delhi to SFO in May', 'Travel from Mumbai to New York next week'." | |
| ) | |
| ] |