govoplan-calendar/src/govoplan_calendar/backend/ews.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from defusedxml import ElementTree as SafeElementTree


EWS_SOAP_NS = "http://schemas.xmlsoap.org/soap/envelope/"
EWS_MESSAGES_NS = (
    "http://schemas.microsoft.com/exchange/services/2006/messages"
)
EWS_TYPES_NS = "http://schemas.microsoft.com/exchange/services/2006/types"


class EwsAdapterError(ValueError):
    """Raised when an EWS response cannot be translated safely."""


def ews_find_item_body(
    *,
    start: datetime,
    end: datetime,
    mailbox: Any | None = None,
) -> str:
    start_text = normalize_datetime(start).isoformat().replace("+00:00", "Z")
    end_text = normalize_datetime(end).isoformat().replace("+00:00", "Z")
    mailbox_xml = ""
    if mailbox:
        mailbox_xml = (
            "<t:Mailbox><t:EmailAddress>"
            f"{xml_escape(str(mailbox))}"
            "</t:EmailAddress></t:Mailbox>"
        )
    return f"""<?xml version="1.0" encoding="utf-8"?>
<s:Envelope xmlns:s="{EWS_SOAP_NS}" xmlns:m="{EWS_MESSAGES_NS}" xmlns:t="{EWS_TYPES_NS}">
  <s:Header>
    <t:RequestServerVersion Version="Exchange2013_SP1" />
  </s:Header>
  <s:Body>
    <m:FindItem Traversal="Shallow">
      <m:ItemShape>
        <t:BaseShape>AllProperties</t:BaseShape>
      </m:ItemShape>
      <m:CalendarView StartDate="{start_text}" EndDate="{end_text}" />
      <m:ParentFolderIds>
        <t:DistinguishedFolderId Id="calendar">{mailbox_xml}</t:DistinguishedFolderId>
      </m:ParentFolderIds>
    </m:FindItem>
  </s:Body>
</s:Envelope>"""


def parse_ews_calendar_items(xml_text: str) -> list[dict[str, Any]]:
    try:
        root = SafeElementTree.fromstring(xml_text)
    except SafeElementTree.ParseError as exc:
        raise EwsAdapterError(f"Invalid EWS response XML: {exc}") from exc
    items: list[dict[str, Any]] = []
    for item in root.findall(f".//{{{EWS_TYPES_NS}}}CalendarItem"):
        item_id = item.find(f"./{{{EWS_TYPES_NS}}}ItemId")
        href = item_id.get("Id") if item_id is not None else None
        if not href:
            continue
        change_key = (
            item_id.get("ChangeKey") if item_id is not None else None
        )
        start = parse_ews_datetime(text_of(item, "Start"))
        end_text = text_of(item, "End")
        end = parse_ews_datetime(end_text) if end_text else start
        uid = text_of(item, "UID") or href
        subject = text_of(item, "Subject") or "(Untitled event)"
        all_day = text_of(item, "IsAllDayEvent") == "true"
        free_busy = text_of(item, "LegacyFreeBusyStatus") or "Busy"
        sensitivity = text_of(item, "Sensitivity") or "Normal"
        body = item.find(f"./{{{EWS_TYPES_NS}}}Body")
        provider_payload = element_to_dict(item)
        items.append(
            {
                "href": href,
                "uid": uid,
                "recurrence_id": (
                    href
                    if text_of(item, "CalendarItemType")
                    in {"Occurrence", "Exception"}
                    else None
                ),
                "sequence": int_or_default(
                    text_of(item, "AppointmentSequenceNumber"),
                    0,
                ),
                "summary": subject,
                "description": body.text if body is not None else None,
                "location": text_of(item, "Location"),
                "status": (
                    "CANCELLED"
                    if text_of(item, "IsCancelled") == "true"
                    else "CONFIRMED"
                ),
                "transparency": (
                    "TRANSPARENT"
                    if free_busy.lower() == "free"
                    else "OPAQUE"
                ),
                "classification": (
                    "PRIVATE"
                    if sensitivity.lower() == "private"
                    else "PUBLIC"
                ),
                "start_at": start,
                "end_at": end,
                "duration_seconds": int((end - start).total_seconds()),
                "all_day": all_day,
                "timezone": "UTC",
                "organizer": ews_mailbox_record(
                    item.find(
                        f"./{{{EWS_TYPES_NS}}}Organizer/"
                        f"{{{EWS_TYPES_NS}}}Mailbox"
                    )
                ),
                "attendees": ews_attendees(item),
                "categories": [
                    category.text
                    for category in item.findall(
                        f"./{{{EWS_TYPES_NS}}}Categories/"
                        f"{{{EWS_TYPES_NS}}}String"
                    )
                    if category.text
                ],
                "rrule": None,
                "rdate": [],
                "exdate": [],
                "reminders": ews_reminders(item),
                "attachments": [],
                "related_to": [],
                "etag": change_key,
                "icalendar": {
                    "component": "VEVENT",
                    "schema_version": 1,
                    "provider": "ews",
                    "ews": provider_payload,
                },
                "metadata": {"ews": provider_payload},
            }
        )
    return items


def parse_ews_datetime(value: str | None) -> datetime:
    if not value:
        return datetime.now(timezone.utc)
    return normalize_datetime(
        datetime.fromisoformat(value.replace("Z", "+00:00"))
    )


def text_of(item: Any, name: str) -> str | None:
    child = item.find(f"./{{{EWS_TYPES_NS}}}{name}")
    return child.text if child is not None else None


def ews_mailbox_record(mailbox: Any | None) -> dict[str, Any] | None:
    if mailbox is None:
        return None
    return {
        "name": text_of(mailbox, "Name") or "",
        "email": text_of(mailbox, "EmailAddress") or "",
        "routing_type": text_of(mailbox, "RoutingType") or "",
    }


def ews_attendees(item: Any) -> list[dict[str, Any]]:
    attendees: list[dict[str, Any]] = []
    for role, path in (
        ("required", "RequiredAttendees"),
        ("optional", "OptionalAttendees"),
    ):
        for attendee in item.findall(
            f"./{{{EWS_TYPES_NS}}}{path}/"
            f"{{{EWS_TYPES_NS}}}Attendee"
        ):
            mailbox = attendee.find(f"./{{{EWS_TYPES_NS}}}Mailbox")
            record = ews_mailbox_record(mailbox) or {}
            record["role"] = role
            response = text_of(attendee, "ResponseType")
            if response:
                record["status"] = response
            attendees.append(record)
    return attendees


def ews_reminders(item: Any) -> list[dict[str, Any]]:
    if text_of(item, "ReminderIsSet") == "true":
        minutes = int_or_default(
            text_of(item, "ReminderMinutesBeforeStart"),
            15,
        )
        return [{"action": "DISPLAY", "trigger_minutes_before": minutes}]
    return []


def element_to_dict(element: Any) -> dict[str, Any]:
    tag = element.tag.rsplit("}", 1)[-1]
    children = list(element)
    result: dict[str, Any] = {"tag": tag}
    if element.attrib:
        result["attributes"] = dict(element.attrib)
    if element.text and element.text.strip():
        result["text"] = element.text.strip()
    if children:
        result["children"] = [
            element_to_dict(child) for child in children
        ]
    return result


def int_or_default(value: str | None, default: int) -> int:
    try:
        return int(value) if value is not None else default
    except ValueError:
        return default


def xml_escape(value: str) -> str:
    return (
        value.replace("&", "&amp;")
        .replace("<", "&lt;")
        .replace(">", "&gt;")
        .replace('"', "&quot;")
    )


def normalize_datetime(value: datetime) -> datetime:
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)