550 lines
21 KiB
Python
550 lines
21 KiB
Python
"""Outlook sync provider — bidirectional sync with Microsoft Graph API (U7).
|
|
|
|
Uses ``httpx.AsyncClient`` for all Graph API calls. Conflict resolution is
|
|
last-write-wins based on ``last_modified``; conflicts emit a
|
|
``calendar_sync_conflict`` WS notification via the injectable ``notify_callback``
|
|
(G4).
|
|
|
|
ponytail: browser OAuth flow (auth-code grant + redirect) is deferred to U12
|
|
settings UI. This provider assumes tokens are already stored in
|
|
``ExternalCalendarConfig.credentials``. Upgrade: add an ``OutlookOAuthFlow``
|
|
helper that performs the device-code or auth-code flow and writes tokens.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from collections.abc import Awaitable, Callable
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, quote, urlparse
|
|
|
|
import httpx
|
|
|
|
from agentkit.calendar.db import (
|
|
DEFAULT_CALENDAR_DB_PATH,
|
|
get_event_by_external_id,
|
|
insert_event,
|
|
update_event,
|
|
update_external_config,
|
|
)
|
|
from agentkit.calendar.models import CalendarEvent, ExternalCalendarConfig, _now_iso
|
|
from agentkit.calendar.sync.base import AbstractSyncProvider
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Async callback signature: (event_type: str, payload: dict) -> None
|
|
NotifyCallback = Callable[[str, dict[str, Any]], Awaitable[None]]
|
|
|
|
GRAPH_BASE = "https://graph.microsoft.com/v1.0"
|
|
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
|
|
DEFAULT_SCOPE = "https://graph.microsoft.com/Calendars.ReadWrite offline_access"
|
|
|
|
_DAY_MAP = {
|
|
"monday": "MO",
|
|
"tuesday": "TU",
|
|
"wednesday": "WE",
|
|
"thursday": "TH",
|
|
"friday": "FR",
|
|
"saturday": "SA",
|
|
"sunday": "SU",
|
|
}
|
|
_DAY_MAP_REVERSE = {v: k for k, v in _DAY_MAP.items()}
|
|
_FREQ_MAP = {
|
|
"daily": "DAILY",
|
|
"weekly": "WEEKLY",
|
|
"absoluteMonthly": "MONTHLY",
|
|
"absoluteYearly": "YEARLY",
|
|
}
|
|
_FREQ_MAP_REVERSE = {v: k for k, v in _FREQ_MAP.items()}
|
|
|
|
|
|
def _parse_iso(dt_str: str) -> datetime:
|
|
"""Parse ISO 8601 string to UTC-aware datetime."""
|
|
dt = datetime.fromisoformat(dt_str)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def _outlook_dt_to_iso(dt_obj: dict[str, Any]) -> str:
|
|
"""Convert Outlook dateTimeTimeZone to ISO 8601 UTC.
|
|
|
|
ponytail: assumes Graph returns UTC (no ``Prefer: outlook.timezone`` header
|
|
is sent). If a non-UTC timezone is returned, it's treated as UTC. Upgrade:
|
|
use ``zoneinfo`` with a Windows→IANA timezone mapping for correct conversion.
|
|
"""
|
|
if not dt_obj:
|
|
return ""
|
|
dt_str = dt_obj.get("dateTime", "")
|
|
if not dt_str:
|
|
return ""
|
|
if "T" in dt_str:
|
|
dt = datetime.fromisoformat(dt_str)
|
|
else:
|
|
# Date only (all-day event)
|
|
dt = datetime.fromisoformat(dt_str + "T00:00:00")
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc).isoformat()
|
|
|
|
|
|
def _iso_to_outlook_dt(iso_str: str, is_all_day: bool) -> dict[str, str]:
|
|
"""Convert ISO 8601 UTC to Outlook dateTimeTimeZone."""
|
|
if not iso_str:
|
|
return {"dateTime": "", "timeZone": "UTC"}
|
|
dt = _parse_iso(iso_str)
|
|
if is_all_day:
|
|
return {"dateTime": dt.strftime("%Y-%m-%d"), "timeZone": "UTC"}
|
|
return {"dateTime": dt.strftime("%Y-%m-%dT%H:%M:%S"), "timeZone": "UTC"}
|
|
|
|
|
|
def _outlook_recurrence_to_rrule(recurrence: dict[str, Any] | None) -> str | None:
|
|
"""Convert Outlook recurrence pattern to RRULE string."""
|
|
if not recurrence:
|
|
return None
|
|
pattern = recurrence.get("pattern", {}) or {}
|
|
range_obj = recurrence.get("range", {}) or {}
|
|
|
|
parts: list[str] = []
|
|
freq = _FREQ_MAP.get(pattern.get("type", ""))
|
|
if freq:
|
|
parts.append(f"FREQ={freq}")
|
|
|
|
interval = pattern.get("interval")
|
|
if interval and interval > 1:
|
|
parts.append(f"INTERVAL={interval}")
|
|
|
|
days = pattern.get("daysOfWeek", [])
|
|
if days:
|
|
bydays = [_DAY_MAP[d] for d in days if d in _DAY_MAP]
|
|
if bydays:
|
|
parts.append(f"BYDAY={','.join(bydays)}")
|
|
|
|
count = pattern.get("numberOfOccurrences")
|
|
if count:
|
|
parts.append(f"COUNT={count}")
|
|
elif range_obj.get("type") == "endDate" and range_obj.get("endDate"):
|
|
end_date = range_obj["endDate"] # "2026-12-31"
|
|
parts.append(f"UNTIL={end_date.replace('-', '')}T235959Z")
|
|
|
|
return ";".join(parts) if parts else None
|
|
|
|
|
|
def _rrule_to_outlook_recurrence(rrule: str, start_date: str) -> dict[str, Any] | None:
|
|
"""Convert RRULE string to Outlook recurrence pattern.
|
|
|
|
``start_date`` is the event's start date in ``YYYY-MM-DD`` format (required
|
|
by Graph API for the ``range.startDate`` field).
|
|
"""
|
|
parts: dict[str, str] = {}
|
|
for part in rrule.split(";"):
|
|
if "=" in part:
|
|
k, v = part.split("=", 1)
|
|
parts[k.upper()] = v
|
|
|
|
freq = parts.get("FREQ", "").upper()
|
|
pattern_type = _FREQ_MAP_REVERSE.get(freq)
|
|
if not pattern_type:
|
|
return None
|
|
|
|
pattern: dict[str, Any] = {"type": pattern_type}
|
|
|
|
interval = parts.get("INTERVAL")
|
|
pattern["interval"] = int(interval) if interval else 1
|
|
|
|
byday = parts.get("BYDAY")
|
|
if byday:
|
|
pattern["daysOfWeek"] = [
|
|
_DAY_MAP_REVERSE[d] for d in byday.split(",") if d in _DAY_MAP_REVERSE
|
|
]
|
|
|
|
count = parts.get("COUNT")
|
|
until = parts.get("UNTIL")
|
|
|
|
if count:
|
|
pattern["numberOfOccurrences"] = int(count)
|
|
range_obj: dict[str, Any] = {
|
|
"type": "numbered",
|
|
"startDate": start_date,
|
|
"numberOfOccurrences": int(count),
|
|
}
|
|
elif until:
|
|
# UNTIL=20261231T235959Z → "2026-12-31"
|
|
date_str = until[:8]
|
|
end_date = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}"
|
|
range_obj = {"type": "endDate", "startDate": start_date, "endDate": end_date}
|
|
else:
|
|
range_obj = {"type": "noEnd", "startDate": start_date}
|
|
|
|
return {"pattern": pattern, "range": range_obj}
|
|
|
|
|
|
def _extract_delta_token(delta_link: str) -> str | None:
|
|
"""Extract ``$deltaToken`` from a Graph delta link URL."""
|
|
parsed = urlparse(delta_link)
|
|
params = parse_qs(parsed.query)
|
|
values = params.get("$deltaToken", [])
|
|
return values[0] if values else None
|
|
|
|
|
|
class OutlookSyncProvider(AbstractSyncProvider):
|
|
"""Bidirectional Outlook sync provider via Microsoft Graph REST API.
|
|
|
|
The ``client_factory`` parameter allows tests to inject a mock
|
|
``httpx.AsyncClient`` without making real HTTP calls. When ``None``, a
|
|
real ``httpx.AsyncClient`` is constructed per-operation.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
db_path: str | Path | None = None,
|
|
client_factory: Callable[[], Any] | None = None,
|
|
notify_callback: NotifyCallback | None = None,
|
|
) -> None:
|
|
self.db_path = Path(db_path) if db_path is not None else DEFAULT_CALENDAR_DB_PATH
|
|
self._client_factory = client_factory
|
|
self._notify = notify_callback
|
|
# ponytail: conflicts list is in-memory only; if the process restarts
|
|
# before a sync completes, conflict history is lost. Upgrade: persist
|
|
# to a calendar_sync_conflicts table.
|
|
|
|
# ------------------------------------------------------------------
|
|
# Client / auth
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get_client(self) -> Any:
|
|
"""Return an httpx.AsyncClient (real or mock from factory)."""
|
|
if self._client_factory is not None:
|
|
return self._client_factory()
|
|
return httpx.AsyncClient(timeout=30.0)
|
|
|
|
def _load_creds(self, config: ExternalCalendarConfig) -> dict[str, Any]:
|
|
return json.loads(config.credentials) if config.credentials else {}
|
|
|
|
def _save_creds(self, config: ExternalCalendarConfig, creds: dict[str, Any]) -> None:
|
|
config.credentials = json.dumps(creds)
|
|
|
|
async def _refresh_token(self, client: Any, config: ExternalCalendarConfig) -> dict[str, Any]:
|
|
"""Refresh the access token using the refresh_token grant.
|
|
|
|
Posts to the Azure AD token endpoint, updates ``config.credentials``
|
|
in-memory, and persists the new credentials to the DB.
|
|
"""
|
|
creds = self._load_creds(config)
|
|
resp = await client.request(
|
|
"POST",
|
|
TOKEN_URL,
|
|
data={
|
|
"client_id": creds.get("client_id", ""),
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": creds.get("refresh_token", ""),
|
|
"scope": DEFAULT_SCOPE,
|
|
},
|
|
)
|
|
if resp.status_code in (400, 401):
|
|
logger.error("Outlook refresh token expired or revoked (status=%s)", resp.status_code)
|
|
if self._notify is not None:
|
|
await self._notify(
|
|
"calendar_sync_error",
|
|
{
|
|
"config_id": config.id,
|
|
"message": "Outlook authentication expired, please re-authenticate",
|
|
},
|
|
)
|
|
# ponytail: re-auth UI is in U12. Upgrade: trigger OAuth re-flow automatically.
|
|
raise RuntimeError("Outlook refresh token expired — re-authentication required")
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
creds["access_token"] = payload["access_token"]
|
|
if "refresh_token" in payload:
|
|
creds["refresh_token"] = payload["refresh_token"]
|
|
creds["expires_at"] = (
|
|
datetime.now(timezone.utc) + timedelta(seconds=payload.get("expires_in", 3600))
|
|
).isoformat()
|
|
self._save_creds(config, creds)
|
|
await update_external_config(config.id, {"credentials": config.credentials}, self.db_path)
|
|
return creds
|
|
|
|
async def _request(
|
|
self,
|
|
client: Any,
|
|
config: ExternalCalendarConfig,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
json_body: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Make an authenticated Graph API request with 401 auto-refresh + retry."""
|
|
creds = self._load_creds(config)
|
|
headers = {"Authorization": f"Bearer {creds.get('access_token', '')}"}
|
|
resp = await client.request(method, url, headers=headers, json=json_body)
|
|
if resp.status_code == 401:
|
|
creds = await self._refresh_token(client, config)
|
|
headers = {"Authorization": f"Bearer {creds.get('access_token', '')}"}
|
|
resp = await client.request(method, url, headers=headers, json=json_body)
|
|
resp.raise_for_status()
|
|
return resp.json() if resp.text else {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Pull
|
|
# ------------------------------------------------------------------
|
|
|
|
async def pull_changes(
|
|
self, config: ExternalCalendarConfig, since: str | None = None
|
|
) -> list[CalendarEvent]:
|
|
"""Pull remote Outlook events via delta query.
|
|
|
|
First call (no ``sync_token``) is a full sync within a date range.
|
|
Subsequent calls use the stored delta token for incremental sync.
|
|
Returns pulled/updated events.
|
|
"""
|
|
client = self._get_client()
|
|
try:
|
|
remote_events, delta_token = await self._pull_delta(client, config)
|
|
finally:
|
|
await client.aclose()
|
|
|
|
# Persist delta token for next incremental sync
|
|
if delta_token:
|
|
config.sync_token = delta_token
|
|
await update_external_config(config.id, {"sync_token": delta_token}, self.db_path)
|
|
|
|
result: list[CalendarEvent] = []
|
|
for remote in remote_events:
|
|
local = await get_event_by_external_id(
|
|
remote.external_id, "outlook", config.user_id, self.db_path
|
|
)
|
|
if local is None:
|
|
# New remote event → create local
|
|
await insert_event(remote, self.db_path)
|
|
result.append(remote)
|
|
else:
|
|
# Existing → check for conflict
|
|
resolved = await self._resolve_pull_conflict(local, remote)
|
|
if resolved is not None:
|
|
result.append(resolved)
|
|
return result
|
|
|
|
async def _pull_delta(
|
|
self, client: Any, config: ExternalCalendarConfig
|
|
) -> tuple[list[CalendarEvent], str | None]:
|
|
"""Call /me/calendarView/delta. Returns (events, delta_token)."""
|
|
url = self._build_delta_url(config)
|
|
# ponytail: single-page fetch; pagination via @odata.nextLink is not
|
|
# followed. Upgrade: loop on nextLink until exhausted, then read
|
|
# deltaLink from the final page.
|
|
payload = await self._request(client, config, "GET", url)
|
|
events: list[CalendarEvent] = []
|
|
for raw in payload.get("value", []):
|
|
parsed = self._parse_outlook_event(raw, config.user_id)
|
|
if parsed is not None:
|
|
events.append(parsed)
|
|
delta_link = payload.get("@odata.deltaLink")
|
|
delta_token = _extract_delta_token(delta_link) if delta_link else None
|
|
return events, delta_token
|
|
|
|
def _build_delta_url(self, config: ExternalCalendarConfig) -> str:
|
|
"""Build the delta query URL. Uses sync_token if present (incremental)."""
|
|
if config.sync_token:
|
|
return f"{GRAPH_BASE}/me/calendarView/delta?$deltaToken={quote(config.sync_token, safe='')}"
|
|
# Initial sync — use date range to scope the fetch
|
|
start = (datetime.now(timezone.utc) - timedelta(days=365)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
end = (datetime.now(timezone.utc) + timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
return f"{GRAPH_BASE}/me/calendarView/delta?startDateTime={start}&endDateTime={end}"
|
|
|
|
def _parse_outlook_event(self, raw: dict[str, Any], user_id: str) -> CalendarEvent | None:
|
|
"""Convert a Graph event JSON to a CalendarEvent dataclass."""
|
|
eid = raw.get("id")
|
|
if not eid:
|
|
return None
|
|
title = raw.get("subject") or ""
|
|
if not title:
|
|
return None
|
|
|
|
start_str = _outlook_dt_to_iso(raw.get("start", {}))
|
|
end_str = _outlook_dt_to_iso(raw.get("end", {})) or start_str
|
|
is_all_day = bool(raw.get("isAllDay", False))
|
|
|
|
body = raw.get("body", {}) or {}
|
|
description = body.get("content", "") or ""
|
|
|
|
location_obj = raw.get("location", {}) or {}
|
|
location = location_obj.get("displayName", "") or ""
|
|
|
|
rrule = _outlook_recurrence_to_rrule(raw.get("recurrence"))
|
|
|
|
last_modified = raw.get("lastModifiedDateTime", "") or _now_iso()
|
|
now = _now_iso()
|
|
return CalendarEvent(
|
|
id=uuid.uuid4().hex,
|
|
user_id=user_id,
|
|
title=title,
|
|
description=description,
|
|
start_time=start_str,
|
|
end_time=end_str,
|
|
is_all_day=is_all_day,
|
|
location=location,
|
|
rrule=rrule,
|
|
source="manual",
|
|
external_id=eid,
|
|
external_provider="outlook",
|
|
last_modified=last_modified,
|
|
created_at=now,
|
|
)
|
|
|
|
async def _resolve_pull_conflict(
|
|
self, local: CalendarEvent, remote: CalendarEvent
|
|
) -> CalendarEvent | None:
|
|
"""Resolve conflict when both local and remote exist (last-write-wins).
|
|
|
|
If remote is newer → update local. If local is newer → conflict
|
|
(last-write-wins keeps local, but log + notify). If equal → no-op.
|
|
Returns the winning event (or None if local kept unchanged).
|
|
"""
|
|
local_lm = (
|
|
_parse_iso(local.last_modified)
|
|
if local.last_modified
|
|
else datetime.min.replace(tzinfo=timezone.utc)
|
|
)
|
|
remote_lm = (
|
|
_parse_iso(remote.last_modified)
|
|
if remote.last_modified
|
|
else datetime.min.replace(tzinfo=timezone.utc)
|
|
)
|
|
|
|
if remote_lm > local_lm:
|
|
# Remote wins → update local
|
|
await self._notify_conflict(local, remote, winner="remote")
|
|
fields = {
|
|
"title": remote.title,
|
|
"description": remote.description,
|
|
"start_time": remote.start_time,
|
|
"end_time": remote.end_time,
|
|
"is_all_day": remote.is_all_day,
|
|
"location": remote.location,
|
|
"rrule": remote.rrule,
|
|
"last_modified": remote.last_modified,
|
|
}
|
|
await update_event(local.id, fields, self.db_path)
|
|
return remote
|
|
|
|
if local_lm > remote_lm:
|
|
# Local wins → conflict, keep local, notify
|
|
await self._notify_conflict(local, remote, winner="local")
|
|
return None
|
|
|
|
# Equal timestamps → no change needed
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Push
|
|
# ------------------------------------------------------------------
|
|
|
|
async def push_changes(
|
|
self, config: ExternalCalendarConfig, events: list[CalendarEvent]
|
|
) -> list[CalendarEvent]:
|
|
"""Push local events to Outlook. Returns events with external_id set."""
|
|
client = self._get_client()
|
|
try:
|
|
result: list[CalendarEvent] = []
|
|
for event in events:
|
|
updated = await self._push_single(client, config, event)
|
|
result.append(updated)
|
|
finally:
|
|
await client.aclose()
|
|
return result
|
|
|
|
async def _push_single(
|
|
self, client: Any, config: ExternalCalendarConfig, event: CalendarEvent
|
|
) -> CalendarEvent:
|
|
"""Push a single event to Outlook, return event with external_id set."""
|
|
body = self._event_to_outlook(event)
|
|
if event.external_id:
|
|
# Update existing remote event
|
|
url = f"{GRAPH_BASE}/me/events/{event.external_id}"
|
|
await self._request(client, config, "PATCH", url, json_body=body)
|
|
fields = {"last_modified": _now_iso()}
|
|
await update_event(event.id, fields, self.db_path)
|
|
return event
|
|
# Create new remote event
|
|
url = f"{GRAPH_BASE}/me/events"
|
|
payload = await self._request(client, config, "POST", url, json_body=body)
|
|
new_id = payload.get("id")
|
|
if new_id:
|
|
fields = {
|
|
"external_id": new_id,
|
|
"external_provider": "outlook",
|
|
"last_modified": _now_iso(),
|
|
}
|
|
await update_event(event.id, fields, self.db_path)
|
|
event.external_id = new_id
|
|
event.external_provider = "outlook"
|
|
return event
|
|
|
|
def _event_to_outlook(self, event: CalendarEvent) -> dict[str, Any]:
|
|
"""Convert CalendarEvent to Outlook Graph event JSON."""
|
|
body: dict[str, Any] = {
|
|
"subject": event.title,
|
|
"start": _iso_to_outlook_dt(event.start_time, event.is_all_day),
|
|
"end": _iso_to_outlook_dt(event.end_time, event.is_all_day),
|
|
"isAllDay": event.is_all_day,
|
|
}
|
|
if event.description:
|
|
body["body"] = {"contentType": "Text", "content": event.description}
|
|
if event.location:
|
|
body["location"] = {"displayName": event.location}
|
|
if event.rrule:
|
|
start_date = event.start_time[:10] if event.start_time else "2026-01-01"
|
|
rec = _rrule_to_outlook_recurrence(event.rrule, start_date)
|
|
if rec is not None:
|
|
body["recurrence"] = rec
|
|
return body
|
|
|
|
# ------------------------------------------------------------------
|
|
# Test connection
|
|
# ------------------------------------------------------------------
|
|
|
|
async def test_connection(self, config: ExternalCalendarConfig) -> tuple[bool, str]:
|
|
"""Test Outlook connectivity via GET /me. Returns (ok, error_msg)."""
|
|
client = self._get_client()
|
|
try:
|
|
try:
|
|
await self._request(client, config, "GET", f"{GRAPH_BASE}/me")
|
|
ok, msg = True, ""
|
|
except Exception as e:
|
|
ok, msg = False, str(e)
|
|
finally:
|
|
await client.aclose()
|
|
return ok, msg
|
|
|
|
# ------------------------------------------------------------------
|
|
# Conflict notification
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _notify_conflict(
|
|
self, local: CalendarEvent, remote: CalendarEvent, winner: str
|
|
) -> None:
|
|
"""Log conflict and send WS notification via callback (G4)."""
|
|
logger.info(
|
|
"Sync conflict for event %s (external_id=%s): local_lm=%s remote_lm=%s winner=%s",
|
|
local.id,
|
|
local.external_id,
|
|
local.last_modified,
|
|
remote.last_modified,
|
|
winner,
|
|
)
|
|
if self._notify is not None:
|
|
payload = {
|
|
"event_id": local.id,
|
|
"title": local.title,
|
|
"external_id": local.external_id,
|
|
"local_last_modified": local.last_modified,
|
|
"remote_last_modified": remote.last_modified,
|
|
"winner": winner,
|
|
}
|
|
await self._notify("calendar_sync_conflict", payload)
|