Complete recurrence editing and calendar preferences

2026-07-31 05:46:50 +02:00
parent 1e063f51cc
commit f00cff1d8c
23 changed files with 2354 additions and 368 deletions
@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+import uuid
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+
+def graph_event_payload(item: dict[str, Any]) -> dict[str, Any]:
+    """Translate one Microsoft Graph event into the calendar event contract."""
+
+    start = graph_datetime(item.get("start"))
+    end = graph_datetime(item.get("end"))
+    all_day = bool(item.get("isAllDay"))
+    uid = str(
+        item.get("iCalUId")
+        or item.get("uid")
+        or item.get("id")
+        or uuid.uuid4()
+    )
+    return {
+        "uid": uid,
+        "recurrence_id": (
+            str(item.get("id"))
+            if item.get("type") == "occurrence"
+            else None
+        ),
+        "sequence": int(item.get("sequence") or 0),
+        "summary": str(item.get("subject") or "(Untitled event)"),
+        "description": graph_body_text(item),
+        "location": graph_location_text(item.get("location")),
+        "status": (
+            "CANCELLED" if item.get("isCancelled") else "CONFIRMED"
+        ),
+        "transparency": (
+            "TRANSPARENT"
+            if item.get("showAs") in {"free", "workingElsewhere"}
+            else "OPAQUE"
+        ),
+        "classification": (
+            "PRIVATE"
+            if item.get("sensitivity") == "private"
+            else "PUBLIC"
+        ),
+        "start_at": start,
+        "end_at": end,
+        "duration_seconds": int((end - start).total_seconds()),
+        "all_day": all_day,
+        "timezone": (
+            (item.get("start") or {}).get("timeZone")
+            if isinstance(item.get("start"), dict)
+            else None
+        )
+        or "UTC",
+        "organizer": graph_party(item.get("organizer")),
+        "attendees": [
+            graph_party(attendee)
+            for attendee in item.get("attendees") or []
+        ],
+        "categories": [
+            str(category) for category in item.get("categories") or []
+        ],
+        "rrule": (
+            item.get("recurrence")
+            if isinstance(item.get("recurrence"), dict)
+            else None
+        ),
+        "rdate": [],
+        "exdate": [],
+        "reminders": graph_reminders(item),
+        "attachments": [],
+        "related_to": [],
+        "etag": item.get("@odata.etag"),
+        "icalendar": {
+            "component": "VEVENT",
+            "schema_version": 1,
+            "provider": "graph",
+            "graph": item,
+        },
+        "metadata": {"graph": item},
+    }
+
+
+def graph_datetime(value: Any) -> datetime:
+    if not isinstance(value, dict) or not value.get("dateTime"):
+        return datetime.now(timezone.utc)
+    raw = str(value["dateTime"]).replace("Z", "+00:00")
+    parsed = datetime.fromisoformat(raw)
+    tz_name = str(value.get("timeZone") or "UTC")
+    if parsed.tzinfo is None:
+        try:
+            parsed = parsed.replace(tzinfo=ZoneInfo(tz_name))
+        except ZoneInfoNotFoundError:
+            parsed = parsed.replace(tzinfo=timezone.utc)
+    return normalize_datetime(parsed)
+
+
+def graph_body_text(item: dict[str, Any]) -> str | None:
+    body = item.get("body")
+    if isinstance(body, dict) and body.get("content"):
+        return str(body["content"])
+    return str(item.get("bodyPreview")) if item.get("bodyPreview") else None
+
+
+def graph_location_text(value: Any) -> str | None:
+    if isinstance(value, dict) and value.get("displayName"):
+        return str(value["displayName"])
+    return None
+
+
+def graph_party(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        return {}
+    email = (
+        value.get("emailAddress")
+        if isinstance(value.get("emailAddress"), dict)
+        else value
+    )
+    result = {
+        "name": (
+            str(email.get("name") or "")
+            if isinstance(email, dict)
+            else ""
+        ),
+        "email": (
+            str(email.get("address") or "")
+            if isinstance(email, dict)
+            else ""
+        ),
+    }
+    if value.get("type"):
+        result["role"] = value["type"]
+    if value.get("status"):
+        result["status"] = value["status"]
+    return result
+
+
+def graph_reminders(item: dict[str, Any]) -> list[dict[str, Any]]:
+    if (
+        item.get("isReminderOn")
+        and item.get("reminderMinutesBeforeStart") is not None
+    ):
+        return [
+            {
+                "action": "DISPLAY",
+                "trigger_minutes_before": int(
+                    item["reminderMinutesBeforeStart"]
+                ),
+            }
+        ]
+    return []
+
+
+def normalize_datetime(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)