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