"""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 dataclasses import dataclass 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, completed_since: datetime | None = None) -> list[dict]: """Outstanding VTODOs from one CalDAV task list, plus -- when completed_since is given -- ones completed at or after that cutoff (see routers/common.py's get_or_refresh_tasks_for_widget, which passes "now - 24h" when TaskWidgetConfig.show_completed is on; None, the default, means completed tasks are dropped entirely, the original behavior). {"summary", "due" (ISO date/datetime string or None), "completed_at" (ISO datetime string, or None for an outstanding task)}, ... . Outstanding tasks sort first (by due date, no-due-date last), any included completed ones after (most recently completed first). 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 outstanding: list[dict] = [] completed: 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() summary = str(component.get("SUMMARY") or "(untitled)") if status == "COMPLETED": completed_prop = component.get("COMPLETED") completed_dt = completed_prop.dt if completed_prop is not None else None if completed_since is None or completed_dt is None or completed_dt < completed_since: continue completed.append({"summary": summary, "due": None, "completed_at": completed_dt.isoformat()}) else: due = component.get("DUE") outstanding.append({ "summary": summary, "due": due.dt.isoformat() if due is not None else None, "completed_at": None, }) outstanding.sort(key=lambda t: (t["due"] is None, t["due"] or "")) completed.sort(key=lambda t: t["completed_at"], reverse=True) return outstanding + completed @dataclass class TaskSource: """One task list to merge in -- CalDAV only, no ICS variant (a plain ICS subscription has no VTODO collection to speak of). owner_display_name tags every task pulled from this source so a merged checklist can show whose task is whose; color_index (2-5, into image_pipeline.DEFAULT_PALETTE_RGB) is this list's manually pinned color, or None for calendar_render.py's auto-cycle-by-owner- name fallback -- see models.FrameTaskList.""" owner_display_name: str url: str username: str password: str color_index: int | None = None def merge_tasks(sources: list[TaskSource], completed_since: datetime | None = None) -> tuple[list[dict], str]: """Fetches each source independently -- one broken list never blanks another's tasks. Returns (merged_tasks, fetch_summary); fetch_summary is "" when every source succeeded, else "N of M task lists unavailable" (same no-naming-names posture as calendar_feed. merge_events). No cross-list duplicate collapsing (unlike merge_events) -- a task synced to two lists at once is rare enough, and lower-stakes than a duplicated calendar event, not to be worth the same de-dup machinery.""" merged: list[dict] = [] failures = 0 for source in sources: try: tasks = fetch_tasks(source.url, source.username, source.password, completed_since=completed_since) except CalDavError: failures += 1 continue for task in tasks: merged.append({ **task, "owner_display_name": source.owner_display_name, "color_index": source.color_index, }) outstanding = [t for t in merged if t["completed_at"] is None] completed = [t for t in merged if t["completed_at"] is not None] outstanding.sort(key=lambda t: (t["due"] is None, t["due"] or "")) completed.sort(key=lambda t: t["completed_at"], reverse=True) summary = f"{failures} of {len(sources)} task lists unavailable" if failures else "" return outstanding + completed, summary