158 lines
4.6 KiB
Python
158 lines
4.6 KiB
Python
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)
|