File size: 6,577 Bytes
77da5ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """
calendar_intake.py — Extract life-state signals from Google Calendar.
Real OAuth flow mirrors gmail_intake.py. Falls back to demo_signals.json
automatically when credentials.json is absent (hackathon demo mode).
SETUP (real mode):
1. Enable Google Calendar API in console.cloud.google.com
2. Download credentials.json to the project root
3. pip install google-auth google-auth-oauthlib google-api-python-client
"""
import os
import json
from datetime import datetime, timedelta, timezone
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
_DEMO_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data', 'demo_signals.json')
class CalendarIntake:
# ── Real OAuth path ──────────────────────────────────────────────────
def authenticate(self):
"""Return an authenticated Calendar API service. Raises if credentials missing."""
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
creds = None
token_file = 'calendar_token.json'
if os.path.exists(token_file):
creds = Credentials.from_authorized_user_file(token_file, SCOPES)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
if not os.path.exists('credentials.json'):
raise FileNotFoundError("credentials.json missing.")
flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
creds = flow.run_local_server(port=0)
with open(token_file, 'w') as f:
f.write(creds.to_json())
return build('calendar', 'v3', credentials=creds)
def extract_signals(self, service, days: int = 7) -> dict:
"""Pull real calendar data and return structured signals."""
now = datetime.now(timezone.utc)
end = now + timedelta(days=days)
events_result = service.events().list(
calendarId='primary',
timeMin=now.isoformat(),
timeMax=end.isoformat(),
singleEvents=True,
orderBy='startTime',
maxResults=100,
).execute()
events = events_result.get('items', [])
total_minutes = 0
back_to_back = 0
personal = 0
deadlines = []
prev_end = None
for ev in events:
start_str = ev.get('start', {}).get('dateTime') or ev.get('start', {}).get('date')
end_str = ev.get('end', {}).get('dateTime') or ev.get('end', {}).get('date')
if not start_str or not end_str:
continue
try:
s = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
e = datetime.fromisoformat(end_str.replace('Z', '+00:00'))
except ValueError:
continue
duration = (e - s).total_seconds() / 60
total_minutes += duration
if prev_end and (s - prev_end).total_seconds() < 600:
back_to_back += 1
prev_end = e
title = ev.get('summary', '').lower()
if any(w in title for w in ('personal', 'gym', 'family', 'date', 'birthday', 'doctor')):
personal += 1
importance = ev.get('colorId')
if importance in ('11', '4') or any(w in title for w in ('deadline', 'submit', 'launch', 'board', 'review')):
deadlines.append({
"title": ev.get('summary', 'Untitled'),
"due_in_hours": round((s - now).total_seconds() / 3600),
"priority": "critical" if importance == '11' else "high",
})
working_minutes = days * 8 * 60
occupancy = min(100, round(total_minutes / working_minutes * 100))
avg_meeting_h = round(total_minutes / 60 / days, 1)
focus_blocks = max(0, days - back_to_back - 1)
return {
"week_occupancy_pct": occupancy,
"avg_meeting_hours_per_day": avg_meeting_h,
"back_to_back_blocks": back_to_back,
"focus_blocks_count": focus_blocks,
"personal_events_this_week": personal,
"upcoming_deadlines": deadlines[:3],
"summary": (
f"{occupancy}% of working hours booked. "
f"{avg_meeting_h}h meetings/day. "
f"{back_to_back} back-to-back chains. "
f"{len(deadlines)} deadlines upcoming."
),
}
def to_life_metrics(self, signals: dict) -> dict:
"""Map calendar signals to LifeMetrics deltas."""
occ = signals.get('week_occupancy_pct', 50)
btb = signals.get('back_to_back_blocks', 0)
focus = signals.get('focus_blocks_count', 3)
return {
"time.free_hours_per_week": -((occ - 50) / 5),
"time.schedule_control": -(occ / 10),
"mental_wellbeing.stress_level": (occ / 10) + (btb * 2),
"mental_wellbeing.focus_quality": focus * 5 - 10,
"career.workload": (occ - 50) / 2,
}
# ── Demo fallback ────────────────────────────────────────────────────
@staticmethod
def demo_signals() -> dict:
with open(_DEMO_PATH) as f:
return json.load(f)['calendar']
@staticmethod
def demo_life_metrics() -> dict:
with open(_DEMO_PATH) as f:
d = json.load(f)
return {k: v for k, v in d['derived_metric_deltas'].items()
if k.startswith('time.') or k.startswith('career.')}
# ── Unified entry point ──────────────────────────────────────────────
def sync(self) -> tuple[dict, dict, bool]:
"""
Returns (signals, metric_deltas, is_demo).
Tries real OAuth first; silently falls back to demo on any failure.
"""
try:
svc = self.authenticate()
sigs = self.extract_signals(svc)
return sigs, self.to_life_metrics(sigs), False
except Exception:
return self.demo_signals(), self.demo_life_metrics(), True
|