""" Google Calendar Integration Service Handles voice-controlled calendar operations """ import requests from datetime import datetime, timedelta from typing import List, Dict, Optional import logging logger = logging.getLogger(__name__) class GoogleCalendarService: """Service for Google Calendar API operations""" API_BASE = "https://www.googleapis.com/calendar/v3" def __init__(self, access_token: str, refresh_token: str = None): self.access_token = access_token self.refresh_token = refresh_token def _get_headers(self) -> Dict: """Get headers with authentication""" return { 'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json' } def get_events(self, calendar_id: str = 'primary', time_min: str = None, time_max: str = None, max_results: int = 10) -> Dict: """Get calendar events""" # Default time range: next 7 days if not time_min: time_min = datetime.utcnow().isoformat() + 'Z' if not time_max: time_max = (datetime.utcnow() + timedelta(days=7)).isoformat() + 'Z' params = { 'timeMin': time_min, 'timeMax': time_max, 'maxResults': max_results, 'singleEvents': True, 'orderBy': 'startTime' } try: response = requests.get( f"{self.API_BASE}/calendars/{calendar_id}/events", headers=self._get_headers(), params=params ) if response.status_code == 200: return response.json() elif response.status_code == 401: # Token expired, need refresh logger.warning("Access token expired") return {'error': 'token_expired'} else: logger.error(f"Calendar API error: {response.status_code}") return {'error': response.text} except Exception as e: logger.error(f"Failed to get events: {e}") return {'error': str(e)} def create_event(self, calendar_id: str = 'primary', title: str = None, description: str = None, start_time: str = None, end_time: str = None, location: str = None, attendees: List[str] = None) -> Dict: """Create a new calendar event""" event_data = { 'summary': title or 'New Event', 'description': description or '', 'start': {}, 'end': {} } # Handle time format if 'T' in start_time: # DateTime format event_data['start']['dateTime'] = start_time event_data['end']['dateTime'] = end_time or start_time else: # All-day event event_data['start']['date'] = start_time event_data['end']['date'] = end_time or start_time if location: event_data['location'] = location if attendees: event_data['attendees'] = [{'email': email} for email in attendees] try: response = requests.post( f"{self.API_BASE}/calendars/{calendar_id}/events", headers=self._get_headers(), json=event_data ) if response.status_code in [200, 201]: return response.json() else: logger.error(f"Create event error: {response.status_code}") return {'error': response.text} except Exception as e: logger.error(f"Failed to create event: {e}") return {'error': str(e)} def update_event(self, calendar_id: str = 'primary', event_id: str = None, **updates) -> Dict: """Update an existing event""" try: # Get current event data first get_response = requests.get( f"{self.API_BASE}/calendars/{calendar_id}/events/{event_id}", headers=self._get_headers() ) if get_response.status_code != 200: return {'error': 'Event not found'} event_data = get_response.json() # Update fields for key, value in updates.items(): if value is not None: event_data[key] = value # Send update response = requests.put( f"{self.API_BASE}/calendars/{calendar_id}/events/{event_id}", headers=self._get_headers(), json=event_data ) if response.status_code == 200: return response.json() else: return {'error': response.text} except Exception as e: logger.error(f"Failed to update event: {e}") return {'error': str(e)} def delete_event(self, calendar_id: str = 'primary', event_id: str = None) -> Dict: """Delete an event""" try: response = requests.delete( f"{self.API_BASE}/calendars/{calendar_id}/events/{event_id}", headers=self._get_headers() ) if response.status_code == 204: return {'success': True} else: return {'error': response.text} except Exception as e: logger.error(f"Failed to delete event: {e}") return {'error': str(e)} def parse_natural_datetime(self, text: str) -> Optional[Dict]: """Parse natural language datetime into structured format Examples: - "tomorrow at 3pm" -> datetime object - "next Monday" -> datetime object - "June 15th" -> datetime object """ # Simple implementation - can be enhanced with NLP from datetime import datetime import re now = datetime.utcnow() text_lower = text.lower() # Tomorrow if 'tomorrow' in text_lower: target_date = now + timedelta(days=1) time_match = re.search(r'at (\d+)(?::(\d+))?\s*(am|pm)?', text_lower) if time_match: hour = int(time_match.group(1)) minute = int(time_match.group(2)) if time_match.group(2) else 0 if time_match.group(3) == 'pm' and hour < 12: hour += 12 return { 'start': target_date.replace(hour=hour, minute=minute, second=0).isoformat() + 'Z', 'end': (target_date.replace(hour=hour + 1, minute=minute, second=0)).isoformat() + 'Z' } return { 'start': target_date.replace(hour=9, minute=0, second=0).isoformat() + 'Z', 'end': (target_date.replace(hour=10, minute=0, second=0)).isoformat() + 'Z' } # Next week/day if 'next' in text_lower: days_map = { 'monday': 0, 'tuesday': 1, 'wednesday': 2, 'thursday': 3, 'friday': 4, 'saturday': 5, 'sunday': 6 } for day_name, day_num in days_map.items(): if day_name in text_lower: today = now.weekday() days_until = (day_num - today + 7) % 7 if days_until == 0: days_until = 7 target_date = now + timedelta(days=days_until) return { 'start': target_date.replace(hour=9, minute=0, second=0).isoformat() + 'Z', 'end': (target_date.replace(hour=10, minute=0, second=0)).isoformat() + 'Z' } return None def format_events_summary(self, events: List[Dict]) -> str: """Format events into natural language summary""" if not events: return "You have no upcoming events." summary = [] for event in events[:5]: title = event.get('summary', 'No title') start = event.get('start', {}) if 'dateTime' in start: time_str = start['dateTime'] elif 'date' in start: time_str = start['date'] else: time_str = 'Unknown time' summary.append(f"{title} at {time_str}") return "You have " + ', '.join(summary) + "." def get_calendar_service(instance) -> GoogleCalendarService: """Get calendar service for instance""" if not instance.google_access_token: return None return GoogleCalendarService( access_token=instance.google_access_token, refresh_token=instance.google_refresh_token )