espresso_frame/server/app/caldav_client.py

"""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