Construire un planificateur de voyage propulsé par l'IA

import spacy
from spacy import displacy
import re
 
class TravelIntentClassifier:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.intent_patterns = {
            'destination_inquiry': [
                r'where.*go', r'best.*place', r'recommend.*destination'
            ],
            'activity_request': [
                r'what.*do', r'activities', r'attractions'
            ],
            'budget_planning': [
                r'budget', r'cost', r'expensive', r'cheap'
            ],
            'accommodation': [
                r'hotel', r'stay', r'accommodation', r'booking'
            ]
        }
    
    def classify_intent(self, text):
        doc = self.nlp(text)
        intent_scores = {}
        
        for intent, patterns in self.intent_patterns.items():
            score = 0
            for pattern in patterns:
                if re.search(pattern, text.lower()):
                    score += 1
            intent_scores[intent] = score
        
        return max(intent_scores, key=intent_scores.get)
    
    def extract_entities(self, text):
        doc = self.nlp(text)
        entities = {
            'destinations': [],
            'dates': [],
            'activities': [],
            'budget': None
        }
        
        for ent in doc.ents:
            if ent.label_ == "GPE":  # Geopolitical entity
                entities['destinations'].append(ent.text)
            elif ent.label_ == "DATE":
                entities['dates'].append(ent.text)
        
        return entities

import openai
from datetime import datetime, timedelta
import json
 
class ItineraryGenerator:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.weather_api = WeatherAPI()
        self.events_api = EventsAPI()
        self.attractions_api = AttractionsAPI()
    
    def generate_itinerary(self, user_preferences):
        prompt = self._build_prompt(user_preferences)
        
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are an expert travel planner with deep knowledge of destinations worldwide."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=1500
        )
        
        return self._parse_itinerary(response.choices[0].message.content)
    
    def _build_prompt(self, preferences):
        return f"""
        Create a detailed travel itinerary for:
        - Destination: {preferences['destination']}
        - Duration: {preferences['duration']} days
        - Budget: {preferences['budget']}
        - Interests: {', '.join(preferences['interests'])}
        - Travel style: {preferences['travel_style']}
        
        Include:
        1. Daily schedule with time slots
        2. Specific attractions and activities
        3. Restaurant recommendations
        4. Transportation options
        5. Budget breakdown
        6. Tips and local insights
        """
    
    def _parse_itinerary(self, ai_response):
        # Parse AI response into structured format
        days = ai_response.split('Day')
        itinerary = []
        
        for day_content in days[1:]:  # Skip first empty split
            day_data = self._extract_day_data(day_content)
            itinerary.append(day_data)
        
        return itinerary

import requests
from datetime import datetime
 
class TravelDataIntegrator:
    def __init__(self):
        self.weather_api_key = "your_weather_api_key"
        self.events_api_key = "your_events_api_key"
        self.booking_api_key = "your_booking_api_key"
    
    def get_weather_forecast(self, destination, dates):
        url = f"http://api.weatherapi.com/v1/forecast.json"
        params = {
            'key': self.weather_api_key,
            'q': destination,
            'days': len(dates)
        }
        
        response = requests.get(url, params=params)
        return response.json()
    
    def get_local_events(self, destination, dates):
        url = f"https://api.eventful.com/json/events/search"
        params = {
            'app_key': self.events_api_key,
            'location': destination,
            'date': f"{dates[0]},{dates[-1]}"
        }
        
        response = requests.get(url, params=params)
        return response.json()
    
    def get_attraction_availability(self, attraction_id, date):
        # Integration with attraction booking systems
        url = f"https://api.attractions.com/availability"
        params = {
            'attraction_id': attraction_id,
            'date': date
        }
        
        response = requests.get(url, params=params)
        return response.json()

from flask import Flask, request, jsonify, render_template
import json
 
app = Flask(__name__)
 
class TravelChatbot:
    def __init__(self):
        self.intent_classifier = TravelIntentClassifier()
        self.itinerary_generator = ItineraryGenerator(api_key="your_openai_key")
        self.data_integrator = TravelDataIntegrator()
        self.user_sessions = {}
    
    def process_message(self, user_id, message):
        # Classify user intent
        intent = self.intent_classifier.classify_intent(message)
        entities = self.intent_classifier.extract_entities(message)
        
        # Update user session
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = {
                'preferences': {},
                'conversation_history': []
            }
        
        session = self.user_sessions[user_id]
        session['conversation_history'].append({
            'user': message,
            'timestamp': datetime.now().isoformat()
        })
        
        # Generate response based on intent
        if intent == 'destination_inquiry':
            response = self._handle_destination_inquiry(entities, session)
        elif intent == 'activity_request':
            response = self._handle_activity_request(entities, session)
        elif intent == 'budget_planning':
            response = self._handle_budget_planning(entities, session)
        else:
            response = self._handle_general_inquiry(message, session)
        
        return response
    
    def _handle_destination_inquiry(self, entities, session):
        if entities['destinations']:
            destination = entities['destinations'][0]
            # Get real-time data
            weather = self.data_integrator.get_weather_forecast(destination, [])
            events = self.data_integrator.get_local_events(destination, [])
            
            response = f"""
            Great choice! {destination} is an amazing destination. 
            
            Current weather: {weather['current']['condition']['text']}
            Temperature: {weather['current']['temp_c']}°C
            
            Popular events happening:
            {self._format_events(events)}
            
            What type of activities are you interested in?
            """
            
            return response
        else:
            return "I'd love to help you find the perfect destination! Where are you thinking of traveling?"
    
    def _format_events(self, events_data):
        if 'events' in events_data and events_data['events']:
            event_list = []
            for event in events_data['events']['event'][:3]:  # Top 3 events
                event_list.append(f"- {event['title']} ({event['start_time']})")
            return '\n'.join(event_list)
        return "No major events scheduled for this period."

# Caching implementation
from functools import lru_cache
import redis
 
class CacheManager:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
    
    @lru_cache(maxsize=1000)
    def get_cached_itinerary(self, destination, duration, budget, interests):
        cache_key = f"itinerary:{destination}:{duration}:{budget}:{hash(tuple(interests))}"
        cached = self.redis_client.get(cache_key)
        
        if cached:
            return json.loads(cached)
        
        # Generate new itinerary
        itinerary = self.generate_itinerary(destination, duration, budget, interests)
        self.redis_client.setex(cache_key, 3600, json.dumps(itinerary))  # 1 hour cache
        
        return itinerary