"""Google Calendar API client for managing events.""" from datetime import datetime, timedelta from typing import Dict, List, Optional from googleapiclient.discovery import build from googleapiclient.errors import HttpError from .oauth_manager import GoogleOAuthManager class CalendarClient: """Client for Google Calendar API operations.""" def __init__(self, oauth_manager: GoogleOAuthManager): """Initialize Calendar client. Args: oauth_manager: Initialized OAuth manager with valid credentials """ self.oauth_manager = oauth_manager self.service = None self._initialize_service() def _initialize_service(self) -> bool: """Initialize Calendar API service. Returns: True if initialized successfully, False otherwise """ credentials = self.oauth_manager.get_credentials() if not credentials: return False try: self.service = build("calendar", "v3", credentials=credentials) return True except Exception as e: print(f"[Calendar] Failed to initialize service: {e}") return False def list_events( self, days_ahead: int = 7, calendar_id: str = "primary", max_results: int = 20, ) -> Dict: """List upcoming events. Args: days_ahead: Number of days ahead to look (max: 30) calendar_id: Calendar ID (default: "primary") max_results: Maximum number of events to return Returns: Dict with success status and events or error """ if not self.service: if not self._initialize_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: # Limit days_ahead to 30 days_ahead = min(days_ahead, 30) now = datetime.utcnow() time_min = now.isoformat() + "Z" time_max = (now + timedelta(days=days_ahead)).isoformat() + "Z" events_result = ( self.service.events() .list( calendarId=calendar_id, timeMin=time_min, timeMax=time_max, maxResults=max_results, singleEvents=True, orderBy="startTime", ) .execute() ) events = events_result.get("items", []) if not events: return { "success": True, "events": [], "count": 0, "summary": f"No events in the next {days_ahead} days.", } # Format events formatted_events = [] for event in events: start = event["start"].get("dateTime", event["start"].get("date")) end = event["end"].get("dateTime", event["end"].get("date")) formatted_events.append({ "id": event["id"], "summary": event.get("summary", "No title"), "start": start, "end": end, "location": event.get("location", ""), "description": event.get("description", ""), "html_link": event.get("htmlLink", ""), }) summary = self._format_events_summary(formatted_events) return { "success": True, "events": formatted_events, "count": len(formatted_events), "summary": summary, } except HttpError as e: return {"success": False, "error": str(e)} def create_event( self, summary: str, start_time: str, end_time: str, description: str = "", location: str = "", calendar_id: str = "primary", ) -> Dict: """Create a new calendar event. Args: summary: Event title start_time: Start time (ISO 8601 format) end_time: End time (ISO 8601 format) description: Optional event description location: Optional event location calendar_id: Calendar ID (default: "primary") Returns: Dict with success status and event details or error """ if not self.service: if not self._initialize_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: # Detect if all-day event (time is 00:00:00) is_all_day = "T00:00:00" in start_time if is_all_day: # Use date format for all-day events event_body = { "summary": summary, "start": {"date": start_time.split("T")[0]}, "end": {"date": end_time.split("T")[0]}, } else: # Use dateTime format for timed events event_body = { "summary": summary, "start": {"dateTime": start_time, "timeZone": "UTC"}, "end": {"dateTime": end_time, "timeZone": "UTC"}, } if description: event_body["description"] = description if location: event_body["location"] = location event = ( self.service.events() .insert(calendarId=calendar_id, body=event_body) .execute() ) return { "success": True, "event_id": event.get("id"), "html_link": event.get("htmlLink"), "summary": event.get("summary"), "start": event["start"].get("dateTime", event["start"].get("date")), } except HttpError as e: return {"success": False, "error": str(e)} def search_events( self, query: str, calendar_id: str = "primary", ) -> Dict: """Search calendar events by text query. Args: query: Search query string calendar_id: Calendar ID (default: "primary") Returns: Dict with success status and matching events or error """ if not self.service: if not self._initialize_service(): return { "success": False, "error": "Not authorized. Run: python bot_runner.py --setup-google", } try: events_result = ( self.service.events() .list( calendarId=calendar_id, q=query, singleEvents=True, orderBy="startTime", ) .execute() ) events = events_result.get("items", []) if not events: return { "success": True, "events": [], "count": 0, "summary": f'No events found matching "{query}".', } # Format events formatted_events = [] for event in events: start = event["start"].get("dateTime", event["start"].get("date")) end = event["end"].get("dateTime", event["end"].get("date")) formatted_events.append({ "id": event["id"], "summary": event.get("summary", "No title"), "start": start, "end": end, "location": event.get("location", ""), "description": event.get("description", ""), }) summary = self._format_events_summary(formatted_events) return { "success": True, "events": formatted_events, "count": len(formatted_events), "summary": summary, } except HttpError as e: return {"success": False, "error": str(e)} def _format_events_summary(self, events: List[Dict]) -> str: """Format events into readable summary. Args: events: List of event dicts Returns: Formatted string summary """ if not events: return "No events." lines = [] for i, event in enumerate(events, 1): start = event["start"] # Parse datetime for better formatting try: if "T" in start: dt = datetime.fromisoformat(start.replace("Z", "+00:00")) start_str = dt.strftime("%b %d at %I:%M %p") else: dt = datetime.fromisoformat(start) start_str = dt.strftime("%b %d (all day)") except: start_str = start lines.append(f"{i}. {event['summary']} - {start_str}") if event.get("location"): lines.append(f" Location: {event['location']}") return "\n".join(lines)