305 lines
9.6 KiB
Python
305 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Google Calendar checker for Echo.
|
|
Returns events for today, tomorrow, this week, or upcoming travel needs.
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from google.oauth2.credentials import Credentials
|
|
from googleapiclient.discovery import build
|
|
|
|
TOKEN_FILE = Path(__file__).parent.parent / 'credentials' / 'google-calendar-token.json'
|
|
TZ = ZoneInfo('Europe/Bucharest')
|
|
|
|
# Keywords that indicate travel to București (needs train + accommodation)
|
|
TRAVEL_KEYWORDS = ['nlp', 'bucuresti', 'bucurești', 'bucharest']
|
|
|
|
def get_service():
|
|
"""Get authenticated Calendar service."""
|
|
creds = Credentials.from_authorized_user_file(str(TOKEN_FILE))
|
|
return build('calendar', 'v3', credentials=creds)
|
|
|
|
def get_events(service, time_min, time_max, max_results=20):
|
|
"""Get events between time_min and time_max."""
|
|
events_result = service.events().list(
|
|
calendarId='primary',
|
|
timeMin=time_min.isoformat(),
|
|
timeMax=time_max.isoformat(),
|
|
maxResults=max_results,
|
|
singleEvents=True,
|
|
orderBy='startTime'
|
|
).execute()
|
|
return events_result.get('items', [])
|
|
|
|
def format_event(event):
|
|
"""Format event for display."""
|
|
start = event['start'].get('dateTime', event['start'].get('date'))
|
|
summary = event.get('summary', '(fără titlu)')
|
|
|
|
# Parse start time
|
|
if 'T' in start:
|
|
dt = datetime.fromisoformat(start)
|
|
time_str = dt.strftime('%H:%M')
|
|
date_str = dt.strftime('%d %b')
|
|
else:
|
|
time_str = 'toată ziua'
|
|
date_str = datetime.fromisoformat(start).strftime('%d %b')
|
|
|
|
return {
|
|
'summary': summary,
|
|
'date': date_str,
|
|
'time': time_str,
|
|
'start': start,
|
|
'is_travel': any(kw in summary.lower() for kw in TRAVEL_KEYWORDS)
|
|
}
|
|
|
|
def check_today_tomorrow():
|
|
"""Get events for today and tomorrow."""
|
|
service = get_service()
|
|
now = datetime.now(TZ)
|
|
|
|
# Today
|
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
today_end = today_start + timedelta(days=1)
|
|
|
|
# Tomorrow
|
|
tomorrow_end = today_end + timedelta(days=1)
|
|
|
|
today_events = get_events(service, today_start, today_end)
|
|
tomorrow_events = get_events(service, today_end, tomorrow_end)
|
|
|
|
result = {
|
|
'today': [format_event(e) for e in today_events],
|
|
'tomorrow': [format_event(e) for e in tomorrow_events]
|
|
}
|
|
|
|
return result
|
|
|
|
def check_week():
|
|
"""Get events for this week (Mon-Sun)."""
|
|
service = get_service()
|
|
now = datetime.now(TZ)
|
|
|
|
# Start of this week (Monday)
|
|
days_since_monday = now.weekday()
|
|
week_start = now.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=days_since_monday)
|
|
week_end = week_start + timedelta(days=7)
|
|
|
|
events = get_events(service, week_start, week_end)
|
|
|
|
return {
|
|
'week_start': week_start.strftime('%d %b'),
|
|
'week_end': (week_end - timedelta(days=1)).strftime('%d %b'),
|
|
'events': [format_event(e) for e in events]
|
|
}
|
|
|
|
def check_travel_upcoming():
|
|
"""Check for travel events in next 14 days that need booking (7-11 days out)."""
|
|
service = get_service()
|
|
now = datetime.now(TZ)
|
|
|
|
# Look 14 days ahead
|
|
future = now + timedelta(days=14)
|
|
events = get_events(service, now, future)
|
|
|
|
reminders = []
|
|
for event in events:
|
|
formatted = format_event(event)
|
|
if formatted['is_travel']:
|
|
# Calculate days until event
|
|
start_str = event['start'].get('dateTime', event['start'].get('date'))
|
|
if 'T' in start_str:
|
|
event_date = datetime.fromisoformat(start_str).date()
|
|
else:
|
|
event_date = datetime.fromisoformat(start_str).date()
|
|
|
|
days_until = (event_date - now.date()).days
|
|
|
|
# Remind if 7-11 days away (booking window)
|
|
if 7 <= days_until <= 11:
|
|
reminders.append({
|
|
**formatted,
|
|
'days_until': days_until,
|
|
'action': 'Cumpără bilete tren + asigură cazare București'
|
|
})
|
|
# Urgent if 3-6 days away and might have missed window
|
|
elif 3 <= days_until <= 6:
|
|
reminders.append({
|
|
**formatted,
|
|
'days_until': days_until,
|
|
'action': '⚠️ URGENT: Verifică dacă ai bilete și cazare!'
|
|
})
|
|
|
|
return {'travel_reminders': reminders}
|
|
|
|
def is_busy_now():
|
|
"""Check if there's an event happening RIGHT NOW."""
|
|
service = get_service()
|
|
now = datetime.now(TZ)
|
|
|
|
# Check events that started before now and end after now
|
|
events_result = service.events().list(
|
|
calendarId='primary',
|
|
timeMin=(now - timedelta(hours=4)).isoformat(),
|
|
timeMax=(now + timedelta(minutes=30)).isoformat(),
|
|
singleEvents=True,
|
|
orderBy='startTime'
|
|
).execute()
|
|
|
|
for event in events_result.get('items', []):
|
|
start_str = event['start'].get('dateTime', event['start'].get('date'))
|
|
end_str = event['end'].get('dateTime', event['end'].get('date'))
|
|
|
|
# Skip all-day events for "busy now" check
|
|
if 'T' not in start_str:
|
|
continue
|
|
|
|
start = datetime.fromisoformat(start_str)
|
|
end = datetime.fromisoformat(end_str)
|
|
|
|
if start <= now <= end:
|
|
return {
|
|
'busy': True,
|
|
'event': event.get('summary', '(fără titlu)'),
|
|
'ends': end.strftime('%H:%M')
|
|
}
|
|
|
|
return {'busy': False}
|
|
|
|
def check_upcoming_hours(hours=2):
|
|
"""Check for events in the next N hours."""
|
|
service = get_service()
|
|
now = datetime.now(TZ)
|
|
future = now + timedelta(hours=hours)
|
|
|
|
events = get_events(service, now, future)
|
|
|
|
alerts = []
|
|
for event in events:
|
|
start_str = event['start'].get('dateTime', event['start'].get('date'))
|
|
summary = event.get('summary', '(fără titlu)')
|
|
|
|
if 'T' in start_str:
|
|
start = datetime.fromisoformat(start_str)
|
|
minutes_until = int((start - now).total_seconds() / 60)
|
|
if minutes_until > 0:
|
|
alerts.append({
|
|
'summary': summary,
|
|
'minutes_until': minutes_until,
|
|
'time': start.strftime('%H:%M')
|
|
})
|
|
|
|
return {'upcoming': alerts}
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: calendar_check.py [today|week|travel|busy|soon|all]")
|
|
sys.exit(1)
|
|
|
|
mode = sys.argv[1].lower()
|
|
|
|
if mode == 'today':
|
|
result = check_today_tomorrow()
|
|
elif mode == 'week':
|
|
result = check_week()
|
|
elif mode == 'travel':
|
|
result = check_travel_upcoming()
|
|
elif mode == 'busy':
|
|
result = is_busy_now()
|
|
elif mode == 'soon':
|
|
hours = int(sys.argv[2]) if len(sys.argv) > 2 else 2
|
|
result = check_upcoming_hours(hours)
|
|
elif mode == 'all':
|
|
result = {
|
|
**check_today_tomorrow(),
|
|
**check_week(),
|
|
**check_travel_upcoming()
|
|
}
|
|
else:
|
|
print(f"Unknown mode: {mode}")
|
|
sys.exit(1)
|
|
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
|
|
def create_event(summary, start_datetime, duration_minutes=60, description=None,
|
|
reminders=None, is_travel=False):
|
|
"""
|
|
Create a calendar event with reminders.
|
|
|
|
Args:
|
|
summary: Event title
|
|
start_datetime: datetime object or ISO string (e.g., "2026-02-05T15:00:00")
|
|
duration_minutes: Duration in minutes (default 60)
|
|
description: Optional description
|
|
reminders: List of minutes before event for reminders, or None for defaults
|
|
e.g., [120, 30] = 2 hours and 30 min before
|
|
is_travel: If True, uses travel reminders (evening before + 2h before)
|
|
|
|
Returns:
|
|
Created event details
|
|
"""
|
|
service = get_service()
|
|
|
|
# Parse start time if string
|
|
if isinstance(start_datetime, str):
|
|
start = datetime.fromisoformat(start_datetime)
|
|
else:
|
|
start = start_datetime
|
|
|
|
# Ensure timezone
|
|
if start.tzinfo is None:
|
|
start = start.replace(tzinfo=TZ)
|
|
|
|
end = start + timedelta(minutes=duration_minutes)
|
|
|
|
# Set up reminders
|
|
if reminders is None:
|
|
if is_travel:
|
|
# Travel: evening before (calculate minutes to 18:00 day before) + 2h before
|
|
evening_before = start.replace(hour=18, minute=0, second=0) - timedelta(days=1)
|
|
minutes_to_evening = int((start - evening_before).total_seconds() / 60)
|
|
reminders = [minutes_to_evening, 120] # Evening before + 2 hours
|
|
else:
|
|
# Default: 30 min before
|
|
reminders = [30]
|
|
|
|
event = {
|
|
'summary': summary,
|
|
'start': {
|
|
'dateTime': start.isoformat(),
|
|
'timeZone': 'Europe/Bucharest',
|
|
},
|
|
'end': {
|
|
'dateTime': end.isoformat(),
|
|
'timeZone': 'Europe/Bucharest',
|
|
},
|
|
'reminders': {
|
|
'useDefault': False,
|
|
'overrides': [
|
|
{'method': 'popup', 'minutes': m} for m in reminders
|
|
],
|
|
},
|
|
}
|
|
|
|
if description:
|
|
event['description'] = description
|
|
|
|
created = service.events().insert(calendarId='primary', body=event).execute()
|
|
|
|
return {
|
|
'id': created['id'],
|
|
'summary': created['summary'],
|
|
'start': created['start'].get('dateTime'),
|
|
'link': created.get('htmlLink'),
|
|
'reminders': reminders
|
|
}
|