Implement external calendar integration and enhance API functionality
- Added support for fetching and parsing external ICS calendars, allowing events to be displayed on the duty grid. - Introduced a new API endpoint `/api/calendar-events` to retrieve calendar events within a specified date range. - Updated configuration to include `EXTERNAL_CALENDAR_ICS_URL` for specifying the ICS calendar URL. - Enhanced the web application to visually indicate days with events and provide event summaries on hover. - Improved documentation in the README to include details about the new calendar integration and configuration options. - Updated tests to cover the new calendar functionality and ensure proper integration.
This commit is contained in:
124
api/calendar_ics.py
Normal file
124
api/calendar_ics.py
Normal file
@@ -0,0 +1,124 @@
|
||||
"""Fetch and parse external ICS calendar; in-memory cache with 7-day TTL."""
|
||||
|
||||
import logging
|
||||
from datetime import date, datetime, timedelta
|
||||
from urllib.request import Request, urlopen
|
||||
from urllib.error import URLError
|
||||
|
||||
from icalendar import Calendar
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# In-memory cache: url -> (cached_at_timestamp, raw_ics_bytes)
|
||||
_ics_cache: dict[str, tuple[float, bytes]] = {}
|
||||
CACHE_TTL_SECONDS = 7 * 24 * 3600 # 1 week
|
||||
FETCH_TIMEOUT_SECONDS = 15
|
||||
|
||||
|
||||
def _fetch_ics(url: str) -> bytes | None:
|
||||
"""GET url, return response body or None on error."""
|
||||
try:
|
||||
req = Request(url, headers={"User-Agent": "DutyTeller/1.0"})
|
||||
with urlopen(req, timeout=FETCH_TIMEOUT_SECONDS) as resp:
|
||||
return resp.read()
|
||||
except URLError as e:
|
||||
log.warning("Failed to fetch ICS from %s: %s", url, e)
|
||||
return None
|
||||
except OSError as e:
|
||||
log.warning("Error fetching ICS from %s: %s", url, e)
|
||||
return None
|
||||
|
||||
|
||||
def _to_date(dt) -> date | None:
|
||||
"""Convert icalendar DATE or DATE-TIME to date. Return None if invalid."""
|
||||
if isinstance(dt, datetime):
|
||||
return dt.date()
|
||||
if isinstance(dt, date):
|
||||
return dt
|
||||
return None
|
||||
|
||||
|
||||
def _event_date_range(component) -> tuple[date | None, date | None]:
|
||||
"""
|
||||
Get (start_date, end_date) for a VEVENT. DTEND is exclusive in iCalendar;
|
||||
last day of event = DTEND date - 1 day. Returns (None, None) if invalid.
|
||||
"""
|
||||
dtstart = component.get("dtstart")
|
||||
if not dtstart:
|
||||
return (None, None)
|
||||
start_d = _to_date(dtstart.dt)
|
||||
if not start_d:
|
||||
return (None, None)
|
||||
|
||||
dtend = component.get("dtend")
|
||||
if not dtend:
|
||||
return (start_d, start_d)
|
||||
|
||||
end_dt = dtend.dt
|
||||
end_d = _to_date(end_dt)
|
||||
if not end_d:
|
||||
return (start_d, start_d)
|
||||
# DTEND is exclusive: last day of event is end_d - 1 day
|
||||
last_d = end_d - timedelta(days=1)
|
||||
return (start_d, last_d)
|
||||
|
||||
|
||||
def _get_events_from_ics(raw: bytes, from_date: str, to_date: str) -> list[dict]:
|
||||
"""Parse ICS bytes and return list of {date, summary} in [from_date, to_date]. One-time events only."""
|
||||
result: list[dict] = []
|
||||
try:
|
||||
cal = Calendar.from_ical(raw)
|
||||
if not cal:
|
||||
return result
|
||||
except Exception as e:
|
||||
log.warning("Failed to parse ICS: %s", e)
|
||||
return result
|
||||
|
||||
from_d = date.fromisoformat(from_date)
|
||||
to_d = date.fromisoformat(to_date)
|
||||
|
||||
for component in cal.walk():
|
||||
if component.name != "VEVENT":
|
||||
continue
|
||||
if component.get("rrule"):
|
||||
continue # skip recurring in first iteration
|
||||
start_d, end_d = _event_date_range(component)
|
||||
if not start_d or not end_d:
|
||||
continue
|
||||
summary = component.get("summary")
|
||||
summary_str = str(summary) if summary else ""
|
||||
|
||||
d = start_d
|
||||
while d <= end_d:
|
||||
if from_d <= d <= to_d:
|
||||
result.append({"date": d.strftime("%Y-%m-%d"), "summary": summary_str})
|
||||
d += timedelta(days=1)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def get_calendar_events(
|
||||
url: str,
|
||||
from_date: str,
|
||||
to_date: str,
|
||||
) -> list[dict]:
|
||||
"""
|
||||
Return list of {date: "YYYY-MM-DD", summary: "..."} for events in [from_date, to_date].
|
||||
Uses in-memory cache with TTL 7 days. On fetch/parse error returns [].
|
||||
"""
|
||||
if not url or from_date > to_date:
|
||||
return []
|
||||
|
||||
now = datetime.now().timestamp()
|
||||
raw: bytes | None = None
|
||||
if url in _ics_cache:
|
||||
cached_at, cached_raw = _ics_cache[url]
|
||||
if now - cached_at < CACHE_TTL_SECONDS:
|
||||
raw = cached_raw
|
||||
if raw is None:
|
||||
raw = _fetch_ics(url)
|
||||
if raw is None:
|
||||
return []
|
||||
_ics_cache[url] = (now, raw)
|
||||
|
||||
return _get_events_from_ics(raw, from_date, to_date)
|
||||
Reference in New Issue
Block a user