rex/calendar_service.py
Rama master · réplica de lectura del repositorio original.
"""
Calendar service module for Rex AI Assistant.
Provides calendar integration with read/write capabilities using per-user
stores (stub/mock data or an ICS backend resolved per authorized account).
Per-user isolation (issue #303):
- Every operation that reads or mutates calendar data requires an explicit,
validated ``user_id``. Missing, blank, malformed, or traversal-style
identities fail closed (``CalendarIdentityError``) before any account or
credential lookup.
- Account selection is restricted to the requesting user's authorized
accounts via :class:`rex.calendar_accounts.CalendarAccountResolver`;
explicit foreign or nonexistent accounts raise the generic
``CalendarAccountAccessError`` (indistinguishable to the caller).
- Backends are cached per ``(user_id, account_id)``; a backend resolved for
one user is never reused for another.
- Stub/mock mode keeps a separate mutable event store per user so one
user's create/update/delete never alters another user's view.
- Private event fields (titles, attendees, locations, descriptions) are
published only on user-scoped topics (``{topic}.user.{user_id}``); shared
topics carry a safe envelope (user_id, count, success) only.
"""
from __future__ import annotations
import json
import logging
import os
import uuid
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from rex.assistant_errors import IntegrationNotConfiguredError
from rex.calendar_accounts import (
ACCOUNT_UNAVAILABLE_MSG,
DEFAULT_PROFILE,
CalendarAccountAccessError,
CalendarAccountDefinition,
CalendarAccountResolver,
require_user_id,
)
from rex.openclaw.event_bus import EventBus
logger = logging.getLogger(__name__)
_REPO_SEED_PATH = Path("data/mock_calendar.json")
_NOT_CONFIGURED_FOR_USER_MSG = "Calendar: not configured"
def _runtime_calendar_path(user_id: str) -> Path:
"""Return a writable per-user runtime path for calendar persistence.
Uses OS-appropriate app data directories so the repo's
data/mock_calendar.json is never modified at runtime. The legacy
single-user file (``rex-ai/calendar.json``) is preserved for the
``default`` profile only; named users get isolated per-user files.
"""
user_id = require_user_id(user_id)
if os.name == "nt":
base = Path(os.environ.get("LOCALAPPDATA", str(Path.home() / "AppData" / "Local")))
else:
base = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local" / "share")))
if user_id == DEFAULT_PROFILE:
return base / "rex-ai" / "calendar.json"
return base / "rex-ai" / "calendar" / f"{user_id}.json"
class _NoOpEventBus:
"""Fallback event bus used when no EventBus is provided."""
def publish(self, *_args: Any, **_kwargs: Any) -> None:
return
def _ensure_aware_utc(dt: datetime) -> datetime:
"""
Ensure datetime is timezone-aware in UTC.
If a naive datetime is provided, it is assumed to already be UTC.
"""
if dt.tzinfo is None:
return dt.replace(tzinfo=UTC)
return dt.astimezone(UTC)
@dataclass(slots=True, init=False)
class CalendarEvent:
event_id: str
title: str
start_time: datetime
end_time: datetime
attendees: list[str] = field(default_factory=list)
location: str | None = None
description: str | None = None
all_day: bool = False
def __init__(
self,
*,
event_id: str | None = None,
id: str | None = None,
title: str,
start_time: datetime,
end_time: datetime,
attendees: list[str] | None = None,
location: str | None = None,
description: str | None = None,
all_day: bool = False,
) -> None:
resolved_id = event_id or id or str(uuid.uuid4())
self.event_id = resolved_id
self.title = title
self.start_time = _ensure_aware_utc(start_time)
self.end_time = _ensure_aware_utc(end_time)
self.attendees = list(attendees) if attendees is not None else []
self.location = location
self.description = description
self.all_day = all_day
@property
def id(self) -> str:
"""Compatibility alias for code that expects `id`."""
return self.event_id
@id.setter
def id(self, value: str) -> None:
self.event_id = value
def to_summary(self) -> dict[str, Any]:
return {
"event_id": self.event_id,
"id": self.event_id,
"title": self.title,
"start_time": self.start_time.isoformat(),
"end_time": self.end_time.isoformat(),
"attendees": list(self.attendees),
"location": self.location,
"description": self.description,
"all_day": self.all_day,
}
def overlaps_with(self, other: CalendarEvent) -> bool:
"""Return True if this event overlaps with another event."""
return self.start_time < other.end_time and self.end_time > other.start_time
class _EventStore:
"""One user's isolated mutable event list, optionally disk-backed."""
__slots__ = ("account_id", "events", "storage_path")
def __init__(
self,
events: list[CalendarEvent],
storage_path: Path | None,
account_id: str | None,
) -> None:
self.events = events
self.storage_path = storage_path
self.account_id = account_id
class CalendarService:
"""
Read/write calendar service with per-user isolated stores.
Storage modes:
- ``mock_events`` provided: in-memory only, no disk writes. The events
belong to ``owner_user_id`` (default: the ``default`` profile); other
users see their own empty isolated stores.
- ``mock_data_path`` provided: read/write to that path (tests use
tmp_path) for ``owner_user_id`` only; other users get isolated
in-memory stores.
- Neither provided: accounts are resolved per validated user via
:class:`CalendarAccountResolver`. Users whose resolved account is an
ICS source get a read-seeded in-memory store; stub users get an
isolated per-user runtime file seeded from ``data/mock_calendar.json``.
When real accounts are configured, a user with no authorized account
fails closed (reads return ``[]``, writes raise
:class:`IntegrationNotConfiguredError`) — never another user's
account.
"""
def __init__(
self,
event_bus: EventBus | None = None,
*,
mock_data_path: Path | str | None = None,
mock_events: list[CalendarEvent] | None = None,
owner_user_id: str = DEFAULT_PROFILE,
account_resolver: CalendarAccountResolver | None = None,
) -> None:
self._event_bus = event_bus if event_bus is not None else _NoOpEventBus()
self._owner = require_user_id(owner_user_id)
self._mock_events: list[CalendarEvent] | None = (
list(mock_events) if mock_events is not None else None
)
self._mock_data_path: Path | None = Path(mock_data_path) if mock_data_path else None
# Per-user isolated stores, keyed by (user_id, account_id or "").
self._stores: dict[tuple[str, str], _EventStore] = {}
# ICS backends cached per (user_id, account_id). A backend created
# for one user is never served to another.
self._user_backends: dict[tuple[str, str], Any] = {}
# Injected authorization/routing resolver (tests/embedding only; the
# production path loads lazily so config changes — including account
# revocations — are picked up without a restart).
self._injected_resolver = account_resolver
self._resolver_cache: CalendarAccountResolver | None = None
self._resolver_stamp: int | None = None
self.connected = False
# ------------------------------------------------------------------
# Resolver / store plumbing
# ------------------------------------------------------------------
def _get_resolver(self) -> CalendarAccountResolver:
"""Return the account resolver, refreshing when the config changes.
Authorization is re-checked on every operation, so cached stores and
backends never outlive their owner's assignment.
"""
if self._injected_resolver is not None:
return self._injected_resolver
from rex import calendar_accounts as _calendar_accounts
stamp = _calendar_accounts.config_stamp()
if self._resolver_cache is None or stamp != self._resolver_stamp:
self._resolver_cache = CalendarAccountResolver.load()
self._resolver_stamp = stamp
return self._resolver_cache
def _seeded_stub_events(self, user_id: str) -> tuple[list[CalendarEvent], Path | None]:
"""Load a user's stub store from disk (runtime file, then repo seed)."""
storage_path = _runtime_calendar_path(user_id)
for path in (storage_path, _REPO_SEED_PATH):
if path is not None and path.exists():
try:
return self._load_mock_events(path), storage_path
except Exception as exc:
logger.warning("Failed to load calendar data from %s: %s", path, exc)
return [], storage_path
def _get_store(
self,
user_id: str,
account_id: str | None = None,
) -> _EventStore | None:
"""Return *user_id*'s isolated store, after ownership validation.
Returns ``None`` when real accounts are configured but none is
available/usable for this user (documented not-configured result —
callers fail closed, never through another user's account).
Raises:
CalendarIdentityError: On missing or invalid identity.
CalendarAccountAccessError: When *account_id* is explicitly
requested but not available to this user.
"""
validated = require_user_id(user_id)
# In-memory / explicit-path modes are bound to exactly one owner.
if self._mock_events is not None or self._mock_data_path is not None:
self._reject_explicit_account(validated, account_id)
return self._owner_bound_store(validated)
resolver = self._get_resolver()
if not resolver.has_configured_accounts():
# Pure stub mode: isolated per-user disk store.
self._reject_explicit_account(validated, account_id)
return self._stub_disk_store(validated, None)
# Accounts are configured: ownership check before anything else.
definition = resolver.resolve_account(validated, account_id)
if definition is None:
logger.warning("No usable calendar account for user %r", validated)
return None
if definition.provider == "ics":
return self._ics_store(validated, definition)
if definition.provider == "stub":
return self._stub_disk_store(validated, definition.id)
# google/outlook accounts are served by the provider-API surface
# (rex.integrations.calendar_service), not this store-backed service.
logger.warning(
"Calendar account for user %r uses provider %r, which this surface does not serve",
validated,
definition.provider,
)
return None
@staticmethod
def _reject_explicit_account(validated: str, account_id: str | None) -> None:
"""Reject explicit account selection where no accounts exist.
Unauthorized and nonexistent accounts are indistinguishable.
"""
if account_id:
raise CalendarAccountAccessError(
ACCOUNT_UNAVAILABLE_MSG.format(account_id=account_id, user_id=validated)
)
def _owner_bound_store(self, validated: str) -> _EventStore:
"""Store for in-memory / explicit-path modes (owner-bound).
The configured owner gets the injected events or path; any other
user gets an isolated empty in-memory store.
"""
key = (validated, "")
store = self._stores.get(key)
if store is not None:
return store
if validated != self._owner:
# Isolated empty in-memory store for any other user.
store = _EventStore([], None, None)
elif self._mock_events is not None:
store = _EventStore(list(self._mock_events), None, None)
else:
events: list[CalendarEvent] = []
assert self._mock_data_path is not None
if self._mock_data_path.exists():
try:
events = self._load_mock_events(self._mock_data_path)
except Exception as exc:
logger.warning(
"Failed to load calendar data from %s: %s",
self._mock_data_path,
exc,
)
store = _EventStore(events, self._mock_data_path, None)
self._stores[key] = store
return store
def _stub_disk_store(self, validated: str, account_id: str | None) -> _EventStore:
"""The user's isolated, disk-backed stub store."""
key = (validated, account_id or "")
store = self._stores.get(key)
if store is None:
events, storage_path = self._seeded_stub_events(validated)
store = _EventStore(events, storage_path, account_id)
self._stores[key] = store
return store
def _ics_store(self, user_id: str, definition: CalendarAccountDefinition) -> _EventStore | None:
"""Return the user's store seeded from their authorized ICS source."""
key = (user_id, definition.id)
store = self._stores.get(key)
if store is not None:
return store
backend = self._user_backends.get(key)
if backend is None:
from rex.calendar_backends.ics_backend import ICSCalendarBackend
backend = ICSCalendarBackend(
source=definition.ics_source,
url_timeout=definition.ics_url_timeout,
)
if not backend.connect():
logger.warning("ICS calendar backend failed to connect for user %r", user_id)
return None
self._user_backends[key] = backend
events = list(backend.fetch_events())
store = _EventStore(events, None, definition.id)
self._stores[key] = store
return store
# ------------------------------------------------------------------
# Event publishing
# ------------------------------------------------------------------
def _publish(self, topic: str, payload: dict[str, Any]) -> None:
try:
self._event_bus.publish(topic, payload)
except Exception as exc:
logger.debug("EventBus publish failed for %s: %s", topic, exc)
def _publish_user_event(
self,
topic: str,
user_id: str,
shared_payload: dict[str, Any],
private_payload: dict[str, Any] | None = None,
) -> None:
"""Publish a safe envelope on the shared topic and the full payload
on the owner's user-scoped topic.
The shared topic must never carry private event fields (titles,
attendees, locations, descriptions); those go only to
``{topic}.user.{user_id}``.
"""
self._publish(topic, shared_payload)
if private_payload is not None:
self._publish(f"{topic}.user.{user_id}", private_payload)
# ------------------------------------------------------------------
# Connection
# ------------------------------------------------------------------
def connect(self, user_id: str | None = None) -> bool:
"""
Connect the calendar service for one validated user.
Loads that user's isolated events. Fails closed (returns ``False``)
on missing/invalid identity or when no account is available to the
user.
"""
try:
validated = require_user_id(user_id)
except PermissionError:
logger.warning("Calendar connect refused: a valid user identity is required")
return False
try:
store = self._get_store(validated)
except PermissionError:
return False
except Exception as exc:
logger.error("Failed to connect calendar service: %s", exc, exc_info=True)
self._publish(
"calendar.connected",
{"connected": False, "user_id": validated, "error": str(exc)},
)
return False
if store is None:
self._publish(
"calendar.connected",
{"connected": False, "user_id": validated},
)
return False
self.connected = True
logger.info("Calendar service connected (local storage) for user %s", validated)
self._publish(
"calendar.connected",
{"connected": True, "count": len(store.events), "user_id": validated},
)
return True
# ------------------------------------------------------------------
# Read operations
# ------------------------------------------------------------------
def _events_for(self, user_id: str, account_id: str | None = None) -> list[CalendarEvent]:
"""Return a copy of the requesting user's events (fail closed to [])."""
store = self._get_store(user_id, account_id)
if store is None:
return []
return list(store.events)
def list_upcoming(
self,
*,
horizon_hours: int = 72,
user_id: str | None = None,
account_id: str | None = None,
) -> list[CalendarEvent]:
validated = require_user_id(user_id)
events = self._events_for(validated, account_id)
now = datetime.now(UTC)
horizon = now + timedelta(hours=horizon_hours)
# Upcoming includes events that start within horizon or are currently ongoing
upcoming = [
event for event in events if (event.start_time <= horizon) and (event.end_time > now)
]
upcoming.sort(key=lambda e: e.start_time)
self._publish_user_event(
"calendar.upcoming",
validated,
{"count": len(upcoming), "user_id": validated},
{
"count": len(upcoming),
"user_id": validated,
"events": [e.to_summary() for e in upcoming],
},
)
return upcoming
def refresh_upcoming(
self, *, user_id: str | None = None, account_id: str | None = None
) -> list[CalendarEvent]:
return self.list_upcoming(user_id=user_id, account_id=account_id)
def get_events(
self,
start: datetime,
end: datetime,
*,
user_id: str | None = None,
account_id: str | None = None,
) -> list[CalendarEvent]:
"""
Get the requesting user's calendar events that overlap a time range.
"""
validated = require_user_id(user_id)
start_utc = _ensure_aware_utc(start)
end_utc = _ensure_aware_utc(end)
events = self._events_for(validated, account_id)
result = [e for e in events if e.start_time < end_utc and e.end_time > start_utc]
result.sort(key=lambda e: e.start_time)
self._publish_user_event(
"calendar.range",
validated,
{
"count": len(result),
"user_id": validated,
"start": start_utc.isoformat(),
"end": end_utc.isoformat(),
},
)
return result
def list_past_events(
self, *, lookback_hours: int = 72, user_id: str | None = None
) -> list[CalendarEvent]:
"""Return the user's events that ended within the lookback window."""
now = datetime.now(UTC)
start = now - timedelta(hours=lookback_hours)
events = self.get_events(start, now, user_id=user_id)
past_events = [event for event in events if event.end_time <= now]
past_events.sort(key=lambda e: e.end_time)
return past_events
def get_upcoming_events(
self, days: int = 7, *, user_id: str | None = None
) -> list[CalendarEvent]:
now = datetime.now(UTC)
end = now + timedelta(days=days)
return self.get_events(now, end, user_id=user_id)
def get_all_events(self, *, user_id: str | None = None) -> list[CalendarEvent]:
validated = require_user_id(user_id)
return self._events_for(validated)
def get_past_events(
self,
hours: int = 72,
*,
now: datetime | None = None,
user_id: str | None = None,
) -> list[CalendarEvent]:
"""Get the user's events that ended within the specified time window.
Args:
hours: Look back this many hours for ended events.
now: Current time (defaults to UTC now).
user_id: Requesting user (required).
Returns:
List of events that ended within the window, sorted by end_time.
"""
validated = require_user_id(user_id)
check_time = now or datetime.now(UTC)
window_start = check_time - timedelta(hours=hours)
events = self._events_for(validated)
past_events = [e for e in events if e.end_time <= check_time and e.end_time >= window_start]
past_events.sort(key=lambda e: e.end_time, reverse=True)
return past_events
# ------------------------------------------------------------------
# Write operations
# ------------------------------------------------------------------
def create_event(
self,
title: str,
start_time: datetime,
end_time: datetime,
*,
location: str | None = None,
attendees: Iterable[str] | None = None,
description: str | None = None,
all_day: bool = False,
user_id: str | None = None,
account_id: str | None = None,
) -> CalendarEvent:
"""
Create a new calendar event in the requesting user's store.
Raises:
CalendarIdentityError: On missing or invalid identity.
CalendarAccountAccessError: When *account_id* is not available
to this user (unauthorized or nonexistent).
IntegrationNotConfiguredError: When accounts are configured but
none is available to this user (fail closed — never another
user's account).
"""
validated = require_user_id(user_id)
store = self._get_store(validated, account_id)
if store is None:
raise IntegrationNotConfiguredError(_NOT_CONFIGURED_FOR_USER_MSG)
event = CalendarEvent(
event_id=str(uuid.uuid4()),
title=title,
start_time=start_time,
end_time=end_time,
location=location,
attendees=list(attendees) if attendees is not None else [],
description=description,
all_day=all_day,
)
store.events.append(event)
store.events.sort(key=lambda e: e.start_time)
self._save_store(store)
self._publish_user_event(
"calendar.created",
validated,
{"user_id": validated, "created": True},
{"user_id": validated, "created": True, "event": event.to_summary()},
)
return event
def update_event(
self,
event_id: str,
updates: dict[str, Any],
*,
user_id: str | None = None,
) -> CalendarEvent | None:
"""
Update an event by id within the requesting user's store only.
"""
validated = require_user_id(user_id)
store = self._get_store(validated)
if store is None:
logger.warning("Calendar update refused: no account for user %r", validated)
return None
for i, event in enumerate(store.events):
if event.event_id != event_id:
continue
# Apply supported updates safely
if "title" in updates:
event.title = str(updates["title"])
if "start_time" in updates and isinstance(updates["start_time"], datetime):
event.start_time = _ensure_aware_utc(updates["start_time"])
if "end_time" in updates and isinstance(updates["end_time"], datetime):
event.end_time = _ensure_aware_utc(updates["end_time"])
if "location" in updates:
event.location = updates["location"]
if "description" in updates:
event.description = updates["description"]
if "all_day" in updates:
event.all_day = bool(updates["all_day"])
if "attendees" in updates:
att = updates["attendees"]
if isinstance(att, (list, tuple)):
event.attendees = [str(a) for a in att]
store.events[i] = event
store.events.sort(key=lambda e: e.start_time)
self._save_store(store)
self._publish_user_event(
"calendar.updated",
validated,
{"user_id": validated, "updated": True},
{"user_id": validated, "updated": True, "event": event.to_summary()},
)
return event
logger.warning("Event not found for update: %s", event_id)
self._publish_user_event(
"calendar.updated",
validated,
{"user_id": validated, "updated": False},
)
return None
def delete_event(self, event_id: str, *, user_id: str | None = None) -> bool:
"""
Delete an event by id within the requesting user's store only.
"""
validated = require_user_id(user_id)
store = self._get_store(validated)
if store is None:
logger.warning("Calendar delete refused: no account for user %r", validated)
return False
new_events = [e for e in store.events if e.event_id != event_id]
if len(new_events) == len(store.events):
logger.warning("Event not found for delete: %s", event_id)
self._publish_user_event(
"calendar.deleted",
validated,
{"user_id": validated, "deleted": False},
)
return False
store.events[:] = new_events
self._save_store(store)
self._publish_user_event(
"calendar.deleted",
validated,
{"user_id": validated, "deleted": True},
{"user_id": validated, "deleted": True, "event_id": event_id},
)
return True
# ------------------------------------------------------------------
# Conflict detection
# ------------------------------------------------------------------
def detect_conflicts(
self, event: CalendarEvent, *, user_id: str | None = None
) -> list[CalendarEvent]:
"""
Detect conflicts between the provided event and the user's events.
"""
validated = require_user_id(user_id)
conflicts: list[CalendarEvent] = []
for existing in self._events_for(validated):
if existing.event_id == event.event_id:
continue
if existing.overlaps_with(event):
conflicts.append(existing)
return conflicts
def find_conflicts(
self,
events: list[CalendarEvent] | None = None,
*,
user_id: str | None = None,
) -> list[tuple[CalendarEvent, CalendarEvent]]:
"""
Find overlapping event pairs within the user's events.
"""
validated = require_user_id(user_id)
events_to_check = list(events) if events is not None else self._events_for(validated)
events_to_check.sort(key=lambda e: e.start_time)
conflicts: list[tuple[CalendarEvent, CalendarEvent]] = []
for i, e1 in enumerate(events_to_check):
for e2 in events_to_check[i + 1 :]:
if e1.overlaps_with(e2):
conflicts.append((e1, e2))
return conflicts
# ------------------------------------------------------------------
# Persistence helpers
# ------------------------------------------------------------------
def _load_mock_events(self, path: Path) -> list[CalendarEvent]:
"""
Supports two file formats:
Format A:
{ "events": [ { "event_id": "...", "title": "...", "start_time": "...", "end_time": "...", ... } ] }
Format B:
[ { "id": "...", "title": "...", "start_time": "...", "end_time": "...", ... }, ... ]
"""
payload = json.loads(path.read_text(encoding="utf-8"))
raw_events: list[dict[str, Any]]
if isinstance(payload, dict) and isinstance(payload.get("events"), list):
raw_events = payload["events"]
elif isinstance(payload, list):
raw_events = payload
else:
logger.warning("Unrecognized mock calendar format in %s", path)
return []
events: list[CalendarEvent] = []
for item in raw_events:
event_id = item.get("event_id") or item.get("id") or str(uuid.uuid4())
title = item.get("title", "Untitled event")
start_raw = item.get("start_time")
end_raw = item.get("end_time")
if isinstance(start_raw, str):
start_dt = datetime.fromisoformat(start_raw)
elif isinstance(start_raw, datetime):
start_dt = start_raw
else:
start_dt = datetime.now(UTC)
if isinstance(end_raw, str):
end_dt = datetime.fromisoformat(end_raw)
elif isinstance(end_raw, datetime):
end_dt = end_raw
else:
end_dt = start_dt + timedelta(hours=1)
attendees = item.get("attendees") or []
if not isinstance(attendees, list):
attendees = []
events.append(
CalendarEvent(
event_id=str(event_id),
title=str(title),
start_time=_ensure_aware_utc(start_dt),
end_time=_ensure_aware_utc(end_dt),
location=item.get("location"),
description=item.get("description"),
attendees=[str(a) for a in attendees],
all_day=bool(item.get("all_day", False)),
)
)
events.sort(key=lambda e: e.start_time)
return events
def _save_store(self, store: _EventStore) -> None:
"""Persist a store to its own path (no-op for in-memory stores).
The repo seed file is never modified.
"""
if store.storage_path is None:
return
try:
store.storage_path.parent.mkdir(parents=True, exist_ok=True)
data = {"events": [e.to_summary() for e in store.events]}
store.storage_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
except Exception as exc:
logger.error("Failed to save calendar data: %s", exc, exc_info=True)
# ------------------------------------------------------------------
# Follow-up cues
# ------------------------------------------------------------------
def generate_followup_cues(
self,
user_id: str,
*,
lookback_hours: int = 72,
expire_hours: int = 168,
now: datetime | None = None,
) -> int:
"""Generate follow-up cues from the user's recent past events.
Creates cues for events that have ended within the lookback window.
Skips:
- Events that already have a cue
- All-day events that look like holidays
- Events with 'no-followup' in metadata/description
Args:
user_id: User ID whose events are read and whose cues are created.
lookback_hours: Only consider events ended within this many hours.
expire_hours: Hours until the cue expires.
now: Current time (defaults to UTC now).
Returns:
Number of cues created.
"""
validated = require_user_id(user_id)
try:
from rex.cue_store import get_cue_store
except ImportError:
logger.warning("CueStore not available, cannot generate followup cues")
return 0
check_time = now or datetime.now(UTC)
past_events = self.get_past_events(hours=lookback_hours, now=check_time, user_id=validated)
cue_store = get_cue_store()
created_count = 0
for event in past_events:
# Skip if cue already exists for this event
if cue_store.has_cue_for_source(validated, "calendar", event.event_id):
continue
# Skip all-day events that look like holidays
if event.all_day and self._looks_like_holiday(event):
continue
# Skip events marked as no-followup
if self._is_no_followup(event):
continue
# Create the cue
prompt = f"How did '{event.title}' go?"
cue_store.add_cue(
user_id=validated,
source_type="calendar",
source_id=event.event_id,
title=event.title,
prompt=prompt,
eligible_after=event.end_time,
expires_in=timedelta(hours=expire_hours),
metadata={
"event_id": event.event_id,
"start_time": event.start_time.isoformat(),
"end_time": event.end_time.isoformat(),
"location": event.location,
},
)
created_count += 1
logger.debug(f"Created followup cue for event '{event.title}'")
if created_count:
logger.info(f"Generated {created_count} followup cue(s) from calendar events")
return created_count
def _looks_like_holiday(self, event: CalendarEvent) -> bool:
"""Check if an all-day event looks like a holiday.
Simple heuristic based on common holiday keywords.
"""
if not event.all_day:
return False
title_lower = event.title.lower()
holiday_keywords = [
"holiday",
"day off",
"vacation",
"pto",
"christmas",
"thanksgiving",
"easter",
"new year",
"independence day",
"memorial day",
"labor day",
"birthday",
"anniversary",
]
for keyword in holiday_keywords:
if keyword in title_lower:
return True
return False
def _is_no_followup(self, event: CalendarEvent) -> bool:
"""Check if an event is marked as no-followup.
Checks for 'no-followup' or 'nofollowup' in:
- Description
- Title (unlikely but possible)
"""
markers = ["no-followup", "nofollowup", "no_followup", "[no followup]"]
# Check title
title_lower = event.title.lower()
for marker in markers:
if marker in title_lower:
return True
# Check description
if event.description:
desc_lower = event.description.lower()
for marker in markers:
if marker in desc_lower:
return True
return False
# Global calendar service instance (optional convenience)
_calendar_service: CalendarService | None = None
def get_calendar_service(
event_bus: EventBus | None = None,
config: dict | None = None,
) -> CalendarService:
"""Get the global calendar service instance.
The returned service enforces per-user account ownership internally;
every operation requires a validated ``user_id``. Accounts come from
``calendar.accounts`` plus the legacy global calendar configuration
(``calendar.backend = "ics"`` / ``calendar.provider``), which is usable
only by the explicit ``default`` profile.
Raises:
IntegrationNotConfiguredError: when no calendar accounts are
configured at all.
"""
global _calendar_service
if _calendar_service is not None:
return _calendar_service
if config is not None:
resolver = CalendarAccountResolver.from_raw_config(config)
injected: CalendarAccountResolver | None = resolver
else:
resolver = CalendarAccountResolver.load()
# No resolver is injected: the service reloads it when the config
# file changes, so account revocations take effect in long-lived
# processes without a restart.
injected = None
if not resolver.has_configured_accounts():
raise IntegrationNotConfiguredError("Calendar: not configured")
service = CalendarService(event_bus=event_bus, account_resolver=injected)
# Legacy compatibility: when the default profile resolves to an ICS
# account, connect it eagerly so misconfigured sources fail fast (the
# pre-#303 behaviour).
try:
default_definition = resolver.resolve_account(DEFAULT_PROFILE)
except PermissionError:
default_definition = None
if default_definition is not None and default_definition.provider == "ics":
if not service.connect(user_id=DEFAULT_PROFILE):
raise IntegrationNotConfiguredError("Calendar ICS backend failed to connect")
_calendar_service = service
return _calendar_service
def set_calendar_service(service: CalendarService | None) -> None:
"""Set the global calendar service instance (for testing)."""
global _calendar_service
_calendar_service = service
__all__ = [
"CalendarEvent",
"CalendarService",
"IntegrationNotConfiguredError",
"get_calendar_service",
"set_calendar_service",
]