"""CalDAV account support: discovering which calendars an account exposes, and fetching one calendar's events or tasks -- the second way (alongside calendar_feed.py's single-file ICS subscription) a user can link a calendar for calendar frame mode (Nextcloud, Fastmail, iCloud, Radicale, Baikal, ...). Task lists (VTODO collections) are CalDAV-only -- a plain ICS subscription doesn't meaningfully have one -- see fetch_tasks. Thin wrapper around the `caldav` PyPI package (RFC 4791 client). NOTE ON LICENSING: `caldav` itself is dual-licensed GPL-3.0-or-later / Apache-2.0, but it hard-depends on `icalendar-searcher`, which is AGPL-3.0-or-later -- the strongest copyleft license in this project's dependency tree, and the one whose network-use clause is specifically written for server applications like this one. This was an explicit, informed call by the project owner to accept that exposure rather than hand-roll a CalDAV client -- see the server README's Notes section. Anyone redistributing this project (as opposed to just self-hosting it) should reread that tradeoff for their own situation. Pure functions -- no ORM, no FastAPI Depends -- same testability philosophy as calendar_feed.py. Event parsing/expansion reuses icalendar + recurring_ical_events directly (rather than trusting each CalDAV server's own possibly-inconsistent RRULE expansion) so a CalDAV calendar and an ICS subscription behave identically once fetched. """ from __future__ import annotations import logging from datetime import date, datetime import caldav import icalendar import recurring_ical_events logger = logging.getLogger(__name__) HTTP_TIMEOUT_S = 15 class CalDavError(Exception): """Discovery or fetch failed -- network, auth, or an unexpected server response. Raised loudly; callers (Settings' discover endpoint, calendar_feed.merge_events) decide what to do. Wraps whatever the caldav package/its transport raised, since that exception hierarchy isn't something call sites should need to know about directly.""" def discover_calendars(base_url: str, username: str, password: str) -> list[dict]: """[{"href": absolute_calendar_url, "display_name": str}, ...] for every calendar in this account. base_url is the server's CalDAV entry point (e.g. "https://cloud.example.com/remote.php/dav/" for Nextcloud) -- the caller supplies it directly, same idiom as the plain ICS subscription URL.""" try: client = caldav.DAVClient(url=base_url, username=username, password=password, timeout=HTTP_TIMEOUT_S) calendars = client.principal().calendars() except Exception as e: raise CalDavError(str(e)) from e result = [] for cal in calendars: try: display_name = cal.get_display_name() or cal.name except Exception: display_name = None result.append({"href": str(cal.url), "display_name": display_name or str(cal.url)}) return result def fetch_calendar_events(calendar_url: str, username: str, password: str, window_start: date, window_end: date) -> list[dict]: """One CalDAV calendar's events in [window_start, window_end] -- same event dict shape as calendar_feed.fetch_source_events (no "owner_display_name"; the caller adds that). Deliberately does NOT use the calendar-query REPORT's server-side time-range filter (caldav.Calendar.date_search) -- RFC 4791 leaves that corner case underspecified and real servers disagree on it (the caldav package's own docs warn "servers often behave differently when presented with a search request"; confirmed here too, once against a real server, as a calendar whose events just silently never came back despite discovery/auth both working fine). Instead this fetches every event in the calendar unfiltered (get_events() is a plain "list VEVENTs" REPORT with no time-range element -- the much more universally-supported case) and does 100% of the date-window filtering/recurrence-expansion client-side via icalendar + recurring_ical_events, exactly like calendar_feed.py already does for plain ICS feeds. Heavier per-fetch (the whole calendar, not just the window) but far more reliable.""" try: client = caldav.DAVClient(url=calendar_url, username=username, password=password, timeout=HTTP_TIMEOUT_S) calendar = caldav.Calendar(client=client, url=calendar_url) objects = calendar.get_events() except Exception as e: raise CalDavError(str(e)) from e events: list[dict] = [] for obj in objects: try: ical = icalendar.Calendar.from_ical(obj.data) occurrences = recurring_ical_events.of(ical).between(window_start, window_end) except Exception as e: # one malformed resource shouldn't blank the whole calendar logger.warning("Could not parse a CalDAV event from %s: %s", calendar_url, e) continue for occ in occurrences: dtstart = occ.get("DTSTART") dtend = occ.get("DTEND") if dtstart is None: continue start_dt = dtstart.dt end_dt = dtend.dt if dtend is not None else start_dt all_day = not isinstance(start_dt, datetime) events.append({ "summary": str(occ.get("SUMMARY") or "(untitled)"), "start": start_dt.isoformat(), "end": end_dt.isoformat(), "all_day": all_day, }) return events def fetch_tasks(calendar_url: str, username: str, password: str) -> list[dict]: """Outstanding (not-completed) VTODOs from one CalDAV task list, sorted by due date (tasks with no due date sort last). {"summary", "due" (ISO date/datetime string, or None)}, ... Fetches every task including completed ones and filters/sorts client-side rather than trusting get_todos()'s own include_completed/sort_keys server-side filtering, same reasoning as fetch_calendar_events not trusting the time-range REPORT filter -- a simpler filter than a time range, but not worth re-litigating which server-side filters are reliable one at a time.""" try: client = caldav.DAVClient(url=calendar_url, username=username, password=password, timeout=HTTP_TIMEOUT_S) calendar = caldav.Calendar(client=client, url=calendar_url) objects = calendar.get_todos(include_completed=True) except Exception as e: raise CalDavError(str(e)) from e tasks: list[dict] = [] for obj in objects: try: ical = icalendar.Calendar.from_ical(obj.data) except Exception as e: # one malformed resource shouldn't blank the whole list logger.warning("Could not parse a CalDAV task from %s: %s", calendar_url, e) continue for component in ical.walk("VTODO"): status = str(component.get("STATUS") or "NEEDS-ACTION").upper() if status == "COMPLETED": continue due = component.get("DUE") tasks.append({ "summary": str(component.get("SUMMARY") or "(untitled)"), "due": due.dt.isoformat() if due is not None else None, }) tasks.sort(key=lambda t: (t["due"] is None, t["due"] or "")) return tasks