297 lines
9.2 KiB
Python
297 lines
9.2 KiB
Python
|
"""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)
|