pyr0ball
37d151725e
Implements idempotent calendar push for Apple Calendar (CalDAV) and Google Calendar from the Interviews kanban. - db: add calendar_event_id column (migration) + set_calendar_event_id helper - integrations/apple_calendar: create_event / update_event via caldav + icalendar - integrations/google_calendar: create_event / update_event via google-api-python-client; test() now makes a real API call instead of checking file existence - scripts/calendar_push: orchestrates push/update, builds event title from stage + job title + company, attaches job URL and company brief to description, defaults to noon UTC / 1hr duration - app/pages/5_Interviews: "Add to Calendar" / "Update Calendar" button shown when interview date is set and a calendar integration is configured - environment.yml: pin caldav, icalendar, google-api-python-client, google-auth - tests/test_calendar_push: 9 tests covering create, update, error handling, event timing, idempotency, and missing job/date guards
79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
from __future__ import annotations
|
|
import os
|
|
from datetime import datetime
|
|
from scripts.integrations.base import IntegrationBase
|
|
|
|
|
|
class GoogleCalendarIntegration(IntegrationBase):
|
|
name = "google_calendar"
|
|
label = "Google Calendar"
|
|
tier = "paid"
|
|
|
|
def __init__(self):
|
|
self._config: dict = {}
|
|
|
|
def fields(self) -> list[dict]:
|
|
return [
|
|
{"key": "calendar_id", "label": "Calendar ID", "type": "text",
|
|
"placeholder": "primary or xxxxx@group.calendar.google.com", "required": True,
|
|
"help": "Settings → Calendars → [name] → Integrate calendar → Calendar ID"},
|
|
{"key": "credentials_json", "label": "Service Account JSON path", "type": "text",
|
|
"placeholder": "~/credentials/google-calendar-sa.json", "required": True,
|
|
"help": "Download from Google Cloud Console → Service Accounts → Keys"},
|
|
]
|
|
|
|
def connect(self, config: dict) -> bool:
|
|
self._config = config
|
|
return bool(config.get("calendar_id") and config.get("credentials_json"))
|
|
|
|
def test(self) -> bool:
|
|
try:
|
|
service = self._build_service()
|
|
service.calendars().get(calendarId=self._config["calendar_id"]).execute()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def _build_service(self):
|
|
from google.oauth2 import service_account
|
|
from googleapiclient.discovery import build
|
|
creds_path = os.path.expanduser(self._config["credentials_json"])
|
|
creds = service_account.Credentials.from_service_account_file(
|
|
creds_path,
|
|
scopes=["https://www.googleapis.com/auth/calendar"],
|
|
)
|
|
return build("calendar", "v3", credentials=creds)
|
|
|
|
def _fmt(self, dt: datetime) -> str:
|
|
return dt.strftime("%Y-%m-%dT%H:%M:%S") + "Z"
|
|
|
|
def create_event(self, uid: str, title: str, start_dt: datetime,
|
|
end_dt: datetime, description: str = "") -> str:
|
|
"""Create a Google Calendar event. Returns the Google event ID."""
|
|
service = self._build_service()
|
|
body = {
|
|
"summary": title,
|
|
"description": description,
|
|
"start": {"dateTime": self._fmt(start_dt), "timeZone": "UTC"},
|
|
"end": {"dateTime": self._fmt(end_dt), "timeZone": "UTC"},
|
|
"extendedProperties": {"private": {"peregrine_uid": uid}},
|
|
}
|
|
result = service.events().insert(
|
|
calendarId=self._config["calendar_id"], body=body
|
|
).execute()
|
|
return result["id"]
|
|
|
|
def update_event(self, uid: str, title: str, start_dt: datetime,
|
|
end_dt: datetime, description: str = "") -> str:
|
|
"""Update an existing Google Calendar event by its stored event ID (uid is the gcal id)."""
|
|
service = self._build_service()
|
|
body = {
|
|
"summary": title,
|
|
"description": description,
|
|
"start": {"dateTime": self._fmt(start_dt), "timeZone": "UTC"},
|
|
"end": {"dateTime": self._fmt(end_dt), "timeZone": "UTC"},
|
|
}
|
|
result = service.events().update(
|
|
calendarId=self._config["calendar_id"], eventId=uid, body=body
|
|
).execute()
|
|
return result["id"]
|