""" Appointment scheduling router """ import uuid import json import os from datetime import datetime from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks from typing import Dict, Any, List, Optional from models.appointment import AppointmentRequest, AppointmentResponse, MeetingIntentDetectionResponse from core.dependencies import get_logger, get_conversation_handler router = APIRouter(prefix="/appointment", tags=["appointment"]) # In-memory storage for appointments (in a production app, this would be a database) appointments = {} # Path to store appointments data APPOINTMENTS_FILE = "data/appointments.json" # Ensure the data directory exists os.makedirs(os.path.dirname(APPOINTMENTS_FILE), exist_ok=True) # Load existing appointments if the file exists if os.path.exists(APPOINTMENTS_FILE): try: with open(APPOINTMENTS_FILE, "r") as f: appointments = json.load(f) except Exception: appointments = {} def save_appointments(): """Save appointments to file""" with open(APPOINTMENTS_FILE, "w") as f: json.dump(appointments, f, indent=2) @router.post("/schedule", response_model=AppointmentResponse) async def schedule_appointment( request: AppointmentRequest, background_tasks: BackgroundTasks, logger=Depends(get_logger), conversation_handler=Depends(get_conversation_handler) ): """ Schedule an appointment with a senior consultant This endpoint: 1. Accepts appointment details 2. Creates an appointment record 3. Returns confirmation details """ try: logger.info(f"Received appointment request from {request.name} ({request.email})") # Generate a unique appointment ID appointment_id = str(uuid.uuid4()) # Create appointment record appointment = { "appointment_id": appointment_id, "name": request.name, "email": request.email, "phone": request.phone, "company": request.company, "topic": request.topic, "preferred_date": request.preferred_date, "preferred_time": request.preferred_time, "additional_notes": request.additional_notes, "conversation_id": request.conversation_id, "status": "pending", "created_at": str(datetime.now()) } # Store the appointment appointments[appointment_id] = appointment # Save appointments in the background background_tasks.add_task(save_appointments) # Log the appointment but don't add a message to the conversation # The message will be handled by the enhanced_response_generator if request.conversation_id: try: # Just log the appointment for debugging logger.info(f"Appointment {appointment_id} scheduled for conversation {request.conversation_id}") # We're not adding a message here anymore to avoid duplicates # The enhanced_response_generator will handle the response except Exception as e: logger.error(f"Error logging appointment: {str(e)}") # Return the response return AppointmentResponse( appointment_id=appointment_id, status="pending", message="Your appointment request has been received. Our team will contact you soon to confirm the details.", contact_person="Senior Consultant Team" ) except Exception as e: logger.error(f"Error scheduling appointment: {str(e)}") raise HTTPException(status_code=500, detail=f"Error scheduling appointment: {str(e)}") @router.get("/detect-intent", response_model=MeetingIntentDetectionResponse) async def detect_meeting_intent( message: str, logger=Depends(get_logger) ): """ Detect if a message contains a meeting request intent This endpoint: 1. Analyzes the message for meeting request intent 2. Returns detection results and confidence score """ try: # Fast path for simple greetings to avoid unnecessary processing message_lower = message.lower().strip() simple_greetings = ["hi", "hello", "hey", "good morning", "good afternoon", "good evening", "hi there", "hello there"] if any(message_lower == greeting or message_lower.startswith(greeting + " ") for greeting in simple_greetings): logger.info("Detected simple greeting, not a meeting request") return MeetingIntentDetectionResponse( is_meeting_request=False, confidence=0.0, suggested_topic=None, missing_fields=[], extracted_info={} ) # Enhanced keywords that might indicate a meeting request meeting_keywords = [ # Direct meeting terms "meeting", "appointment", "schedule", "book", "consult", "consultation", "discuss", "talk", "call", "meet", # Business terms "pricing", "quote", "estimate", "cost", "price", "proposal", "project", "service", "hire", "engage", # Time-related terms "available", "availability", "time", "slot", "calendar", "tomorrow", "next week", "monday", "tuesday", "wednesday", "thursday", "friday", "weekend", "morning", "afternoon", "evening", # Intent phrases "would like to", "want to", "interested in", "looking for", "need to", "can we", "possible to" ] # Phrase patterns that strongly indicate meeting intent meeting_phrases = [ "set up a meeting", "schedule a call", "book an appointment", "arrange a consultation", "get in touch", "speak with", "talk to someone", "discuss my project", "discuss my requirements", "get a quote", "learn more about your services", "would like to meet", "want to meet", "need consultation", "interested in your services", "when are you available", "what's your availability", "can we meet", "can we talk", "can we discuss" ] # Check for meeting intent in the message message_lower = message.lower() # Check for keyword matches found_keywords = [keyword for keyword in meeting_keywords if keyword in message_lower] # Check for phrase matches (stronger indicators) found_phrases = [phrase for phrase in meeting_phrases if phrase in message_lower] # Calculate confidence based on keywords and phrases found # Phrases are stronger indicators so they get higher weight keyword_score = min(len(found_keywords) * 0.15, 0.6) phrase_score = min(len(found_phrases) * 0.3, 0.9) confidence = max(keyword_score, phrase_score) # Boost confidence if multiple indicators are present if keyword_score > 0 and phrase_score > 0: confidence = min(confidence + 0.1, 0.95) # Boost confidence if message contains date/time information date_time_indicators = ["monday", "tuesday", "wednesday", "thursday", "friday", "weekend", "tomorrow", "next week", "morning", "afternoon", "evening", "pm", "am", ":00", "o'clock"] if any(indicator in message_lower for indicator in date_time_indicators): confidence = min(confidence + 0.1, 0.95) # Determine if this is likely a meeting request is_meeting_request = confidence > 0.25 # Lower threshold for better recall # Determine which fields might be missing missing_fields = [] extracted_info = {} suggested_topic = None if is_meeting_request: # Check for name name_indicators = ["name is", "i am", "i'm", "my name", "this is"] name_found = False for indicator in name_indicators: if indicator in message_lower: parts = message.split(indicator, 1) if len(parts) > 1: name_text = parts[1].strip() # Extract up to the first punctuation or line break for end_char in ['.', ',', '!', '?', '\n']: if end_char in name_text: name_text = name_text.split(end_char, 1)[0] extracted_info["name"] = name_text.strip() name_found = True break if not name_found: missing_fields.append("name") # Check for email import re email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' email_matches = re.findall(email_pattern, message) if email_matches: extracted_info["email"] = email_matches[0] else: missing_fields.append("email") # Check for phone number phone_pattern = r'\b(?:\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b' phone_matches = re.findall(phone_pattern, message) if phone_matches: extracted_info["phone"] = phone_matches[0] else: missing_fields.append("phone") # Check for date/time preferences date_pattern = r'\b(?:(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{1,2}(?:st|nd|rd|th)?|\d{1,2}(?:st|nd|rd|th)?\s+(?:of\s+)?(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*|tomorrow|next\s+(?:mon|tues|wednes|thurs|fri|satur|sun)day|\d{1,2}[/.-]\d{1,2}(?:[/.-]\d{2,4})?)\b' date_matches = re.findall(date_pattern, message_lower, re.IGNORECASE) if date_matches: extracted_info["preferred_date"] = date_matches[0] time_pattern = r'\b(?:(?:1[0-2]|0?[1-9])(?::[0-5][0-9])?\s*(?:am|pm)|(?:1[0-2]|0?[1-9])\s*(?:o\'clock)\s*(?:am|pm)?|(?:[01]?[0-9]|2[0-3]):[0-5][0-9])\b' time_matches = re.findall(time_pattern, message_lower, re.IGNORECASE) if time_matches: extracted_info["preferred_time"] = time_matches[0] # Try to extract a topic if it's a meeting request topic_indicators = ["about", "regarding", "discuss", "talk about", "meeting about", "consultation for", "interested in", "need help with", "looking for"] suggested_topic = None for indicator in topic_indicators: if indicator in message_lower: # Extract text after the indicator parts = message.split(indicator, 1) if len(parts) > 1: # Take up to 100 characters after the indicator topic_text = parts[1].strip()[:100] # Cut at the first sentence ending for end_char in ['.', '!', '?', '\n']: if end_char in topic_text: topic_text = topic_text.split(end_char, 1)[0] suggested_topic = topic_text.strip() extracted_info["topic"] = suggested_topic break # If no specific topic was found but it's a meeting request if not suggested_topic and is_meeting_request: # Use a generic topic based on keywords found if any(kw in message_lower for kw in ["website", "web", "site"]): suggested_topic = "Website Development" extracted_info["topic"] = suggested_topic elif any(kw in message_lower for kw in ["app", "mobile", "android", "ios"]): suggested_topic = "Mobile App Development" extracted_info["topic"] = suggested_topic elif any(kw in message_lower for kw in ["ai", "machine learning", "ml", "artificial intelligence"]): suggested_topic = "AI/ML Development" extracted_info["topic"] = suggested_topic else: suggested_topic = "General Consultation" extracted_info["topic"] = suggested_topic if "topic" not in extracted_info: missing_fields.append("topic") return MeetingIntentDetectionResponse( is_meeting_request=is_meeting_request, confidence=confidence, suggested_topic=suggested_topic if suggested_topic else extracted_info.get("topic"), missing_fields=missing_fields, extracted_info=extracted_info ) except Exception as e: logger.error(f"Error detecting meeting intent: {str(e)}") raise HTTPException(status_code=500, detail=f"Error detecting meeting intent: {str(e)}") @router.get("/list", response_model=List[AppointmentResponse]) async def list_appointments( logger=Depends(get_logger) ): """ List all appointments """ try: return [ AppointmentResponse( appointment_id=appointment_id, status=appointment["status"], message="Appointment details", scheduled_date=appointment.get("preferred_date"), scheduled_time=appointment.get("preferred_time"), contact_person="Senior Consultant Team" ) for appointment_id, appointment in appointments.items() ] except Exception as e: logger.error(f"Error listing appointments: {str(e)}") raise HTTPException(status_code=500, detail=f"Error listing appointments: {str(e)}") @router.get("/{appointment_id}", response_model=AppointmentResponse) async def get_appointment( appointment_id: str, logger=Depends(get_logger) ): """ Get details of a specific appointment """ try: if appointment_id not in appointments: raise HTTPException(status_code=404, detail=f"Appointment {appointment_id} not found") appointment = appointments[appointment_id] return AppointmentResponse( appointment_id=appointment_id, status=appointment["status"], message="Appointment details", scheduled_date=appointment.get("preferred_date"), scheduled_time=appointment.get("preferred_time"), contact_person="Senior Consultant Team" ) except HTTPException: raise except Exception as e: logger.error(f"Error getting appointment: {str(e)}") raise HTTPException(status_code=500, detail=f"Error getting appointment: {str(e)}")