Nikolay Tatarinov
86f6d66865
All checks were successful
CI / lint-and-test (push) Successful in 17s
- Created a new `CHANGELOG.md` file to document all notable changes to the project, adhering to the Keep a Changelog format. - Updated `CONTRIBUTING.md` to include instructions for building and previewing documentation using MkDocs. - Added `mkdocs.yml` configuration for documentation generation, including navigation structure and theme settings. - Enhanced various documentation files, including API reference, architecture overview, configuration reference, and runbook, to provide comprehensive guidance for users and developers. - Included new sections in the README for changelog and documentation links, improving accessibility to project information.
134 lines
3.9 KiB
Python
134 lines
3.9 KiB
Python
"""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]:
|
|
"""Fetch ICS from URL and return events in the given date range.
|
|
|
|
Uses in-memory cache with TTL 7 days. Recurring events are skipped.
|
|
On fetch or parse error returns an empty list.
|
|
|
|
Args:
|
|
url: URL of the ICS calendar.
|
|
from_date: Start date YYYY-MM-DD.
|
|
to_date: End date YYYY-MM-DD.
|
|
|
|
Returns:
|
|
List of dicts with keys "date" (YYYY-MM-DD) and "summary". Empty on error.
|
|
"""
|
|
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)
|