"""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 from fabledassistant.models import async_session from fabledassistant.models.event import Event logger = logging.getLogger(__name__) _SYNC_PAST_DAYS = 30 _SYNC_FUTURE_DAYS = 180 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 fabledassistant.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) loop = asyncio.get_running_loop() try: remote_events: list[dict] = await loop.run_in_executor( None, _sync_one_user, config, user_id ) except Exception: logger.warning("CalDAV pull sync failed for user %d", user_id, exc_info=True) return {"error": "CalDAV fetch failed"} created = updated = unchanged = 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 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 await session.commit() logger.info( "CalDAV sync user %d: %d created, %d updated, %d unchanged", user_id, created, updated, unchanged, ) return {"created": created, "updated": updated, "unchanged": unchanged} 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 fabledassistant.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)