daa4de8832
Add comprehensive morning report skill with collectors for calendar, email, tasks, infrastructure status, news, stocks, and weather. Add stock lookup skill for quote queries.
138 lines
5.0 KiB
Python
Executable File
138 lines
5.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Calendar collector using existing gcal skill."""
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
|
|
def fetch_events(mode: str = "today") -> list:
|
|
"""Fetch calendar events directly using gmail_mcp library."""
|
|
os.environ.setdefault('GMAIL_CREDENTIALS_PATH', os.path.expanduser('~/.gmail-mcp/credentials.json'))
|
|
|
|
try:
|
|
# Add gmail venv to path
|
|
venv_site = Path.home() / ".claude/mcp/gmail/venv/lib/python3.13/site-packages"
|
|
if str(venv_site) not in sys.path:
|
|
sys.path.insert(0, str(venv_site))
|
|
|
|
from gmail_mcp.utils.GCP.gmail_auth import get_calendar_service
|
|
|
|
service = get_calendar_service()
|
|
now = datetime.utcnow()
|
|
|
|
if mode == 'today':
|
|
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end = start + timedelta(days=1)
|
|
elif mode == 'tomorrow':
|
|
start = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end = start + timedelta(days=1)
|
|
else:
|
|
start = now
|
|
end = now + timedelta(days=7)
|
|
|
|
events_result = service.events().list(
|
|
calendarId='primary',
|
|
timeMin=start.isoformat() + 'Z',
|
|
timeMax=end.isoformat() + 'Z',
|
|
singleEvents=True,
|
|
orderBy='startTime',
|
|
maxResults=20
|
|
).execute()
|
|
|
|
return events_result.get('items', [])
|
|
|
|
except Exception as e:
|
|
return [{"error": str(e)}]
|
|
|
|
|
|
def format_events(today_events: list, tomorrow_events: list = None) -> str:
|
|
"""Format calendar events - no LLM needed, structured data."""
|
|
lines = []
|
|
|
|
# Today's events
|
|
if today_events and (len(today_events) == 0 or "error" not in today_events[0]):
|
|
if not today_events:
|
|
lines.append("No events today")
|
|
else:
|
|
for event in today_events:
|
|
start = event.get("start", {})
|
|
time_str = ""
|
|
|
|
if "dateTime" in start:
|
|
# Timed event
|
|
dt = datetime.fromisoformat(start["dateTime"].replace("Z", "+00:00"))
|
|
time_str = dt.strftime("%I:%M %p").lstrip("0")
|
|
elif "date" in start:
|
|
time_str = "All day"
|
|
|
|
summary = event.get("summary", "(No title)")
|
|
duration = ""
|
|
|
|
# Calculate duration if end time available
|
|
end = event.get("end", {})
|
|
if "dateTime" in start and "dateTime" in end:
|
|
start_dt = datetime.fromisoformat(start["dateTime"].replace("Z", "+00:00"))
|
|
end_dt = datetime.fromisoformat(end["dateTime"].replace("Z", "+00:00"))
|
|
mins = int((end_dt - start_dt).total_seconds() / 60)
|
|
if mins >= 60:
|
|
hours = mins // 60
|
|
remaining = mins % 60
|
|
duration = f" ({hours}h{remaining}m)" if remaining else f" ({hours}h)"
|
|
else:
|
|
duration = f" ({mins}m)"
|
|
|
|
lines.append(f" • {time_str} - {summary}{duration}")
|
|
elif today_events and "error" in today_events[0]:
|
|
error = today_events[0].get("error", "Unknown")
|
|
lines.append(f"⚠️ Could not fetch calendar: {error}")
|
|
else:
|
|
lines.append("No events today")
|
|
|
|
# Tomorrow preview
|
|
if tomorrow_events is not None:
|
|
if tomorrow_events and (len(tomorrow_events) == 0 or "error" not in tomorrow_events[0]):
|
|
count = len(tomorrow_events)
|
|
if count > 0:
|
|
first = tomorrow_events[0]
|
|
start = first.get("start", {})
|
|
if "dateTime" in start:
|
|
dt = datetime.fromisoformat(start["dateTime"].replace("Z", "+00:00"))
|
|
first_time = dt.strftime("%I:%M %p").lstrip("0")
|
|
else:
|
|
first_time = "All day"
|
|
lines.append(f"Tomorrow: {count} event{'s' if count > 1 else ''}, first at {first_time}")
|
|
else:
|
|
lines.append("Tomorrow: No events")
|
|
|
|
return "\n".join(lines) if lines else "No calendar data"
|
|
|
|
|
|
def collect(config: dict) -> dict:
|
|
"""Main collector entry point."""
|
|
cal_config = config.get("calendar", {})
|
|
show_tomorrow = cal_config.get("show_tomorrow", True)
|
|
|
|
today_events = fetch_events("today")
|
|
tomorrow_events = fetch_events("tomorrow") if show_tomorrow else None
|
|
|
|
formatted = format_events(today_events, tomorrow_events)
|
|
|
|
has_error = today_events and len(today_events) == 1 and "error" in today_events[0]
|
|
|
|
return {
|
|
"section": "Today",
|
|
"icon": "📅",
|
|
"content": formatted,
|
|
"raw": {"today": today_events, "tomorrow": tomorrow_events},
|
|
"error": today_events[0].get("error") if has_error else None
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
config = {"calendar": {"show_tomorrow": True}}
|
|
result = collect(config)
|
|
print(f"## {result['icon']} {result['section']}")
|
|
print(result["content"])
|