"""
Voice API routes - Process voice commands and integrate with Google services
"""
from flask import Blueprint, request, jsonify
from flask_login import login_required, current_user
from app import db
from app.models import Instance, Conversation
from app.services.google_calendar import get_calendar_service
from app.services.gmail_service import get_gmail_service
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
voice_bp = Blueprint('voice', __name__, url_prefix='/api/voice')
@voice_bp.route('/process', methods=['POST'])
def process_voice_command():
"""Process voice command and return AI response"""
data = request.get_json()
instance_id = data.get('instance_id')
text = data.get('text')
if not instance_id or not text:
return jsonify({'error': 'Missing required parameters'}), 400
# Get instance
instance = Instance.query.filter_by(instance_id=instance_id).first()
if not instance:
return jsonify({'error': 'Instance not found'}), 404
try:
# Determine intent and process
result = process_command(instance, text)
# Save conversation
conversation = Conversation(
instance_id=instance.id,
session_id=data.get('session_id', 'default'),
role='user',
content=text
)
db.session.add(conversation)
if result.get('response'):
assistant_conv = Conversation(
instance_id=instance.id,
session_id=data.get('session_id', 'default'),
role='assistant',
content=result['response']
)
db.session.add(assistant_conv)
db.session.commit()
return jsonify(result)
except Exception as e:
logger.error(f"Error processing command: {e}")
return jsonify({'error': str(e)}), 500
def process_command(instance: Instance, text: str) -> dict:
"""Process voice command and route to appropriate service"""
text_lower = text.lower()
# Calendar commands
if instance.enable_google_calendar and any(word in text_lower for word in
['calendar', 'schedule', 'event', 'meeting', 'appointment']):
return process_calendar_command(instance, text)
# Email commands
elif instance.enable_gmail and any(word in text_lower for word in
['email', 'mail', 'message', 'send', 'inbox']):
return process_email_command(instance, text)
# Default: AI conversation
else:
return process_ai_conversation(instance, text)
def process_calendar_command(instance: Instance, text: str) -> dict:
"""Process calendar-related commands"""
calendar_service = get_calendar_service(instance)
if not calendar_service:
return {'response': 'Google Calendar is not enabled for this instance.'}
text_lower = text.lower()
# Check events
if any(word in text_lower for word in ['what', 'show', 'list', 'do i have', 'my schedule']):
events_result = calendar_service.get_events(max_results=10)
if 'error' in events_result:
return {'response': 'Failed to access your calendar. Please check Google authorization.'}
events = events_result.get('items', [])
summary = calendar_service.format_events_summary(events)
return {
'response': summary,
'used_calendar': True
}
# Create event
elif any(word in text_lower for word in ['create', 'schedule', 'add', 'set up', 'make']):
# Parse event details
parsed = calendar_service.parse_natural_datetime(text)
if not parsed:
return {'response': 'I couldn\'t understand when you want to schedule this. Please specify a time.'}
# Extract event title
title = extract_event_title(text)
if not title:
title = 'New Event'
event_result = calendar_service.create_event(
title=title,
start_time=parsed['start'],
end_time=parsed['end']
)
if 'error' in event_result:
return {'response': 'Failed to create event. Please try again.'}
return {
'response': f'Created event "{title}" successfully.',
'used_calendar': True
}
return {'response': 'I can help you check your calendar or create events. What would you like?'}
def process_email_command(instance: Instance, text: str) -> dict:
"""Process email-related commands"""
gmail_service = get_gmail_service(instance)
if not gmail_service:
return {'response': 'Gmail is not enabled for this instance.'}
text_lower = text.lower()
# Check emails
if any(word in text_lower for word in ['check', 'read', 'show', 'inbox', 'emails', 'messages']):
emails = gmail_service.get_inbox_summary(max_results=5)
if not emails:
return {'response': 'You have no new emails.', 'used_gmail': True}
summary = gmail_service.format_email_summary(emails)
unread_count = gmail_service.get_unread_count()
response = f'You have {unread_count} unread emails. {summary}'
return {
'response': response,
'used_gmail': True
}
# Send email
elif any(word in text_lower for word in ['send', 'compose', 'write', 'email to']):
# Parse email details
recipient = extract_email_recipient(text)
subject = extract_email_subject(text)
body = extract_email_body(text)
if not recipient:
return {'response': 'Who would you like to send this email to?'}
# If details missing, ask user
if not subject:
subject = 'Message from Wakelai'
if not body:
body = text # Use original text as body
result = gmail_service.send_message(
to=recipient,
subject=subject,
body=body
)
if 'error' in result:
return {'response': 'Failed to send email. Please check the recipient address.'}
return {
'response': f'Email sent to {recipient}.',
'used_gmail': True
}
return {'response': 'I can help you check your emails or send messages. What would you like?'}
def process_ai_conversation(instance: Instance, text: str) -> dict:
"""Process general AI conversation"""
# TODO: Integrate with LLM (Claude, etc.)
# For now, return a placeholder response
responses = [
"I understand. How can I help you further?",
"That's interesting. Tell me more.",
"I can help with that. What would you like me to do?",
"I'm processing your request."
]
import random
response = random.choice(responses)
return {
'response': response,
'canvas_html': generate_canvas_response(text)
}
def generate_canvas_response(text: str) -> str:
"""Generate HTML canvas response based on text"""
text_lower = text.lower()
# Dashboard request
if any(word in text_lower for word in ['dashboard', 'overview', 'stats']):
return '''
'''
# Report request
elif any(word in text_lower for word in ['report', 'summary']):
return '''
Weekly Report
Period: Last 7 days
- Total interactions: 156
- Emails processed: 23
- Calendar events managed: 8
- Tasks completed: 12
'''
return None
# Helper functions for parsing
def extract_event_title(text: str) -> str:
"""Extract event title from text"""
# Simple extraction - looks for patterns like "meeting with John" or "lunch"
words = text.lower().split()
stop_words = {'create', 'schedule', 'add', 'set', 'up', 'a', 'an', 'the', 'for', 'on', 'at', 'with'}
title_words = [w for w in words if w not in stop_words and w.isalpha()]
if title_words:
return ' '.join(title_words[:3]).title()
return None
def extract_email_recipient(text: str) -> str:
"""Extract email recipient from text"""
import re
# Look for email pattern
email_match = re.search(r'[\w\.-]+@[\w\.-]+', text)
if email_match:
return email_match.group()
# Look for "to [person]" pattern
to_match = re.search(r'to (\w+)', text.lower())
if to_match:
return f"{to_match.group(1)}@example.com"
return None
def extract_email_subject(text: str) -> str:
"""Extract email subject from text"""
import re
# Look for "about [subject]" or "subject: [text]"
about_match = re.search(r'about (.+?)(?:\.|$|and)', text, re.IGNORECASE)
if about_match:
return about_match.group(1).strip()
subject_match = re.search(r'subject[:\s]+(.+?)(?:\.|$)', text, re.IGNORECASE)
if subject_match:
return subject_match.group(1).strip()
return None
def extract_email_body(text: str) -> str:
"""Extract email body from text"""
# For now, return the full text as body
# In production, this would use LLM to extract actual message content
return text