"""Parse macro-book.ics and return events in a structured form.""" from __future__ import annotations import sys from datetime import date, datetime, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo from icalendar import Calendar VAULT = Path(__file__).resolve().parents[2] ICS_PATH = VAULT / "07-calendar" / "macro-book.ics" def load_calendar(ics_path: Path | str = ICS_PATH) -> Calendar: p = Path(ics_path) if not p.exists(): raise FileNotFoundError(f"No .ics at {p}") with open(p, "rb") as f: cal: Calendar = Calendar.from_ical(f.read().decode("utf-8")) # type: ignore[assignment] return cal def parse_events(ics_path: Path | str = ICS_PATH, window_start: datetime | None = None, window_end: datetime | None = None) -> list[dict]: """Return list of events with UTC start/end, summary, tier, categories. window_start/window_end: filter to events whose DTSTART is in [start, end). If None, no filtering. """ cal = load_calendar(ics_path) out = [] for component in cal.walk("VEVENT"): summary = str(component.get("SUMMARY", "")) dtstart = component.get("DTSTART").dt dtend = component.get("DTEND").dt # Normalize to datetime (date -> midnight) if isinstance(dtstart, date) and not isinstance(dtstart, datetime): dtstart = datetime(dtstart.year, dtstart.month, dtstart.day) if isinstance(dtend, date) and not isinstance(dtend, datetime): dtend = datetime(dtend.year, dtend.month, dtend.day) # Make timezone-aware (UTC default) if dtstart.tzinfo is None: dtstart = dtstart.replace(tzinfo=timezone.utc) if dtend.tzinfo is None: dtend = dtend.replace(tzinfo=timezone.utc) # CATEGORIES may carry tier as "tier-N" but in our generated ics # tier is encoded in the SUMMARY prefix [T1], [T2], [STRUCTURAL] cats = component.get("CATEGORIES") if cats is not None: cats_str = ",".join(c.strip() for c in cats.cats) if hasattr(cats, "cats") else str(cats) else: cats_str = "" tier = None for c in (cats_str.split(",") if cats_str else []): if c.startswith("tier-"): try: tier = int(c.split("-", 1)[1]) except ValueError: pass # Fall back to SUMMARY prefix if tier is None: sm = summary.strip() if sm.startswith("[T1]"): tier = 1 elif sm.startswith("[T2]"): tier = 2 elif sm.startswith("[STRUCTURAL]"): tier = "structural" # UID like "fomc-2026-07-29@macro-book" or "cot-friday@macro-book" uid = str(component.get("UID", "")) ev_id = uid.split("@", 1)[0] out.append({ "id": ev_id, "summary": summary, "dtstart_utc": dtstart, "dtend_utc": dtend, "tier": tier, "categories_raw": cats_str, }) if window_start is not None or window_end is not None: ws = window_start or datetime.min.replace(tzinfo=timezone.utc) we = window_end or datetime.max.replace(tzinfo=timezone.utc) out = [e for e in out if ws <= e["dtstart_utc"] < we] out.sort(key=lambda e: e["dtstart_utc"]) return out def events_for_date(d: date, ics_path: Path | str = ICS_PATH) -> list[dict]: """Return events whose start falls on the given date in Europe/Zurich.""" tz = ZoneInfo("Europe/Zurich") day_start = datetime(d.year, d.month, d.day, tzinfo=tz).astimezone(timezone.utc) day_end = day_start + timedelta(days=1) return parse_events(window_start=day_start, window_end=day_end, ics_path=ics_path) def events_in_window(window_start: datetime, window_end: datetime, ics_path: Path | str = ICS_PATH) -> list[dict]: return parse_events(window_start=window_start, window_end=window_end, ics_path=ics_path) def events_recently_ended(lookback_minutes: int = 70, since_minutes: int = 50, now: datetime | None = None, ics_path: Path | str = ICS_PATH) -> list[dict]: """Return events that ended in the [since, lookback] minutes before now. Default window: events ending 50-70 min ago — i.e. ~1h post-event, with a 20-minute wide window so the alerter catches the event on its 5-min cron even if a few ticks were missed. """ n = (now or datetime.now(timezone.utc)) end_window = n - timedelta(minutes=since_minutes) start_window = n - timedelta(minutes=lookback_minutes) cal = load_calendar(ics_path) out = [] for component in cal.walk("VEVENT"): dtend = component.get("DTEND").dt if isinstance(dtend, date) and not isinstance(dtend, datetime): dtend = datetime(dtend.year, dtend.month, dtend.day) if dtend.tzinfo is None: dtend = dtend.replace(tzinfo=timezone.utc) if start_window <= dtend <= end_window: summary = str(component.get("SUMMARY", "")) dtstart = component.get("DTSTART").dt if isinstance(dtstart, date) and not isinstance(dtstart, datetime): dtstart = datetime(dtstart.year, dtstart.month, dtstart.day) if dtstart.tzinfo is None: dtstart = dtstart.replace(tzinfo=timezone.utc) cats = component.get("CATEGORIES") cats_str = (",".join(c.strip() for c in cats.cats) if cats is not None and hasattr(cats, "cats") else "") tier = None for c in (cats_str.split(",") if cats_str else []): if c.startswith("tier-"): try: tier = int(c.split("-", 1)[1]) except ValueError: pass # Fall back to SUMMARY prefix (consistent with parse_events) if tier is None: sm = summary.strip() if sm.startswith("[T1]"): tier = 1 elif sm.startswith("[T2]"): tier = 2 elif sm.startswith("[STRUCTURAL]"): tier = "structural" uid = str(component.get("UID", "")) ev_id = uid.split("@", 1)[0] out.append({ "id": ev_id, "summary": summary, "dtstart_utc": dtstart, "dtend_utc": dtend, "tier": tier, }) return out if __name__ == "__main__": print(f"ICS: {ICS_PATH}") print(f" exists: {ICS_PATH.exists()}, size: {ICS_PATH.stat().st_size} bytes") events = parse_events() print(f"Total events: {len(events)}") print() print("Today (2026-06-16) in Zurich:") today = events_for_date(date(2026, 6, 16)) for e in today: zurich = e["dtstart_utc"].astimezone(ZoneInfo("Europe/Zurich")) print(f" {zurich.strftime('%H:%M %Z')} T{e['tier']} {e['summary']} ({e['id']})")