from __future__ import annotations from datetime import date, datetime, time, timedelta, timezone from email.utils import format_datetime from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from dateutil.rrule import rruleset, rrulestr from icalendar import Alarm, Calendar, Event, vCalAddress from icalendar.parser import Parameters class ICalendarError(ValueError): """Raised when iCalendar content cannot be parsed as a usable VEVENT.""" GENERATED_EVENT_PROPERTIES = { "UID", "DTSTAMP", "DTSTART", "DTEND", "DURATION", "RECURRENCE-ID", "SUMMARY", "SEQUENCE", "STATUS", "TRANSP", "CLASS", "DESCRIPTION", "LOCATION", "CATEGORIES", "RRULE", "RDATE", "EXDATE", "ORGANIZER", "ATTENDEE", "ATTACH", "RELATED-TO", } class RawICalendarValue: def __init__(self, value: str, params: dict[str, str]) -> None: self.value = value self.params = Parameters(params) def to_ical(self) -> bytes: return self.value.encode("utf-8") def parse_vevent(ics: str) -> dict[str, Any]: events = parse_vevents(ics) if not events: raise ICalendarError("No VEVENT component found") return events[0] def parse_vevents(ics: str) -> list[dict[str, Any]]: try: calendar = Calendar.from_ical(ics) except Exception as exc: # noqa: BLE001 - normalize parser errors for API callers. raise ICalendarError(f"Invalid iCalendar content: {exc}") from exc events = [component for component in calendar.walk("VEVENT") if component.name == "VEVENT"] if not events: raise ICalendarError("No VEVENT component found") normalized_ics = normalize_ics(ics) return [parse_vevent_component(calendar, event, raw_ics=normalized_ics) for event in events] def parse_vevent_component(calendar: Calendar, event: Event, *, raw_ics: str) -> dict[str, Any]: uid = text_value(first_property_value(event, "UID")) if not uid: raise ICalendarError("VEVENT is missing UID") dtstart = first_property_value(event, "DTSTART") if dtstart is None: raise ICalendarError("VEVENT is missing DTSTART") start_at, all_day, tzid = temporal_property(dtstart) end_at = None duration_seconds = None dtend = first_property_value(event, "DTEND") duration = decoded_property(event, "DURATION") if dtend is not None: end_at, _end_all_day, _end_tzid = temporal_property(dtend) duration_seconds = int((end_at - start_at).total_seconds()) elif isinstance(duration, timedelta): end_at = start_at + duration duration_seconds = int(duration.total_seconds()) properties = component_property_records(event) alarms = [component_alarm_record(alarm) for alarm in event.subcomponents if alarm.name == "VALARM"] standard = standard_property_index(properties) return { "uid": uid, "recurrence_id": recurrence_id_value(first_property_value(event, "RECURRENCE-ID")), "sequence": int_or_zero(text_value(first_property_value(event, "SEQUENCE"))), "summary": text_value(first_property_value(event, "SUMMARY")) or "(Untitled event)", "description": text_value(first_property_value(event, "DESCRIPTION")), "location": text_value(first_property_value(event, "LOCATION")), "status": (text_value(first_property_value(event, "STATUS")) or "CONFIRMED").upper(), "transparency": (text_value(first_property_value(event, "TRANSP")) or "OPAQUE").upper(), "classification": (text_value(first_property_value(event, "CLASS")) or "PUBLIC").upper(), "start_at": start_at, "end_at": end_at, "duration_seconds": duration_seconds, "all_day": all_day, "timezone": tzid, "organizer": party_record(first_property_value(event, "ORGANIZER")), "attendees": [party_record(value) for value in property_values(event, "ATTENDEE")], "categories": category_values(event), "rrule": recurrence_rule(first_property_value(event, "RRULE")), "rdate": records_for_name(properties, "RDATE"), "exdate": records_for_name(properties, "EXDATE"), "reminders": [alarm_to_reminder(alarm) for alarm in alarms], "attachments": records_for_name(properties, "ATTACH"), "related_to": records_for_name(properties, "RELATED-TO"), "icalendar": { "component": "VEVENT", "schema_version": 1, "properties": properties, "alarms": alarms, "standard": standard, "method": text_value(first_property_value(calendar, "METHOD")), "timezones": calendar_timezones(calendar), }, "raw_ics": raw_ics, } def event_to_ics(event: Any) -> str: return events_to_ics([event]) def events_to_ics(events: list[Any]) -> str: calendar = Calendar() calendar.add("version", "2.0") calendar.add("prodid", "-//GovOPlaN//Calendar//EN") calendar.add("calscale", "GREGORIAN") method = next((metadata_value(event, "method") for event in events if metadata_value(event, "method")), None) if method: calendar.add("method", method) for event in events: calendar.add_component(event_to_component(event)) return calendar.to_ical().decode("utf-8") def event_to_component(event: Any) -> Event: component = Event() component.add("uid", event.uid) component.add("dtstamp", datetime.now(timezone.utc)) component.add("dtstart", event_temporal_value(event.start_at, all_day=event.all_day, timezone_id=event.timezone)) if event.end_at is not None: component.add("dtend", event_temporal_value(event.end_at, all_day=event.all_day, timezone_id=event.timezone)) elif getattr(event, "duration_seconds", None) is not None: component.add("duration", timedelta(seconds=int(event.duration_seconds))) if getattr(event, "recurrence_id", None): add_recurrence_id(component, str(event.recurrence_id)) component.add("summary", event.summary) component.add("sequence", int(event.sequence or 0)) component.add("status", event.status) component.add("transp", event.transparency) component.add("class", event.classification) if event.description: component.add("description", event.description) if event.location: component.add("location", event.location) if event.categories: component.add("categories", list(event.categories)) if event.rrule: component.add("rrule", normalized_rrule_for_export(event.rrule)) if event.organizer: component.add("organizer", party_for_export(event.organizer)) for attendee in event.attendees or []: component.add("attendee", party_for_export(attendee)) for record in getattr(event, "rdate", None) or []: add_raw_property(component, "RDATE", record) for record in getattr(event, "exdate", None) or []: add_raw_property(component, "EXDATE", record) for record in getattr(event, "attachments", None) or []: add_raw_property(component, "ATTACH", record) for record in getattr(event, "related_to", None) or []: add_raw_property(component, "RELATED-TO", record) add_preserved_properties(component, getattr(event, "icalendar", None) or {}) for alarm in alarms_for_export(event): component.add_component(alarm) return component def expand_event_occurrences(event: Any, range_start: datetime, range_end: datetime, *, limit: int = 1000) -> list[dict[str, Any]]: """Expand an event's recurrence primitives within a range. This is a backend primitive for later API/UI recurrence handling. It expands RRULE/RDATE/EXDATE data stored by this module and returns occurrence starts with the event's original duration. """ start_at = normalize_datetime(event.start_at) end_at = normalize_datetime(event.end_at) if event.end_at is not None else None range_start = normalize_datetime(range_start) range_end = normalize_datetime(range_end) duration = occurrence_duration(event, start_at=start_at, end_at=end_at) if not event.rrule and not event.rdate: return [occurrence_payload(start_at, duration, event)] if occurrence_overlaps(start_at, duration, range_start, range_end) else [] rules = rruleset() if event.rrule: rules.rrule(rrulestr(serialize_key_value_property(event.rrule), dtstart=start_at)) for rdate in temporal_values_from_records(event.rdate or []): rules.rdate(rdate) for exdate in temporal_values_from_records(event.exdate or []): rules.exdate(exdate) starts = rules.between(range_start - duration, range_end, inc=True) occurrences: list[dict[str, Any]] = [] for occurrence_start in starts: occurrence_start = normalize_datetime(occurrence_start) if occurrence_overlaps(occurrence_start, duration, range_start, range_end): occurrences.append(occurrence_payload(occurrence_start, duration, event)) if len(occurrences) >= limit: break return occurrences def first_vevent(calendar: Calendar) -> Event: for component in calendar.walk("VEVENT"): if component.name == "VEVENT": return component raise ICalendarError("No VEVENT component found") def first_property_value(component: Any, name: str) -> Any | None: values = property_values(component, name) return values[0] if values else None def property_values(component: Any, name: str) -> list[Any]: value = component.get(name.upper()) if value is None: return [] return value if isinstance(value, list) else [value] def decoded_property(component: Any, name: str) -> Any | None: try: return component.decoded(name.upper()) except KeyError: return None def temporal_property(value: Any) -> tuple[datetime, bool, str | None]: decoded = decoded_temporal_value(value) params = params_to_json(getattr(value, "params", {})) tzid = first_param(params, "TZID") if isinstance(decoded, datetime): return normalize_datetime(decoded), False, tzid if isinstance(decoded, date): return datetime.combine(decoded, time.min, tzinfo=timezone.utc), True, tzid raise ICalendarError(f"Unsupported temporal value: {value}") def decoded_temporal_value(value: Any) -> Any: decoded = getattr(value, "dt", None) return decoded if decoded is not None else value def recurrence_id_value(value: Any | None) -> str | None: if value is None: return None raw_value = ical_value(value) params = params_to_json(getattr(value, "params", {})) tzid = first_param(params, "TZID") return f"TZID={tzid}:{raw_value}" if tzid else raw_value def recurrence_rule(value: Any | None) -> dict[str, Any] | None: if value is None: return None if hasattr(value, "items"): result: dict[str, Any] = {} for key, raw_items in value.items(): items = raw_items if isinstance(raw_items, list) else [raw_items] normalized = [rrule_item(item) for item in items] result[str(key).upper()] = normalized if len(normalized) > 1 else normalized[0] return result return parse_key_value_property(ical_value(value)) def rrule_item(value: Any) -> str: if isinstance(value, datetime): return format_ical_datetime(value) if isinstance(value, date): return value.strftime("%Y%m%d") return str(value).upper() if isinstance(value, str) else str(value) def category_values(component: Any) -> list[str]: decoded = decoded_property(component, "CATEGORIES") if decoded is None: return [] groups = decoded if isinstance(decoded, list) else [decoded] categories: list[str] = [] for group in groups: values = group if isinstance(group, list) else [group] categories.extend(str(value) for value in values if str(value)) return categories def component_property_records(component: Any) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for name, value in component.items(): for item in value if isinstance(value, list) else [value]: records.append(property_record(name, item)) return records def property_record(name: str, value: Any) -> dict[str, Any]: record: dict[str, Any] = { "name": str(name).upper(), "value": ical_value(value), "params": params_to_json(getattr(value, "params", {})), } decoded = decoded_temporal_value(value) json_decoded = json_value(decoded) if json_decoded is not None and json_decoded != record["value"]: record["decoded"] = json_decoded return record def component_alarm_record(alarm: Any) -> dict[str, Any]: return { "component": "VALARM", "properties": component_property_records(alarm), } def alarm_to_reminder(alarm: dict[str, Any]) -> dict[str, Any]: properties = alarm.get("properties") or [] trigger = first_record(properties, "TRIGGER") return { "action": record_value(properties, "ACTION") or "DISPLAY", "trigger": trigger, "description": record_value(properties, "DESCRIPTION"), "summary": record_value(properties, "SUMMARY"), "attendees": records_for_name(properties, "ATTENDEE"), "duration": first_record(properties, "DURATION"), "repeat": record_value(properties, "REPEAT"), "attachments": records_for_name(properties, "ATTACH"), "properties": properties, } def standard_property_index(properties: list[dict[str, Any]]) -> dict[str, Any]: names = ( "DTSTAMP", "CREATED", "LAST-MODIFIED", "PRIORITY", "URL", "GEO", "COMMENT", "CONTACT", "REQUEST-STATUS", "RESOURCES", ) result: dict[str, Any] = {} for name in names: matching = records_for_name(properties, name) if not matching: continue result[name.lower().replace("-", "_")] = matching if len(matching) > 1 else matching[0] return result def records_for_name(records: list[dict[str, Any]], name: str) -> list[dict[str, Any]]: return [record for record in records if record.get("name") == name.upper()] def first_record(records: list[dict[str, Any]], name: str) -> dict[str, Any] | None: matching = records_for_name(records, name) return matching[0] if matching else None def record_value(records: list[dict[str, Any]], name: str) -> str | None: record = first_record(records, name) if not record: return None return str(record.get("value") or "") def party_record(value: Any | None) -> dict[str, Any] | None: if value is None: return None return {"value": text_value(value) or "", "params": params_to_json(getattr(value, "params", {}))} def party_for_export(party: dict[str, Any]) -> Any: value = str(party.get("value") or party.get("email") or "") cal_address = vCalAddress(value) for key, item in flattened_params(party.get("params") if isinstance(party.get("params"), dict) else {}).items(): cal_address.params[key] = item return cal_address def text_value(value: Any | None) -> str | None: if value is None: return None raw = value.decode("utf-8") if isinstance(value, bytes) else str(value) return raw if raw != "" else None def ical_value(value: Any) -> str: if hasattr(value, "to_ical"): raw = value.to_ical() return raw.decode("utf-8") if isinstance(raw, bytes) else str(raw) if isinstance(value, bytes): return value.decode("utf-8") return str(value) def params_to_json(params: Any) -> dict[str, list[str]]: result: dict[str, list[str]] = {} for key, value in dict(params or {}).items(): values = value if isinstance(value, list) else [value] result[str(key).upper()] = [str(item) for item in values] return result def flattened_params(params: dict[str, Any]) -> dict[str, str]: result: dict[str, str] = {} for key, value in (params or {}).items(): values = value if isinstance(value, list) else [value] if values: result[str(key).upper()] = ",".join(str(item) for item in values) return result def first_param(params: dict[str, list[str]], name: str) -> str | None: values = params.get(name.upper()) or [] return values[0] if values else None def json_value(value: Any) -> Any: if isinstance(value, datetime): return normalize_datetime(value).isoformat() if isinstance(value, date): return value.isoformat() if isinstance(value, timedelta): return int(value.total_seconds()) if isinstance(value, (str, int, float, bool)) or value is None: return value return None def calendar_timezones(calendar: Calendar) -> list[str]: timezones: list[str] = [] for component in calendar.walk("VTIMEZONE"): tzid = text_value(first_property_value(component, "TZID")) if tzid and tzid not in timezones: timezones.append(tzid) return timezones def metadata_value(event: Any, key: str) -> str | None: raw_metadata = getattr(event, "icalendar", None) metadata = raw_metadata if isinstance(raw_metadata, dict) else {} value = metadata.get(key) return str(value) if value else None def event_temporal_value(value: datetime, *, all_day: bool, timezone_id: str | None) -> date | datetime: value = normalize_datetime(value) if all_day: return value.date() if timezone_id: try: return value.astimezone(ZoneInfo(timezone_id)) except ZoneInfoNotFoundError: return value return value def add_raw_property(component: Event | Alarm, default_name: str, record: dict[str, Any]) -> None: name = str(record.get("name") or default_name).upper() value = record.get("value") if not name or value is None: return raw_value = RawICalendarValue(str(value), flattened_params(record.get("params") or {})) component.add(name, raw_value, encode=False) def add_recurrence_id(component: Event, value: str) -> None: params: dict[str, list[str]] = {} raw_value = value if value.startswith("TZID=") and ":" in value: tzid, raw_value = value.removeprefix("TZID=").split(":", 1) params["TZID"] = [tzid] add_raw_property(component, "RECURRENCE-ID", {"name": "RECURRENCE-ID", "value": raw_value, "params": params}) def add_preserved_properties(component: Event, metadata: dict[str, Any]) -> None: for record in metadata.get("properties") or []: if not isinstance(record, dict): continue name = str(record.get("name") or "").upper() if not name or name in GENERATED_EVENT_PROPERTIES: continue add_raw_property(component, name, record) def alarms_for_export(event: Any) -> list[Alarm]: reminders = getattr(event, "reminders", None) or [] raw_metadata = getattr(event, "icalendar", None) if not reminders and isinstance(raw_metadata, dict): reminders = raw_metadata.get("alarms") or [] alarms: list[Alarm] = [] for reminder in reminders: if not isinstance(reminder, dict): continue alarm = Alarm() properties = reminder.get("properties") if isinstance(properties, list): for record in properties: if isinstance(record, dict): add_raw_property(alarm, str(record.get("name") or ""), record) else: alarm.add("action", reminder.get("action") or "DISPLAY") trigger = reminder.get("trigger") if isinstance(trigger, dict): add_raw_property(alarm, "TRIGGER", trigger) if reminder.get("description"): alarm.add("description", reminder["description"]) if reminder.get("summary"): alarm.add("summary", reminder["summary"]) if alarm: alarms.append(alarm) return alarms def normalized_rrule_for_export(value: dict[str, Any]) -> dict[str, list[Any]]: result: dict[str, list[Any]] = {} for key, item in value.items(): result[str(key).lower()] = item if isinstance(item, list) else [item] return result def parse_key_value_property(value: str | None) -> dict[str, Any] | None: if not value: return None result: dict[str, Any] = {} for item in value.split(";"): key, sep, raw = item.partition("=") if sep: values = raw.split(",") if "," in raw else [raw] result[key.upper()] = values if len(values) > 1 else values[0] return result or {"raw": value} def serialize_key_value_property(value: dict[str, Any]) -> str: parts: list[str] = [] for key, item in value.items(): if isinstance(item, list): parts.append(f"{key.upper()}={','.join(str(entry) for entry in item)}") else: parts.append(f"{key.upper()}={item}") return ";".join(parts) def temporal_values_from_records(records: list[dict[str, Any]]) -> list[datetime]: values: list[datetime] = [] for record in records: params = record.get("params") if isinstance(record.get("params"), dict) else {} for raw_value in str(record.get("value") or "").split(","): parsed = parse_ical_temporal(raw_value, params) if parsed is not None: values.append(parsed) return values def parse_ical_temporal(value: str, params: dict[str, Any]) -> datetime | None: if not value: return None value_type = first_param(params_to_json(params), "VALUE") if value_type == "DATE" or (len(value) == 8 and "T" not in value): return datetime.combine(datetime.strptime(value, "%Y%m%d").date(), time.min, tzinfo=timezone.utc) try: if value.endswith("Z"): return datetime.strptime(value, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) parsed = datetime.strptime(value, "%Y%m%dT%H%M%S") except ValueError: return None tzid = first_param(params_to_json(params), "TZID") if tzid: try: return parsed.replace(tzinfo=ZoneInfo(tzid)).astimezone(timezone.utc) except ZoneInfoNotFoundError: return parsed.replace(tzinfo=timezone.utc) return parsed.replace(tzinfo=timezone.utc) def occurrence_duration(event: Any, *, start_at: datetime, end_at: datetime | None) -> timedelta: if end_at is not None: return end_at - start_at if event.duration_seconds is not None: return timedelta(seconds=int(event.duration_seconds)) return timedelta(days=1) if event.all_day else timedelta(0) def occurrence_overlaps(start_at: datetime, duration: timedelta, range_start: datetime, range_end: datetime) -> bool: end_at = start_at + duration return end_at >= range_start and start_at <= range_end def occurrence_payload(start_at: datetime, duration: timedelta, event: Any) -> dict[str, Any]: return { "uid": event.uid, "recurrence_id": format_ical_datetime(start_at), "start_at": start_at, "end_at": start_at + duration if duration else None, "all_day": event.all_day, } def int_or_zero(value: str | None) -> int: try: return int(value or "0") except ValueError: return 0 def normalize_datetime(value: datetime) -> datetime: if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def normalize_ics(ics: str) -> str: stripped = ics.strip().replace("\r\n", "\n").replace("\r", "\n") return "\r\n".join(stripped.split("\n")) + "\r\n" def format_ical_datetime(value: datetime) -> str: return normalize_datetime(value).strftime("%Y%m%dT%H%M%SZ") def http_last_modified(value: datetime) -> str: return format_datetime(normalize_datetime(value), usegmt=True)