371 lines
11 KiB
Python
Executable File
371 lines
11 KiB
Python
Executable File
#!/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()
|
|
registry = EDataServer.SourceRegistry.new_sync(None)
|
|
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))
|