#!/usr/bin/env python3 """Cairn → Google Calendar. Every open card with a due date is an event in a calendar of Cairn's own, named "Cairn". One way: the board is the truth. A card added, edited, moved or snoozed updates its event; ticked or deleted, its event goes. An event changed in Google Calendar is put back the next time its card changes. ⬜ Dragging an event to another day to move the card's date is the next step. - Scope: calendar.app.created. Cairn can make its own calendars and touch only those; it can't see or change any other calendar. The privacy page (jacobkunselman.com/cairn/privacy/) says so; change it if this ever changes. - A card with a date is an all-day event shown as free; with a time, a 30-minute event. No Google reminders: ntfy already pings the phone (notify.py), and two alerts per card would be noise. - The event's colour is the Google colour nearest the category's. - The client and the token are in ~/.config/cairn/, mode 600, outside the repo. Which event is which card is in board/.calendar.json (gitignored). - server.py runs sync() after every change from the app and every 5 minutes, which catches cairn commands too. gcal.py auth-url print the sign-in address (Desktop client, PKCE, redirect to localhost:8771) gcal.py auth-finish URL the address the browser landed on after Allow: exchange its code for a token gcal.py sync sync now, and say what changed gcal.py status signed in? which calendar? how many events? """ import base64, datetime as dt, fcntl, hashlib, json, os, secrets, sys, urllib.error, urllib.parse, urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import board as B import notify CONF = Path.home() / ".config" / "cairn" CLIENT = CONF / "google-client.json" TOKEN = CONF / "google-token.json" PENDING = CONF / "google-auth-pending.json" STATE = B.BOARD / ".calendar.json" SCOPE = "https://www.googleapis.com/auth/calendar.app.created" REDIRECT = "http://localhost:8771" AUTH = "https://accounts.google.com/o/oauth2/v2/auth" TOKEN_URL = "https://oauth2.googleapis.com/token" API = "https://www.googleapis.com/calendar/v3" TZ = "America/Chicago" NAME = "Cairn" # Google Calendar's event colours (colorId → hex), as its API lists them GOOGLE = {"1": "#A4BDFC", "2": "#7AE7BF", "3": "#DBADFF", "4": "#FF887C", "5": "#FBD75B", "6": "#FFB878", "7": "#46D6DB", "8": "#E1E1E1", "9": "#5484ED", "10": "#51B749", "11": "#DC2127"} class NeedAuth(Exception): pass def http(method, url, body=None, form=None, token=None): headers, data = {}, None if form is not None: data, headers["Content-Type"] = urllib.parse.urlencode(form).encode(), "application/x-www-form-urlencoded" elif body is not None: data, headers["Content-Type"] = json.dumps(body).encode(), "application/json" if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=20) as r: raw = r.read() return r.status, json.loads(raw) if raw else {} except urllib.error.HTTPError as e: raw = e.read() try: return e.code, json.loads(raw) except ValueError: return e.code, {"raw": raw[:300].decode(errors="replace")} def private_write(path, obj): CONF.mkdir(mode=0o700, exist_ok=True) tmp = path.with_suffix(".part") with os.fdopen(os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f: json.dump(obj, f) tmp.replace(path) def client(): c = json.loads(CLIENT.read_text()) return c.get("installed") or c["web"] # ---------------------------------------------------------------- signing in def auth_url(): verifier = secrets.token_urlsafe(64) challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() state = secrets.token_urlsafe(16) private_write(PENDING, {"verifier": verifier, "state": state}) return AUTH + "?" + urllib.parse.urlencode({ "client_id": client()["client_id"], "redirect_uri": REDIRECT, "response_type": "code", "scope": SCOPE, "access_type": "offline", "prompt": "consent", "code_challenge": challenge, "code_challenge_method": "S256", "state": state}) def auth_finish(landed): q = dict(urllib.parse.parse_qsl(urllib.parse.urlparse(landed).query)) pending = json.loads(PENDING.read_text()) if q.get("error"): raise SystemExit(f"Google said: {q['error']}") if q.get("state") != pending["state"]: raise SystemExit("that address isn't from the sign-in this started (state differs)") c = client() code, r = http("POST", TOKEN_URL, form={"code": q["code"], "client_id": c["client_id"], "client_secret": c["client_secret"], "redirect_uri": REDIRECT, "grant_type": "authorization_code", "code_verifier": pending["verifier"]}) if code != 200 or "refresh_token" not in r: raise SystemExit(f"token exchange failed: {code} {r.get('error')} {r.get('error_description', '')}") if SCOPE not in r.get("scope", "").split(): raise SystemExit(f"Google granted {r.get('scope')!r}, not the calendar scope: was the box left unticked?") private_write(TOKEN, {"refresh_token": r["refresh_token"], "scope": r["scope"], "obtained": dt.datetime.now().isoformat()}) PENDING.unlink() _access.clear() _access = {} def access_token(): if _access.get("until", 0) > dt.datetime.now().timestamp() + 60: return _access["token"] if not TOKEN.exists(): raise NeedAuth("not signed in") c = client() code, r = http("POST", TOKEN_URL, form={"client_id": c["client_id"], "client_secret": c["client_secret"], "refresh_token": json.loads(TOKEN.read_text())["refresh_token"], "grant_type": "refresh_token"}) if code == 400 and r.get("error") == "invalid_grant": raise NeedAuth("Google revoked the sign-in (invalid_grant)") if code != 200: raise RuntimeError(f"token refresh: {code} {r.get('error')}") _access.update(token=r["access_token"], until=dt.datetime.now().timestamp() + r.get("expires_in", 3600)) return _access["token"] # ---------------------------------------------------------------- what an event looks like def nearest_colour(hex_): rgb = lambda h: [int(h[i:i + 2], 16) for i in (1, 3, 5)] a = rgb(hex_) return min(GOOGLE, key=lambda k: sum((x - y) ** 2 for x, y in zip(a, rgb(GOOGLE[k])))) def event(card, colours): notes = "\n".join(card.notes) ev = {"summary": card.title, "status": "confirmed", "description": (notes + "\n\n" if notes else "") + f"Cairn ^{card.id} · {notify.APP_URL}/#{card.id}", "reminders": {"useDefault": False, "overrides": []}, "source": {"title": "Cairn", "url": f"{notify.APP_URL}/#{card.id}"}} if card.cat in colours: ev["colorId"] = nearest_colour(colours[card.cat]) if len(card.due) > 10: start = dt.datetime.fromisoformat(card.due) ev["start"] = {"dateTime": start.isoformat(timespec="seconds"), "timeZone": TZ} ev["end"] = {"dateTime": (start + dt.timedelta(minutes=30)).isoformat(timespec="seconds"), "timeZone": TZ} ev["transparency"] = "opaque" else: day = dt.date.fromisoformat(card.due) ev["start"], ev["end"] = {"date": day.isoformat()}, {"date": (day + dt.timedelta(days=1)).isoformat()} ev["transparency"] = "transparent" return ev def wanted(board): colours = {c["tag"]: c["colour"] for c in B.categories()} return {c.id: event(c, colours) for col in ("Doing", "Todo", "Later") for c in board.columns[col] if c.due} # ---------------------------------------------------------------- the sync def load_state(): try: return json.loads(STATE.read_text()) except (OSError, ValueError): return {"events": {}} def save_state(state): tmp = STATE.with_suffix(".part") tmp.write_text(json.dumps(state, indent=1)) tmp.replace(STATE) def calendar(state, token): if not state.get("calendar"): code, r = http("POST", f"{API}/calendars", {"summary": NAME, "timeZone": TZ, "description": f"Jacob's task board, written by Cairn ({notify.APP_URL}). Change tasks there: " "an edit made here is put back the next time the task changes."}, token=token) if code != 200: raise RuntimeError(f"making the calendar: {code} {r}") state["calendar"] = r["id"] save_state(state) return state["calendar"] def sync(): """Bring the Cairn calendar into line with the board. Returns (added, changed, removed). One at a time: the server and a hand-run `gcal.py sync` must not both add the same event.""" with open(B.BOARD / ".calendar.lock", "w") as lock: fcntl.flock(lock, fcntl.LOCK_EX) return _sync() def _sync(): token = access_token() state = load_state() events = state.setdefault("events", {}) want = wanted(B.load()) added = changed = removed = 0 for attempt in (1, 2): cal = urllib.parse.quote(calendar(state, token), safe="") try: for cid, ev in want.items(): digest = hashlib.sha1(json.dumps(ev, sort_keys=True).encode()).hexdigest() have = events.get(cid) if have and have["hash"] == digest: continue code = None if have: code, r = http("PUT", f"{API}/calendars/{cal}/events/{have['event']}", ev, token=token) changed += code == 200 if not have or code in (404, 410): code, r = http("POST", f"{API}/calendars/{cal}/events", ev, token=token) added += code == 200 if code == 404: raise LookupError("calendar gone") if code != 200: raise RuntimeError(f"event for ^{cid}: {code} {r.get('error', r)}") events[cid] = {"event": r["id"], "hash": digest} save_state(state) for cid in [c for c in events if c not in want]: code, r = http("DELETE", f"{API}/calendars/{cal}/events/{events[cid]['event']}", token=token) if code not in (200, 204, 404, 410): raise RuntimeError(f"removing ^{cid}: {code} {r.get('error', r)}") del events[cid] removed += 1 save_state(state) return added, changed, removed except LookupError: # the calendar was deleted in Google: make a new one, once state = {"events": {}} events = state["events"] save_state(state) raise RuntimeError("the calendar keeps disappearing") def sync_quietly(): """The server's call: never raises. A lost sign-in is pushed to the phone once, not every 5 minutes.""" if not TOKEN.exists(): return try: result = sync() state = load_state() if state.pop("auth_lost", None): save_state(state) if any(result): print("calendar: +%d ~%d -%d" % result, flush=True) except NeedAuth as e: state = load_state() if not state.get("auth_lost"): state["auth_lost"] = dt.datetime.now().isoformat() save_state(state) print(f"calendar: {e}", flush=True) try: notify.send("Cairn lost Google Calendar", "Sign in again: python3 ~/projects/cairn/gcal.py auth-url", 3, ["warning"]) except Exception: pass except Exception as e: print(f"calendar: {e}", flush=True) if __name__ == "__main__": cmd = sys.argv[1] if len(sys.argv) > 1 else "status" if cmd == "auth-url": print(auth_url()) elif cmd == "auth-finish": auth_finish(sys.argv[2]) print("signed in; the calendar syncs within 5 minutes, or run: gcal.py sync") elif cmd == "sync": print("added %d, changed %d, removed %d" % sync()) elif cmd == "status": state = load_state() print(f"signed in: {TOKEN.exists()} calendar: {state.get('calendar', '-')} events: {len(state.get('events', {}))}" + (f" ⚠ sign-in lost {state['auth_lost']}" if state.get("auth_lost") else "")) else: sys.exit(__doc__)