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