espresso_frame/server/app/calendar_feed.py

"""Fetch, parse, and merge per-user calendar feeds -- ICS subscriptions
and (via caldav_client.py) CalDAV collections -- for calendar frame mode
(see routers/device.py's RENDERERS["calendar"] and calendar_render.py).

Pure functions -- no ORM, no FastAPI Depends. Callers (routers/common.py's
get_or_refresh_calendar_events) supply plain CalendarSource values, not
ORM objects, so this module stays testable against fixture .ics text with
no database or app involved.

Recurring events (RRULE/EXDATE/RDATE, DST-aware) are expanded via
recurring-ical-events rather than hand-rolled -- that's genuinely fiddly
to get right (see its own docs), not worth reinventing. It's LGPL-3.0 (an
ordinary runtime pip dependency, never vendored/modified -- see the
server README's Notes section for why that doesn't put this project's own
code under LGPL terms).
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime

import httpx
import icalendar
import recurring_ical_events

from . import caldav_client

HTTP_TIMEOUT_S = 15.0
FETCH_MAX_BYTES = 10 * 1024 * 1024  # sanity cap -- a real feed is KB, not MB

CHECK_INTERVAL_S = 20 * 60  # don't refetch/reparse any feed more often than this

# How far back/forward each merge-fetch expands recurring events. Households
# look back far less than they plan ahead, hence the asymmetry. Browsing
# outside this window (calendar_browse_offset) just yields an empty view,
# not an error -- self-heals on the next normal wake regardless.
EXPAND_WINDOW_PAST_DAYS = 30
EXPAND_WINDOW_FUTURE_DAYS = 200


class CalendarFetchError(Exception):
    """One feed was unreachable, not valid ICS, or too large. Raised by
    fetch_source_events(); merge_events() is what catches this per-source
    so one broken feed can't blank out another's events."""


def fetch_source_events(url: str, window_start: date, window_end: date) -> list[dict]:
    """One feed: download, parse, expand recurrences within
    [window_start, window_end]. Raises CalendarFetchError on any problem
    -- network, malformed ICS, or an oversized response."""
    try:
        with httpx.stream("GET", url, timeout=HTTP_TIMEOUT_S, follow_redirects=True) as resp:
            resp.raise_for_status()
            chunks = []
            total = 0
            for chunk in resp.iter_bytes():
                total += len(chunk)
                if total > FETCH_MAX_BYTES:
                    raise CalendarFetchError(f"Feed exceeds {FETCH_MAX_BYTES} bytes")
                chunks.append(chunk)
            body = b"".join(chunks)
    except httpx.HTTPError as e:
        raise CalendarFetchError(str(e)) from e

    try:
        cal = icalendar.Calendar.from_ical(body)
        occurrences = recurring_ical_events.of(cal).between(window_start, window_end)
    except Exception as e:  # icalendar/recurring_ical_events raise a mix of ValueError-family exceptions
        raise CalendarFetchError(f"Could not parse ICS feed: {e}") from e

    events = []
    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)  # date, not datetime -- VALUE=DATE
        events.append({
            "summary": str(occ.get("SUMMARY") or "(untitled)"),
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "all_day": all_day,
        })
    return events


@dataclass(frozen=True)
class CalendarSource:
    """One calendar to merge in: either a plain ICS subscription (kind
    "ics", url is the feed itself) or one CalDAV collection (kind
    "caldav", url is the calendar's own URL, username/password its
    account credentials) -- see caldav_client.py. owner_display_name
    tags every event pulled from this source so a merged agenda can show
    whose event is whose."""

    owner_display_name: str
    kind: str
    url: str
    username: str = ""
    password: str = ""


def merge_events(
    sources: list[CalendarSource], window_start: date, window_end: date
) -> tuple[list[dict], str]:
    """Fetches each source independently -- one broken feed never blanks
    another's events. Returns (merged_time_sorted_events, fetch_summary);
    fetch_summary is "" when every source succeeded, else "N of M
    calendars unavailable" (never *which* source -- naming whose feed is
    down to everyone who looks at a shared household display is a bigger
    overshare than the outage itself)."""
    merged: list[dict] = []
    failures = 0
    for source in sources:
        try:
            if source.kind == "caldav":
                events = caldav_client.fetch_calendar_events(
                    source.url, source.username, source.password, window_start, window_end
                )
            else:
                events = fetch_source_events(source.url, window_start, window_end)
        except (CalendarFetchError, caldav_client.CalDavError):
            failures += 1
            continue
        for event in events:
            event["owner_display_name"] = source.owner_display_name
            merged.append(event)

    merged.sort(key=lambda e: e["start"])
    summary = f"{failures} of {len(sources)} calendars unavailable" if failures else ""
    return merged, summary