189 lines
6.7 KiB
Python
189 lines
6.7 KiB
Python
"""ICS (iCalendar) import/export provider (U8).
|
|
|
|
Uses the ``icalendar`` library for RFC 5545 compliant parsing/serialization.
|
|
Import delegates to ``calendar.db.insert_event`` for direct persistence with
|
|
``external_id`` set (so duplicate UIDs can be skipped on re-import).
|
|
Export reads via ``CalendarService.list_events``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import date, datetime, timezone
|
|
from typing import Any
|
|
|
|
from icalendar import Calendar, Event
|
|
from icalendar.prop import vRecur
|
|
|
|
from agentkit.calendar.db import get_event_by_external_id, insert_event
|
|
from agentkit.calendar.models import CalendarEvent, _now_iso
|
|
from agentkit.calendar.service import CalendarService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _to_iso_utc(dt: datetime) -> str:
|
|
"""Convert a datetime to ISO 8601 UTC string."""
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc).isoformat()
|
|
|
|
|
|
def _parse_iso(dt_str: str) -> datetime:
|
|
"""Parse an ISO 8601 string to a UTC-aware datetime."""
|
|
dt = datetime.fromisoformat(dt_str)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def _extract_dt(component: Any, key: str) -> tuple[str, bool]:
|
|
"""Extract a date/datetime property from an icalendar component.
|
|
|
|
Returns ``(iso_string, is_all_day)``. ``is_all_day`` is True when the
|
|
value is a bare ``date`` (not a ``datetime``).
|
|
"""
|
|
prop = component.get(key)
|
|
if prop is None:
|
|
return "", False
|
|
val = prop.dt
|
|
if isinstance(val, date) and not isinstance(val, datetime):
|
|
return val.isoformat(), True
|
|
return _to_iso_utc(val), False
|
|
|
|
|
|
class ICSProvider:
|
|
"""Import/export calendar events as iCalendar (.ics) files."""
|
|
|
|
def __init__(self, service: CalendarService) -> None:
|
|
self.service = service
|
|
|
|
async def import_ics(self, ics_bytes: bytes, user_id: str) -> dict[str, Any]:
|
|
"""Parse ICS bytes and create events for *user_id*.
|
|
|
|
Returns ``{"imported": N, "skipped": M, "errors": [...]}``.
|
|
Raises ``ValueError`` if the ICS content cannot be parsed at all.
|
|
"""
|
|
# ponytail: hard size/count caps to prevent DoS via crafted ICS.
|
|
# Upgrade: stream-parse for unbounded inputs.
|
|
if len(ics_bytes) > 10 * 1024 * 1024:
|
|
raise ValueError("ICS file too large (max 10MB)")
|
|
|
|
imported = 0
|
|
skipped = 0
|
|
errors: list[str] = []
|
|
|
|
try:
|
|
cal = Calendar.from_ical(ics_bytes)
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to parse ICS: {e}") from e
|
|
|
|
components = list(cal.walk("VEVENT"))
|
|
# ponytail: hard size/count caps to prevent DoS via crafted ICS.
|
|
if len(components) > 10000:
|
|
raise ValueError("Too many events in ICS file (max 10000)")
|
|
|
|
for component in components:
|
|
try:
|
|
uid = str(component.get("UID", "") or "") or None
|
|
|
|
# Dedup by (external_id, provider, user_id)
|
|
if uid:
|
|
existing = await get_event_by_external_id(
|
|
uid, "ics", user_id, self.service.db_path
|
|
)
|
|
if existing is not None:
|
|
skipped += 1
|
|
continue
|
|
|
|
title = str(component.get("SUMMARY", "") or "")
|
|
if not title:
|
|
errors.append("VEVENT missing SUMMARY, skipped")
|
|
continue
|
|
|
|
start_str, is_all_day = _extract_dt(component, "DTSTART")
|
|
if not start_str:
|
|
errors.append("VEVENT missing DTSTART, skipped")
|
|
continue
|
|
end_str, _ = _extract_dt(component, "DTEND")
|
|
if not end_str:
|
|
end_str = start_str
|
|
|
|
description = str(component.get("DESCRIPTION", "") or "")
|
|
location = str(component.get("LOCATION", "") or "")
|
|
|
|
rrule_str: str | None = None
|
|
rrule = component.get("RRULE")
|
|
if rrule is not None:
|
|
rrule_str = rrule.to_ical().decode("utf-8")
|
|
|
|
now = _now_iso()
|
|
event = CalendarEvent(
|
|
id=uuid.uuid4().hex,
|
|
user_id=user_id,
|
|
title=title,
|
|
description=description,
|
|
start_time=start_str,
|
|
end_time=end_str,
|
|
is_all_day=is_all_day,
|
|
location=location,
|
|
rrule=rrule_str,
|
|
source="manual",
|
|
external_id=uid,
|
|
external_provider="ics" if uid else None,
|
|
last_modified=now,
|
|
created_at=now,
|
|
)
|
|
await insert_event(event, self.service.db_path)
|
|
imported += 1
|
|
except Exception as e:
|
|
errors.append(f"Failed to import VEVENT: {e}")
|
|
logger.warning("ICS import error: %s", e)
|
|
|
|
return {"imported": imported, "skipped": skipped, "errors": errors}
|
|
|
|
def export_ics(self, events: list[CalendarEvent]) -> bytes:
|
|
"""Serialize *events* to ICS bytes."""
|
|
cal = Calendar()
|
|
cal.add("prodid", "-//Fischer AgentKit//Calendar//EN")
|
|
cal.add("version", "2.0")
|
|
|
|
# ponytail: list_events expands RRULE into occurrences (same event.id).
|
|
# ICS wants one VEVENT with RRULE, not N copies — dedup by id.
|
|
seen_ids: set[str] = set()
|
|
for event in events:
|
|
if event.id in seen_ids:
|
|
continue
|
|
seen_ids.add(event.id)
|
|
cal.add_component(self._event_to_vevent(event))
|
|
|
|
return cal.to_ical()
|
|
|
|
def _event_to_vevent(self, event: CalendarEvent) -> Event:
|
|
"""Convert a :class:`CalendarEvent` to an icalendar ``Event`` component."""
|
|
vevent = Event()
|
|
vevent.add("uid", event.external_id or event.id)
|
|
vevent.add("summary", event.title)
|
|
|
|
start_dt = _parse_iso(event.start_time)
|
|
end_dt = _parse_iso(event.end_time)
|
|
|
|
if event.is_all_day:
|
|
vevent.add("dtstart", start_dt.date())
|
|
vevent.add("dtend", end_dt.date())
|
|
else:
|
|
vevent.add("dtstart", start_dt)
|
|
vevent.add("dtend", end_dt)
|
|
|
|
if event.description:
|
|
vevent.add("description", event.description)
|
|
if event.location:
|
|
vevent.add("location", event.location)
|
|
if event.rrule:
|
|
# ponytail: vRecur.from_ical reorders keys (e.g. COUNT before BYDAY)
|
|
# but the result is semantically equivalent RFC 5545.
|
|
vevent.add("rrule", vRecur.from_ical(event.rrule))
|
|
|
|
return vevent
|