Panama/config/dot/quickshell/scripts/calendar-agenda

#!/usr/bin/env python3

"""Read GNOME's Evolution Data Server calendars for the Panama shell."""

from __future__ import annotations

import json
import re
import sys
import time
from collections.abc import Iterable
from typing import Any
from urllib.parse import urlsplit

try:
    import gi

    gi.require_version("ECal", "2.0")
    gi.require_version("EDataServer", "1.2")
    gi.require_version("ICalGLib", "3.0")
    from gi.repository import ECal, EDataServer, Gio, GLib, ICalGLib

    EDS_READY = True
except (ImportError, ValueError):
    ECal = EDataServer = Gio = GLib = ICalGLib = None
    EDS_READY = False


MEETING_HOST_SUFFIXES = (
    "meet.google.com",
    "zoom.us",
    "teams.microsoft.com",
    "teams.live.com",
    "webex.com",
    "meet.jit.si",
)

URL_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE)
COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}(?:[0-9a-fA-F]{2})?$")
FALLBACK_COLOR = "#82aaff"


def stable_event_id(source_id: str, uid: str, recurrence_id: str) -> str:
    return f"{source_id}:{uid}:{recurrence_id}"


def sort_events(events: list[dict[str, object]]) -> list[dict[str, object]]:
    return sorted(
        events,
        key=lambda event: (
            not bool(event["allDay"]),
            int(event["start"]),
            str(event["id"]),
        ),
    )


def is_allowed_meeting_host(host: str) -> bool:
    normalized = host.lower().rstrip(".")
    return any(
        normalized == suffix or normalized.endswith("." + suffix)
        for suffix in MEETING_HOST_SUFFIXES
    )


def find_join_url(values: Iterable[str | None]) -> str:
    for value in values:
        if not value:
            continue
        for match in URL_PATTERN.findall(value):
            candidate = match.rstrip("),.;!?]}")
            try:
                parsed = urlsplit(candidate)
            except ValueError:
                continue
            if parsed.scheme.lower() != "https" or not parsed.hostname:
                continue
            if is_allowed_meeting_host(parsed.hostname):
                return candidate
    return ""


def _safe_component_value(component: Any, getter: str) -> str:
    try:
        return (getattr(component, getter)() or "").strip()
    except (AttributeError, TypeError):
        return ""


def normalize_instance(
    source_meta: dict[str, str],
    component: Any,
    instance_start: Any,
    instance_end: Any,
) -> dict[str, object]:
    uid = (component.get_uid() or "").strip()
    if not uid:
        raise ValueError("calendar component has no UID")

    start = int(instance_start.as_timet())
    end = int(instance_end.as_timet())
    if end < start:
        raise ValueError("calendar component ends before it starts")

    recurrence = component.get_recurrenceid()
    recurrence_id = str(start)
    if recurrence is not None and not recurrence.is_null_time():
        recurrence_id = recurrence.as_ical_string()

    summary = _safe_component_value(component, "get_summary") or "Untitled event"
    location = _safe_component_value(component, "get_location")
    description = _safe_component_value(component, "get_description")
    component_url = ""
    try:
        url_property = component.get_first_property(ICalGLib.PropertyKind.URL_PROPERTY)
        if url_property is not None:
            component_url = (url_property.get_url() or "").strip()
    except (AttributeError, TypeError):
        component_url = ""

    all_day = bool(instance_start.is_date() or instance_end.is_date())
    return {
        "id": stable_event_id(source_meta["id"], uid, recurrence_id),
        "sourceId": source_meta["id"],
        "uid": uid,
        "summary": summary,
        "start": start,
        "end": end,
        "allDay": all_day,
        "location": location,
        "joinUrl": find_join_url([component_url, location, description]),
    }


def _source_meta(source: Any) -> dict[str, str]:
    extension = source.get_extension(EDataServer.SOURCE_EXTENSION_CALENDAR)
    color = extension.get_color() or FALLBACK_COLOR
    if not COLOR_PATTERN.fullmatch(color):
        color = FALLBACK_COLOR
    return {
        "id": source.get_uid(),
        "name": (source.get_display_name() or "Calendar").strip(),
        "color": color.lower(),
    }


def _unavailable_snapshot() -> dict[str, object]:
    return {
        "ok": False,
        "generatedAt": int(time.time()),
        "sources": [],
        "events": [],
        "errors": [{"code": "eds-unavailable"}],
    }


def probe() -> dict[str, object]:
    if not EDS_READY:
        return {"eds": False, "sourceRegistry": False, "enabledSources": 0}
    try:
        registry = EDataServer.SourceRegistry.new_sync(None)
        sources = registry.list_enabled(EDataServer.SOURCE_EXTENSION_CALENDAR)
    except Exception:
        return {"eds": True, "sourceRegistry": False, "enabledSources": 0}
    return {
        "eds": True,
        "sourceRegistry": True,
        "enabledSources": len(sources),
    }


def collect_snapshot(start: int, end: int) -> dict[str, object]:
    if not EDS_READY:
        return _unavailable_snapshot()
    if start < 0 or end <= start:
        return {
            **_unavailable_snapshot(),
            "errors": [{"code": "invalid-range"}],
        }

    try:
        registry = EDataServer.SourceRegistry.new_sync(None)
        enabled_sources = registry.list_enabled(EDataServer.SOURCE_EXTENSION_CALENDAR)
    except Exception:
        return _unavailable_snapshot()

    sources: list[dict[str, str]] = []
    events_by_id: dict[str, dict[str, object]] = {}
    errors: list[dict[str, str]] = []

    for source in enabled_sources:
        source_meta = _source_meta(source)
        sources.append(source_meta)
        malformed_event = False

        def on_instance(
            component: Any,
            instance_start: Any,
            instance_end: Any,
            _cancellable: Any,
            _callback_data: Any,
        ) -> bool:
            nonlocal malformed_event
            try:
                event = normalize_instance(
                    source_meta,
                    component,
                    instance_start,
                    instance_end,
                )
                events_by_id[str(event["id"])] = event
            except (AttributeError, TypeError, ValueError):
                malformed_event = True
            return True

        try:
            client = ECal.Client.connect_sync(
                source,
                ECal.ClientSourceType.EVENTS,
                5,
                None,
            )
            if client is None:
                raise RuntimeError("calendar client unavailable")
            client.generate_instances_sync(start, end, None, on_instance, None)
        except Exception:
            errors.append({"sourceId": source_meta["id"], "code": "unavailable"})
            continue

        if malformed_event:
            errors.append({"sourceId": source_meta["id"], "code": "event-invalid"})

    sources.sort(key=lambda source: (source["name"].casefold(), source["id"]))
    return {
        "ok": True,
        "generatedAt": int(time.time()),
        "sources": sources,
        "events": sort_events(list(events_by_id.values())),
        "errors": errors,
    }


def _write_snapshot(snapshot: dict[str, object]) -> bool:
    try:
        sys.stdout.write(json.dumps(snapshot, separators=(",", ":")) + "\n")
        sys.stdout.flush()
    except BrokenPipeError:
        return False
    return True


def watch(start: int, end: int) -> int:
    if not EDS_READY:
        _write_snapshot(_unavailable_snapshot())
        return 1

    loop = GLib.MainLoop()
    try:
        registry = EDataServer.SourceRegistry.new_sync(None)
    except Exception:
        _write_snapshot(_unavailable_snapshot())
        return 1
    debounce_source = 0
    subscriptions: list[int] = []

    def emit_snapshot() -> bool:
        nonlocal debounce_source
        debounce_source = 0
        if not _write_snapshot(collect_snapshot(start, end)):
            loop.quit()
        return GLib.SOURCE_REMOVE

    def schedule_refresh(*_args: Any) -> None:
        nonlocal debounce_source
        if debounce_source:
            GLib.source_remove(debounce_source)
        debounce_source = GLib.timeout_add(350, emit_snapshot)

    def periodic_refresh() -> bool:
        emit_snapshot()
        return GLib.SOURCE_CONTINUE

    for signal_name in ("source-added", "source-removed", "source-changed"):
        try:
            registry.connect(signal_name, schedule_refresh)
        except TypeError:
            pass

    try:
        bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
        for signal_name in (
            "EventsAddedOrUpdated",
            "EventsRemoved",
            "ClientDisappeared",
        ):
            subscriptions.append(
                bus.signal_subscribe(
                    "org.gnome.Shell.CalendarServer",
                    "org.gnome.Shell.CalendarServer",
                    signal_name,
                    "/org/gnome/Shell/CalendarServer",
                    None,
                    Gio.DBusSignalFlags.NONE,
                    schedule_refresh,
                    None,
                )
            )
        proxy = Gio.DBusProxy.new_sync(
            bus,
            Gio.DBusProxyFlags.NONE,
            None,
            "org.gnome.Shell.CalendarServer",
            "/org/gnome/Shell/CalendarServer",
            "org.gnome.Shell.CalendarServer",
            None,
        )
        proxy.call_sync(
            "SetTimeRange",
            GLib.Variant("(xxb)", (start, end, True)),
            Gio.DBusCallFlags.NONE,
            -1,
            None,
        )
    except Exception:
        bus = None

    if not _write_snapshot(collect_snapshot(start, end)):
        return 0
    GLib.timeout_add_seconds(600, periodic_refresh)

    try:
        loop.run()
    except KeyboardInterrupt:
        pass
    finally:
        if bus is not None:
            for subscription in subscriptions:
                bus.signal_unsubscribe(subscription)
    return 0


def _usage_error() -> dict[str, object]:
    return {
        "ok": False,
        "generatedAt": int(time.time()),
        "sources": [],
        "events": [],
        "errors": [{"code": "usage"}],
    }


def main(argv: list[str]) -> int:
    command = argv[1] if len(argv) > 1 else ""
    if command == "probe" and len(argv) == 2:
        _write_snapshot(probe())
        return 0
    if command in {"query", "watch"} and len(argv) == 4:
        try:
            start = int(argv[2])
            end = int(argv[3])
        except ValueError:
            _write_snapshot(_usage_error())
            return 64
        if command == "query":
            snapshot = collect_snapshot(start, end)
            _write_snapshot(snapshot)
            return 0 if snapshot["ok"] else 1
        return watch(start, end)
    _write_snapshot(_usage_error())
    return 64


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))