"""CalDAV pull sync — imports remote events into the internal event store. Runs as a scheduled job (hourly) and is also callable via the API. Only syncs events in a rolling 30-day-past / 180-day-future window. """ from __future__ import annotations import asyncio import logging import uuid from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import select, update from scribe.models import async_session from scribe.models.event import Event logger = logging.getLogger(__name__) _SYNC_PAST_DAYS = 30 _SYNC_FUTURE_DAYS = 180 # Wall-clock cap on the blocking CalDAV fetch so a hung/slow server can't # wedge the hourly sweep indefinitely. _SYNC_TIMEOUT_SECONDS = 120 def _parse_dt(val: Any) -> datetime | None: """Convert a date or datetime from an iCal component to a UTC-aware datetime.""" if val is None: return None import datetime as _dt_mod if isinstance(val, _dt_mod.datetime): if val.tzinfo is None: return val.replace(tzinfo=timezone.utc) return val.astimezone(timezone.utc) if isinstance(val, _dt_mod.date): # All-day date: treat as midnight UTC return datetime(val.year, val.month, val.day, tzinfo=timezone.utc) return None def _sync_one_user(config: dict[str, str], user_id: int) -> list[dict]: """Synchronous CalDAV fetch — runs in a thread executor.""" import caldav # noqa: PLC0415 now = datetime.now(timezone.utc) range_start = now - timedelta(days=_SYNC_PAST_DAYS) range_end = now + timedelta(days=_SYNC_FUTURE_DAYS) client = caldav.DAVClient( url=config["caldav_url"], username=config.get("caldav_username") or None, password=config.get("caldav_password") or None, ) principal = client.principal() calendars = principal.calendars() if not calendars: return [] cal_name = config.get("caldav_calendar_name", "") if cal_name: calendars = [c for c in calendars if c.name == cal_name] or calendars events: list[dict] = [] for calendar in calendars: try: results = calendar.date_search(start=range_start, end=range_end, expand=False) except Exception: logger.warning("CalDAV date_search failed for calendar %s", getattr(calendar, "name", "?"), exc_info=True) continue for vevent_obj in results: try: ical = vevent_obj.icalendar_instance for component in ical.walk(): if component.name != "VEVENT": continue dtstart = component.get("DTSTART") dtend = component.get("DTEND") uid = str(component.get("UID", "")) if not uid: continue start_dt = _parse_dt(dtstart.dt if dtstart else None) end_dt = _parse_dt(dtend.dt if dtend else None) if start_dt is None: continue import datetime as _dt_mod all_day = dtstart and isinstance(dtstart.dt, _dt_mod.date) and not isinstance(dtstart.dt, _dt_mod.datetime) rrule = component.get("RRULE") recurrence = rrule.to_ical().decode("utf-8") if rrule else None events.append({ "caldav_uid": uid, "title": str(component.get("SUMMARY", "")), "start_dt": start_dt, "end_dt": end_dt, "all_day": bool(all_day), "description": str(component.get("DESCRIPTION", "")), "location": str(component.get("LOCATION", "")), "recurrence": recurrence, }) except Exception: logger.debug("Failed to parse CalDAV event", exc_info=True) return events async def sync_user_events(user_id: int) -> dict: """Pull CalDAV events for one user and upsert into the DB. Returns a summary dict: {created, updated, unchanged}. """ from scribe.services.caldav import get_caldav_config, is_caldav_configured # noqa: PLC0415 if not await is_caldav_configured(user_id): return {"skipped": True, "reason": "CalDAV not configured"} config = await get_caldav_config(user_id) started = datetime.now(timezone.utc) range_start = started - timedelta(days=_SYNC_PAST_DAYS) range_end = started + timedelta(days=_SYNC_FUTURE_DAYS) loop = asyncio.get_running_loop() try: remote_events: list[dict] = await asyncio.wait_for( loop.run_in_executor(None, _sync_one_user, config, user_id), timeout=_SYNC_TIMEOUT_SECONDS, ) except asyncio.TimeoutError: logger.warning("CalDAV pull sync timed out for user %d after %ds", user_id, _SYNC_TIMEOUT_SECONDS) return {"error": "CalDAV fetch timed out"} except Exception: logger.warning("CalDAV pull sync failed for user %d", user_id, exc_info=True) return {"error": "CalDAV fetch failed"} created = updated = unchanged = skipped = deleted = 0 async with async_session() as session: for ev in remote_events: caldav_uid = ev["caldav_uid"] # Storage uses duration, not end_dt. Convert here so the # rest of this function can compare/upsert in one shape. ev_start = ev["start_dt"] ev_end = ev["end_dt"] ev_duration = ( int((ev_end - ev_start).total_seconds() // 60) if ev_end is not None and ev_start is not None and ev_end > ev_start else None ) ev["duration_minutes"] = ev_duration result = await session.execute( select(Event).where( Event.user_id == user_id, Event.caldav_uid == caldav_uid, ) ) existing = result.scalar_one_or_none() if existing is not None and existing.deleted_at is not None: # The user trashed this event locally. Don't resurrect it by # updating, and don't create a duplicate live copy — leave it # in the trash. (Propagating the delete to the remote server is # tracked separately.) skipped += 1 continue if existing is None: # Create new event new_ev = Event( user_id=user_id, uid=str(uuid.uuid4()), caldav_uid=caldav_uid, title=ev["title"], start_dt=ev_start, duration_minutes=ev_duration, all_day=ev["all_day"], description=ev["description"], location=ev["location"], recurrence=ev["recurrence"], ) session.add(new_ev) created += 1 else: # Update if anything changed changed = False for field in ("title", "start_dt", "duration_minutes", "all_day", "description", "location", "recurrence"): if getattr(existing, field) != ev[field]: setattr(existing, field, ev[field]) changed = True if changed: updated += 1 else: unchanged += 1 # Reconcile deletions: a previously-synced event (has a caldav_uid) # that no longer appears remotely within the synced window is # soft-deleted, so a delete on the remote propagates locally instead # of orphaning forever. Guarded on a non-empty fetch so a spurious # empty result can't wipe every local copy. if remote_events: remote_uids = {e["caldav_uid"] for e in remote_events} orphan_batch = str(uuid.uuid4()) orphan_res = await session.execute( update(Event) .where( Event.user_id == user_id, Event.caldav_uid.isnot(None), Event.caldav_uid.notin_(remote_uids), Event.deleted_at.is_(None), Event.start_dt >= range_start, Event.start_dt <= range_end, ) .values(deleted_at=datetime.now(timezone.utc), deleted_batch_id=orphan_batch) ) deleted = orphan_res.rowcount or 0 await session.commit() elapsed = (datetime.now(timezone.utc) - started).total_seconds() logger.info( "CalDAV sync user %d: %d created, %d updated, %d unchanged, %d skipped (trashed), " "%d deleted (orphaned) in %.1fs", user_id, created, updated, unchanged, skipped, deleted, elapsed, ) return {"created": created, "updated": updated, "unchanged": unchanged, "skipped": skipped, "deleted": deleted} async def sync_all_users() -> None: """Pull CalDAV events for all users with CalDAV configured.""" from sqlalchemy import select as sa_select # noqa: PLC0415 from scribe.models.user import User # noqa: PLC0415 async with async_session() as session: result = await session.execute(sa_select(User.id)) user_ids = [row[0] for row in result.all()] for user_id in user_ids: try: await sync_user_events(user_id) except Exception: logger.warning("CalDAV sync failed for user %d", user_id, exc_info=True)