Initialize GovOPlaN calendar module
This commit is contained in:
309
src/govoplan_calendar/backend/ical.py
Normal file
309
src/govoplan_calendar/backend/ical.py
Normal file
@@ -0,0 +1,309 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, time, timezone
|
||||
from email.utils import format_datetime
|
||||
from typing import Any
|
||||
|
||||
|
||||
class ICalendarError(ValueError):
|
||||
"""Raised when iCalendar content cannot be parsed as a usable VEVENT."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ICalendarProperty:
|
||||
name: str
|
||||
value: str
|
||||
params: dict[str, list[str]]
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {"name": self.name, "value": self.value, "params": self.params}
|
||||
|
||||
|
||||
def parse_vevent(ics: str) -> dict[str, Any]:
|
||||
lines = unfold_ical_lines(ics)
|
||||
event_lines = component_lines(lines, "VEVENT")
|
||||
if not event_lines:
|
||||
raise ICalendarError("No VEVENT component found")
|
||||
|
||||
properties = [parse_content_line(line) for line in event_lines]
|
||||
props = [prop.as_dict() for prop in properties]
|
||||
by_name = group_properties(properties)
|
||||
|
||||
uid = first_value(by_name, "UID")
|
||||
if not uid:
|
||||
raise ICalendarError("VEVENT is missing UID")
|
||||
dtstart_prop = first_property(by_name, "DTSTART")
|
||||
if dtstart_prop is None:
|
||||
raise ICalendarError("VEVENT is missing DTSTART")
|
||||
|
||||
start_at, all_day, tzid = parse_temporal_property(dtstart_prop)
|
||||
end_at = None
|
||||
duration_seconds = None
|
||||
dtend_prop = first_property(by_name, "DTEND")
|
||||
if dtend_prop is not None:
|
||||
end_at, _end_all_day, _end_tzid = parse_temporal_property(dtend_prop)
|
||||
duration_seconds = int((end_at - start_at).total_seconds())
|
||||
elif first_value(by_name, "DURATION"):
|
||||
# Keep full DURATION semantics in raw properties. The first module
|
||||
# stores it but does not yet calculate every RFC 5545 duration form.
|
||||
duration_seconds = None
|
||||
|
||||
summary = first_value(by_name, "SUMMARY") or "(Untitled event)"
|
||||
organizer = property_party(first_property(by_name, "ORGANIZER"))
|
||||
attendees = [property_party(prop) for prop in by_name.get("ATTENDEE", [])]
|
||||
categories = split_csv(first_value(by_name, "CATEGORIES") or "")
|
||||
|
||||
return {
|
||||
"uid": uid,
|
||||
"recurrence_id": first_value(by_name, "RECURRENCE-ID"),
|
||||
"sequence": int_or_zero(first_value(by_name, "SEQUENCE")),
|
||||
"summary": summary,
|
||||
"description": first_value(by_name, "DESCRIPTION"),
|
||||
"location": first_value(by_name, "LOCATION"),
|
||||
"status": (first_value(by_name, "STATUS") or "CONFIRMED").upper(),
|
||||
"transparency": (first_value(by_name, "TRANSP") or "OPAQUE").upper(),
|
||||
"classification": (first_value(by_name, "CLASS") or "PUBLIC").upper(),
|
||||
"start_at": start_at,
|
||||
"end_at": end_at,
|
||||
"duration_seconds": duration_seconds,
|
||||
"all_day": all_day,
|
||||
"timezone": tzid,
|
||||
"organizer": organizer,
|
||||
"attendees": attendees,
|
||||
"categories": categories,
|
||||
"rrule": parse_key_value_property(first_value(by_name, "RRULE")),
|
||||
"rdate": [prop.as_dict() for prop in by_name.get("RDATE", [])],
|
||||
"exdate": [prop.as_dict() for prop in by_name.get("EXDATE", [])],
|
||||
"attachments": [prop.as_dict() for prop in by_name.get("ATTACH", [])],
|
||||
"related_to": [prop.as_dict() for prop in by_name.get("RELATED-TO", [])],
|
||||
"icalendar": {"component": "VEVENT", "properties": props},
|
||||
"raw_ics": normalize_ics(ics),
|
||||
}
|
||||
|
||||
|
||||
def event_to_ics(event: Any) -> str:
|
||||
properties = [
|
||||
("UID", {}, event.uid),
|
||||
("DTSTAMP", {}, format_ical_datetime(datetime.now(timezone.utc))),
|
||||
("DTSTART", date_params(event), format_ical_temporal(event.start_at, all_day=event.all_day)),
|
||||
]
|
||||
if event.end_at is not None:
|
||||
properties.append(("DTEND", date_params(event), format_ical_temporal(event.end_at, all_day=event.all_day)))
|
||||
properties.extend(
|
||||
[
|
||||
("SUMMARY", {}, event.summary),
|
||||
("SEQUENCE", {}, str(event.sequence or 0)),
|
||||
("STATUS", {}, event.status),
|
||||
("TRANSP", {}, event.transparency),
|
||||
("CLASS", {}, event.classification),
|
||||
]
|
||||
)
|
||||
if event.description:
|
||||
properties.append(("DESCRIPTION", {}, event.description))
|
||||
if event.location:
|
||||
properties.append(("LOCATION", {}, event.location))
|
||||
for category in event.categories or []:
|
||||
if category:
|
||||
properties.append(("CATEGORIES", {}, category))
|
||||
if event.rrule:
|
||||
properties.append(("RRULE", {}, serialize_key_value_property(event.rrule)))
|
||||
if event.organizer:
|
||||
properties.append(party_to_property("ORGANIZER", event.organizer))
|
||||
for attendee in event.attendees or []:
|
||||
properties.append(party_to_property("ATTENDEE", attendee))
|
||||
|
||||
lines = ["BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//GovOPlaN//Calendar//EN", "CALSCALE:GREGORIAN", "BEGIN:VEVENT"]
|
||||
lines.extend(fold_ical_line(serialize_content_line(name, params, value)) for name, params, value in properties if value is not None)
|
||||
lines.extend(["END:VEVENT", "END:VCALENDAR", ""])
|
||||
return "\r\n".join(lines)
|
||||
|
||||
|
||||
def unfold_ical_lines(ics: str) -> list[str]:
|
||||
raw_lines = ics.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||||
lines: list[str] = []
|
||||
for line in raw_lines:
|
||||
if not line:
|
||||
continue
|
||||
if line[:1] in {" ", "\t"} and lines:
|
||||
lines[-1] += line[1:]
|
||||
else:
|
||||
lines.append(line)
|
||||
return lines
|
||||
|
||||
|
||||
def component_lines(lines: list[str], component: str) -> list[str]:
|
||||
begin = f"BEGIN:{component.upper()}"
|
||||
end = f"END:{component.upper()}"
|
||||
depth = 0
|
||||
collected: list[str] = []
|
||||
for line in lines:
|
||||
upper = line.upper()
|
||||
if upper == begin:
|
||||
depth += 1
|
||||
if depth == 1:
|
||||
collected = []
|
||||
continue
|
||||
if upper == end and depth:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return collected
|
||||
continue
|
||||
if depth:
|
||||
collected.append(line)
|
||||
return []
|
||||
|
||||
|
||||
def parse_content_line(line: str) -> ICalendarProperty:
|
||||
head, sep, value = line.partition(":")
|
||||
if not sep:
|
||||
raise ICalendarError(f"Invalid iCalendar content line: {line}")
|
||||
parts = head.split(";")
|
||||
name = parts[0].upper()
|
||||
params: dict[str, list[str]] = {}
|
||||
for raw_param in parts[1:]:
|
||||
key, param_sep, raw_value = raw_param.partition("=")
|
||||
if not param_sep:
|
||||
params[key.upper()] = [""]
|
||||
else:
|
||||
params[key.upper()] = [unescape_text(item.strip('"')) for item in raw_value.split(",")]
|
||||
return ICalendarProperty(name=name, params=params, value=unescape_text(value))
|
||||
|
||||
|
||||
def group_properties(properties: list[ICalendarProperty]) -> dict[str, list[ICalendarProperty]]:
|
||||
grouped: dict[str, list[ICalendarProperty]] = {}
|
||||
for prop in properties:
|
||||
grouped.setdefault(prop.name, []).append(prop)
|
||||
return grouped
|
||||
|
||||
|
||||
def first_property(grouped: dict[str, list[ICalendarProperty]], name: str) -> ICalendarProperty | None:
|
||||
items = grouped.get(name.upper()) or []
|
||||
return items[0] if items else None
|
||||
|
||||
|
||||
def first_value(grouped: dict[str, list[ICalendarProperty]], name: str) -> str | None:
|
||||
prop = first_property(grouped, name)
|
||||
return prop.value if prop else None
|
||||
|
||||
|
||||
def parse_temporal_property(prop: ICalendarProperty) -> tuple[datetime, bool, str | None]:
|
||||
value_type = (prop.params.get("VALUE") or [""])[0].upper()
|
||||
tzid = (prop.params.get("TZID") or [None])[0]
|
||||
if value_type == "DATE" or (len(prop.value) == 8 and "T" not in prop.value):
|
||||
parsed_date = datetime.strptime(prop.value, "%Y%m%d").date()
|
||||
return datetime.combine(parsed_date, time.min, tzinfo=timezone.utc), True, tzid
|
||||
return parse_ical_datetime(prop.value), False, tzid
|
||||
|
||||
|
||||
def parse_ical_datetime(value: str) -> datetime:
|
||||
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")
|
||||
return parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def format_ical_temporal(value: datetime, *, all_day: bool) -> str:
|
||||
if all_day:
|
||||
return value.date().strftime("%Y%m%d")
|
||||
return format_ical_datetime(value)
|
||||
|
||||
|
||||
def format_ical_datetime(value: datetime) -> str:
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=timezone.utc)
|
||||
return value.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
||||
|
||||
|
||||
def date_params(event: Any) -> dict[str, list[str]]:
|
||||
if event.all_day:
|
||||
return {"VALUE": ["DATE"]}
|
||||
if event.timezone:
|
||||
return {"TZID": [event.timezone]}
|
||||
return {}
|
||||
|
||||
|
||||
def property_party(prop: ICalendarProperty | None) -> dict[str, Any] | None:
|
||||
if prop is None:
|
||||
return None
|
||||
return {"value": prop.value, "params": prop.params}
|
||||
|
||||
|
||||
def party_to_property(name: str, party: dict[str, Any]) -> tuple[str, dict[str, list[str]], str]:
|
||||
params = party.get("params") if isinstance(party.get("params"), dict) else {}
|
||||
value = str(party.get("value") or party.get("email") or "")
|
||||
normalized_params = {str(key).upper(): [str(item) for item in value] for key, value in params.items() if isinstance(value, list)}
|
||||
return name, normalized_params, value
|
||||
|
||||
|
||||
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:
|
||||
result[key.upper()] = raw.split(",") if "," in raw else raw
|
||||
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 split_csv(value: str) -> list[str]:
|
||||
return [item.strip() for item in value.split(",") if item.strip()]
|
||||
|
||||
|
||||
def int_or_zero(value: str | None) -> int:
|
||||
try:
|
||||
return int(value or "0")
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
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 serialize_content_line(name: str, params: dict[str, list[str]], value: str) -> str:
|
||||
param_text = "".join(f";{key}={','.join(escape_param(item) for item in items)}" for key, items in params.items() if items)
|
||||
return f"{name}{param_text}:{escape_text(value)}"
|
||||
|
||||
|
||||
def fold_ical_line(line: str, limit: int = 75) -> str:
|
||||
if len(line) <= limit:
|
||||
return line
|
||||
parts = [line[:limit]]
|
||||
line = line[limit:]
|
||||
while line:
|
||||
parts.append(" " + line[: limit - 1])
|
||||
line = line[limit - 1 :]
|
||||
return "\r\n".join(parts)
|
||||
|
||||
|
||||
def escape_text(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace("\n", "\\n").replace(";", "\\;").replace(",", "\\,")
|
||||
|
||||
|
||||
def unescape_text(value: str) -> str:
|
||||
return value.replace("\\n", "\n").replace("\\N", "\n").replace("\\;", ";").replace("\\,", ",").replace("\\\\", "\\")
|
||||
|
||||
|
||||
def escape_param(value: str) -> str:
|
||||
if any(char in value for char in [":", ";", ","]):
|
||||
return '"' + value.replace('"', "'") + '"'
|
||||
return value
|
||||
|
||||
|
||||
def http_last_modified(value: datetime) -> str:
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=timezone.utc)
|
||||
return format_datetime(value.astimezone(timezone.utc), usegmt=True)
|
||||
Reference in New Issue
Block a user