diff --git a/src/smorg/integrations/__init__.py b/src/smorg/integrations/__init__.py index 27017b7..405b678 100644 --- a/src/smorg/integrations/__init__.py +++ b/src/smorg/integrations/__init__.py @@ -8,9 +8,10 @@ from __future__ import annotations from smorg.core.contract import Integration -from smorg.integrations import github, linear, spotify +from smorg.integrations import gcal, github, linear, spotify INTEGRATIONS: tuple[Integration, ...] = ( + gcal.INTEGRATION, github.INTEGRATION, linear.INTEGRATION, spotify.INTEGRATION, diff --git a/src/smorg/integrations/gcal/__init__.py b/src/smorg/integrations/gcal/__init__.py new file mode 100644 index 0000000..5c6c706 --- /dev/null +++ b/src/smorg/integrations/gcal/__init__.py @@ -0,0 +1,3 @@ +from smorg.integrations.gcal.manifest import INTEGRATION + +__all__ = ["INTEGRATION"] diff --git a/src/smorg/integrations/gcal/manifest.py b/src/smorg/integrations/gcal/manifest.py new file mode 100644 index 0000000..4ff179c --- /dev/null +++ b/src/smorg/integrations/gcal/manifest.py @@ -0,0 +1,54 @@ +"""Google Calendar's declaration; connects with the OAuth app smorg registered itself, and reads +the REST API. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import timedelta + +import httpx + +from smorg.auth.oauth import BundledProvider, OAuthMethod, ServerMetadata +from smorg.auth.store import Credentials +from smorg.core.contract import AuthPath, Item, Manifest +from smorg.integrations.gcal.panel import CalendarPanel +from smorg.integrations.gcal.source import fetch + +# The app lives in the Google Cloud project "smorg", owned by the maintainer. Google treats an +# installed app's secret as public; it still never reaches output. +CLIENT_ID = "REPLACE-ME.apps.googleusercontent.com" +CLIENT_SECRET = "REPLACE-ME" + +METHOD = OAuthMethod( + provider=BundledProvider( + metadata=ServerMetadata( + authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth", + token_endpoint="https://oauth2.googleapis.com/token", + revocation_endpoint="https://oauth2.googleapis.com/revoke", + ), + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + ), + scopes=("https://www.googleapis.com/auth/calendar.readonly",), +) + +MANIFEST = Manifest( + id="gcal", + display_name="Google Calendar", + connections=(AuthPath(id="oauth", method=METHOD),), + stale_after=timedelta(minutes=5), + actions=(), +) + + +@dataclass(frozen=True) +class CalendarIntegration: + manifest: Manifest = MANIFEST + panel_class: type[CalendarPanel] = CalendarPanel + + def fetch(self, credentials: Credentials, http: httpx.Client) -> tuple[Item, ...]: + return fetch(credentials, http) + + +INTEGRATION = CalendarIntegration() diff --git a/src/smorg/integrations/gcal/palette.py b/src/smorg/integrations/gcal/palette.py new file mode 100644 index 0000000..1bec466 --- /dev/null +++ b/src/smorg/integrations/gcal/palette.py @@ -0,0 +1,10 @@ +"""Google's colours, spent sparingly: blue for today, red for whatever wants attention.""" + +from __future__ import annotations + +TODAY_BLUE = "#1a73e8" +BRAND_BLUE = "#4285f4" +RED = "#ea4335" +YELLOW = "#fbbc04" +GREEN = "#34a853" +BRAND_DOTS = (BRAND_BLUE, RED, YELLOW, GREEN) diff --git a/src/smorg/integrations/gcal/panel.py b/src/smorg/integrations/gcal/panel.py new file mode 100644 index 0000000..eb1880e --- /dev/null +++ b/src/smorg/integrations/gcal/panel.py @@ -0,0 +1,155 @@ +"""Google Calendar's tab: a host panel that swaps between the landing and the calendar views.""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import date, datetime, timedelta, tzinfo +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from textual.app import ComposeResult + +from smorg.core.contract import Item +from smorg.integrations.gcal.source import ( + Calendar, + Calendars, + Event, + EventKind, + Response, + window_for, +) +from smorg.integrations.gcal.views import CalendarView +from smorg.integrations.gcal.views.menu import CalendarMenu +from smorg.shell.animation import FrameClock +from smorg.shell.view_host import HostedView, ViewHostPanel + +# One repaint every ten seconds moves the now-line and the countdown without a fetch. +_CLOCK_FPS = 0.1 + + +class CalendarPanel(ViewHostPanel[CalendarView]): + DEFAULT_CSS = """ + CalendarPanel { align-horizontal: center; } + """ + + def __init__(self) -> None: + super().__init__(CalendarView.MENU) + self.now_provider: Callable[[], datetime] | None = None + self.focused_day: date | None = None + self.viewed: Event | None = None + self.return_view: CalendarView = CalendarView.DAY + self.clock = FrameClock(self, _CLOCK_FPS, self._tick) + + def compose(self) -> ComposeResult: + yield CalendarMenu(self) + + def view_classes(self) -> dict[CalendarView, type[HostedView]]: + return {CalendarView.MENU: CalendarMenu} + + def on_mount(self) -> None: + super().on_mount() + self.clock.start() + + def _tick(self, elapsed: float) -> None: + self.refresh() + + def zone(self) -> tzinfo: + calendars = self.calendars() + if calendars is None: + local_zone = datetime.now().astimezone().tzinfo + if local_zone is not None: + return local_zone + return ZoneInfo("UTC") + try: + return ZoneInfo(calendars.timezone) + except ZoneInfoNotFoundError: + return ZoneInfo("UTC") + + def now(self) -> datetime: + if self.now_provider is not None: + return self.now_provider() + return datetime.now(self.zone()) + + def today(self) -> date: + return self.now().date() + + def focused(self) -> date: + if self.focused_day is None: + return self.today() + return self.focused_day + + def window(self) -> tuple[date, date]: + start, end = window_for(self.now()) + last = end.date() - timedelta(days=1) + return start.date(), last + + def set_focused(self, day: date) -> None: + first, last = self.window() + if day < first: + day = first + if day > last: + day = last + self.focused_day = day + self.refresh() + + def calendars(self) -> Calendars | None: + for item in self.items: + if isinstance(item, Calendars): + return item + return None + + def calendar_for(self, calendar_id: str) -> Calendar | None: + calendars = self.calendars() + if calendars is None: + return None + for calendar in calendars.calendars: + if calendar.id == calendar_id: + return calendar + return None + + def events(self) -> tuple[Event, ...]: + events = [item for item in self.items if isinstance(item, Event)] + return tuple(events) + + def events_on(self, day: date) -> tuple[Event, ...]: + day_start = datetime.combine(day, datetime.min.time(), tzinfo=self.zone()) + day_end = day_start + timedelta(days=1) + on_day = [ + event for event in self.events() if event.start < day_end and event.end > day_start + ] + return tuple(on_day) + + def next_meeting(self) -> Event | None: + now = self.now() + for event in self.events(): + if event.all_day or event.kind is not EventKind.MEETING: + continue + if event.start >= now: + return event + return None + + def invites(self) -> tuple[Event, ...]: + now = self.now() + pending = [ + event + for event in self.events() + if event.my_response is Response.NEEDS_ACTION and event.end >= now + ] + return tuple(pending) + + def open_event(self, event: Event, return_view: CalendarView) -> None: + self.viewed = event + self.return_view = return_view + self.mark_seen(event) + self.show_view(CalendarView.EVENT) + + def close_event(self) -> None: + self.viewed = None + self.show_view(self.return_view) + + def seen_items(self) -> tuple[Item, ...]: + return self.events() + + def selected_item(self) -> Item | None: + if self.active_view is CalendarView.EVENT: + return self.viewed + return None diff --git a/src/smorg/integrations/gcal/source.py b/src/smorg/integrations/gcal/source.py new file mode 100644 index 0000000..f86dba0 --- /dev/null +++ b/src/smorg/integrations/gcal/source.py @@ -0,0 +1,387 @@ +"""Fetch the user's selected calendars and two weeks of their events from the Calendar REST API, +mapped to typed items in the user's own timezone. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import StrEnum +from typing import Any +from urllib.parse import quote, urlsplit +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +import httpx + +from smorg.auth.store import Credentials, now +from smorg.core.contract import ( + AccessNotAllowed, + AuthExpired, + Item, + Malformed, + Unavailable, +) +from smorg.core.shape import optional_string, required_string, timestamp +from smorg.core.text import flatten_html, sanitize_block, sanitize_line + +API = "https://www.googleapis.com/calendar/v3" +TIMEZONE_ENDPOINT = f"{API}/users/me/settings/timezone" +CALENDAR_LIST_ENDPOINT = f"{API}/users/me/calendarList" +CALENDARS_ID = "calendars" +CALENDAR_HOME = "https://calendar.google.com" +WINDOW_WEEKS = 2 +PAGE_SIZE = 250 + + +@dataclass(frozen=True) +class Calendar: + id: str + name: str + color: str + primary: bool + + +@dataclass(frozen=True) +class Calendars(Item): + """The user's selected calendars and the timezone every event was converted into.""" + + timezone: str + calendars: tuple[Calendar, ...] + + +class EventKind(StrEnum): + MEETING = "meeting" + WORKING_LOCATION = "working_location" + OUT_OF_OFFICE = "out_of_office" + FOCUS_TIME = "focus_time" + + +class Response(StrEnum): + ACCEPTED = "accepted" + TENTATIVE = "tentative" + DECLINED = "declined" + NEEDS_ACTION = "needs_action" + NONE = "none" + + +_KINDS = { + "workingLocation": EventKind.WORKING_LOCATION, + "outOfOffice": EventKind.OUT_OF_OFFICE, + "focusTime": EventKind.FOCUS_TIME, +} +_RESPONSES = { + "accepted": Response.ACCEPTED, + "tentative": Response.TENTATIVE, + "declined": Response.DECLINED, + "needsAction": Response.NEEDS_ACTION, +} +UNTITLED = "(no title)" + + +@dataclass(frozen=True) +class Attendee: + name: str + email: str + response: Response + organizer: bool + is_self: bool + + +@dataclass(frozen=True) +class Event(Item): + calendar_id: str + title: str + start: datetime + end: datetime + all_day: bool + kind: EventKind + my_response: Response + attendees: tuple[Attendee, ...] + organizer: str + meet_url: str + location: str + description: str + attachments: tuple[str, ...] + recurring: bool + + +def encoded_calendar_id(calendar_id: str) -> str: + """A calendar id as a path segment: holiday calendars carry a '#' that a raw path would lose + as a fragment.""" + return quote(calendar_id, safe="") + + +def window_for(moment: datetime) -> tuple[datetime, datetime]: + """Monday 00:00 of moment's week through Monday 00:00 two weeks later, in moment's zone.""" + monday = moment.date() - timedelta(days=moment.weekday()) + start = datetime.combine(monday, datetime.min.time(), tzinfo=moment.tzinfo) + end = start + timedelta(weeks=WINDOW_WEEKS) + return start, end + + +def fetch(credentials: Credentials, http: httpx.Client) -> tuple[Item, ...]: + zone_name = _fetch_timezone(credentials, http) + try: + zone = ZoneInfo(zone_name) + except ZoneInfoNotFoundError as error: + raise Malformed(f"Google named an unknown timezone {zone_name!r}") from error + calendars = _fetch_calendars(credentials, http) + calendars_item = Calendars( + id=CALENDARS_ID, + updated_at=now(), + url=CALENDAR_HOME, + timezone=zone_name, + calendars=calendars, + ) + current = now().astimezone(zone) + window_start, window_end = window_for(current) + events: list[Event] = [] + for calendar in calendars: + events.extend(_fetch_events(credentials, http, calendar, zone, window_start, window_end)) + events.sort(key=lambda event: event.start) + return (calendars_item, *events) + + +def _get( + credentials: Credentials, http: httpx.Client, url: str, params: dict[str, Any] | None = None +) -> httpx.Response: + if params is None: + request_params = {} + else: + request_params = params + try: + response = http.get( + url, + params=request_params, + headers={"Authorization": f"Bearer {credentials.access_token}"}, + ) + except httpx.HTTPError as error: + raise Unavailable("could not reach Google Calendar") from error + if response.status_code == 401: + raise AuthExpired("Google rejected the stored token; it may have expired or been revoked") + if response.status_code == 403: + raise AccessNotAllowed("Google refused access to the calendar") + return response + + +def _require_ok(response: httpx.Response) -> None: + if response.status_code != 200: + raise Unavailable(f"Google Calendar returned HTTP {response.status_code}") + + +def _json_object(response: httpx.Response) -> dict[str, Any]: + try: + payload = response.json() + except ValueError as error: + raise Malformed("Google Calendar returned a body that is not JSON") from error + if not isinstance(payload, dict): + raise Malformed(f"Google Calendar returned {type(payload).__name__}, expected an object") + return payload + + +def _items_of(payload: dict[str, Any]) -> list[dict[str, Any]]: + raw_items = payload.get("items", []) + if not isinstance(raw_items, list): + raise Malformed("'items' was not a list") + items: list[dict[str, Any]] = [] + for raw in raw_items: + if not isinstance(raw, dict): + raise Malformed(f"an item was {type(raw).__name__}, expected an object") + items.append(raw) + return items + + +def _fetch_timezone(credentials: Credentials, http: httpx.Client) -> str: + response = _get(credentials, http, TIMEZONE_ENDPOINT) + _require_ok(response) + payload = _json_object(response) + return required_string(payload, "value") + + +def _fetch_calendars(credentials: Credentials, http: httpx.Client) -> tuple[Calendar, ...]: + response = _get(credentials, http, CALENDAR_LIST_ENDPOINT) + _require_ok(response) + payload = _json_object(response) + calendars: list[Calendar] = [] + for raw in _items_of(payload): + if raw.get("selected") is not True: + continue + summary = required_string(raw, "summary") + calendars.append( + Calendar( + id=required_string(raw, "id"), + name=sanitize_line(summary), + color=optional_string(raw, "backgroundColor"), + primary=raw.get("primary") is True, + ) + ) + return tuple(calendars) + + +def _fetch_events( + credentials: Credentials, + http: httpx.Client, + calendar: Calendar, + zone: ZoneInfo, + window_start: datetime, + window_end: datetime, +) -> list[Event]: + url = f"{API}/calendars/{encoded_calendar_id(calendar.id)}/events" + params: dict[str, Any] = { + "timeMin": window_start.isoformat(), + "timeMax": window_end.isoformat(), + "singleEvents": "true", + "orderBy": "startTime", + "maxResults": PAGE_SIZE, + } + events: list[Event] = [] + while True: + response = _get(credentials, http, url, params) + if response.status_code == 404: + return [] + _require_ok(response) + payload = _json_object(response) + for raw in _items_of(payload): + event = _event_of(raw, calendar.id, zone) + if event is not None: + events.append(event) + next_page = payload.get("nextPageToken") + if not isinstance(next_page, str) or not next_page: + return events + params = params | {"pageToken": next_page} + + +def _event_of(raw: dict[str, Any], calendar_id: str, zone: ZoneInfo) -> Event | None: + if raw.get("status") == "cancelled": + return None + attendees = _attendees_of(raw) + my_response = _my_response_of(attendees) + if my_response is Response.DECLINED: + return None + start, all_day = _moment_of(raw, "start", zone) + end, _ = _moment_of(raw, "end", zone) + html_link = required_string(raw, "htmlLink") + event_type = optional_string(raw, "eventType") + location = optional_string(raw, "location") + return Event( + id=required_string(raw, "id"), + updated_at=timestamp(raw, "updated"), + url=_https_url(html_link), + calendar_id=calendar_id, + title=_title_of(raw), + start=start, + end=end, + all_day=all_day, + kind=_KINDS.get(event_type, EventKind.MEETING), + my_response=my_response, + attendees=attendees, + organizer=_organizer_of(raw), + meet_url=_meet_url_of(raw), + location=sanitize_line(location, limit=200), + description=_description_of(raw), + attachments=_attachments_of(raw), + recurring="recurringEventId" in raw, + ) + + +def _moment_of(raw: dict[str, Any], key: str, zone: ZoneInfo) -> tuple[datetime, bool]: + """An event boundary in the user's zone; all-day boundaries are midnight of Google's date.""" + boundary = raw.get(key) + if not isinstance(boundary, dict): + raise Malformed(f"'{key}' was {type(boundary).__name__}, expected an object") + if "dateTime" in boundary: + stamped = timestamp(boundary, "dateTime") + return stamped.astimezone(zone), False + day_text = required_string(boundary, "date") + try: + day = datetime.strptime(day_text, "%Y-%m-%d").date() + except ValueError as error: + raise Malformed(f"'{key}.date' was not a date: {day_text!r}") from error + midnight = datetime.combine(day, datetime.min.time(), tzinfo=zone) + return midnight, True + + +def _title_of(raw: dict[str, Any]) -> str: + summary = optional_string(raw, "summary") + if not summary.strip(): + return UNTITLED + return sanitize_line(summary) + + +def _https_url(url: str) -> str: + scheme = urlsplit(url).scheme + if scheme != "https": + raise Malformed("an event link was not https") + return url + + +def _meet_url_of(raw: dict[str, Any]) -> str: + link = optional_string(raw, "hangoutLink") + if not link: + return "" + return _https_url(link) + + +def _attendees_of(raw: dict[str, Any]) -> tuple[Attendee, ...]: + raw_attendees = raw.get("attendees", []) + if not isinstance(raw_attendees, list): + raise Malformed("'attendees' was not a list") + attendees: list[Attendee] = [] + for entry in raw_attendees: + if not isinstance(entry, dict): + raise Malformed(f"an attendee was {type(entry).__name__}, expected an object") + raw_email = optional_string(entry, "email") + email = sanitize_line(raw_email) + name = optional_string(entry, "displayName") + if not name: + name = email + response_status = optional_string(entry, "responseStatus") + attendees.append( + Attendee( + name=sanitize_line(name), + email=email, + response=_RESPONSES.get(response_status, Response.NONE), + organizer=entry.get("organizer") is True, + is_self=entry.get("self") is True, + ) + ) + return tuple(attendees) + + +def _my_response_of(attendees: tuple[Attendee, ...]) -> Response: + for attendee in attendees: + if attendee.is_self: + return attendee.response + return Response.NONE + + +def _organizer_of(raw: dict[str, Any]) -> str: + organizer = raw.get("organizer") + if not isinstance(organizer, dict): + return "" + name = optional_string(organizer, "displayName") + if not name: + name = optional_string(organizer, "email") + return sanitize_line(name) + + +def _description_of(raw: dict[str, Any]) -> str: + text = optional_string(raw, "description") + if not text: + return "" + flattened = flatten_html(text) + sanitized = sanitize_block(flattened) + return sanitized.strip() + + +def _attachments_of(raw: dict[str, Any]) -> tuple[str, ...]: + raw_attachments = raw.get("attachments", []) + if not isinstance(raw_attachments, list): + raise Malformed("'attachments' was not a list") + titles: list[str] = [] + for entry in raw_attachments: + if not isinstance(entry, dict): + raise Malformed(f"an attachment was {type(entry).__name__}, expected an object") + raw_title = optional_string(entry, "title") + titles.append(sanitize_line(raw_title)) + return tuple(titles) diff --git a/src/smorg/integrations/gcal/views/__init__.py b/src/smorg/integrations/gcal/views/__init__.py new file mode 100644 index 0000000..c277d44 --- /dev/null +++ b/src/smorg/integrations/gcal/views/__init__.py @@ -0,0 +1,14 @@ +"""Google Calendar's views: full-tab looks the host panel swaps between.""" + +from __future__ import annotations + +from enum import StrEnum + + +class CalendarView(StrEnum): + MENU = "menu" + DAY = "day" + WEEK = "week" + INVITES = "invites" + UPCOMING = "upcoming" + EVENT = "event" diff --git a/src/smorg/integrations/gcal/views/menu.py b/src/smorg/integrations/gcal/views/menu.py new file mode 100644 index 0000000..6f9cc6d --- /dev/null +++ b/src/smorg/integrations/gcal/views/menu.py @@ -0,0 +1,277 @@ +"""The landing: today's date in the icon, the next meeting's countdown, Google's dots breathing, +and the way into each view.""" + +from __future__ import annotations + +import math +import webbrowser +from dataclasses import dataclass +from datetime import datetime +from typing import TYPE_CHECKING + +from rich.console import Group, RenderableType +from rich.text import Text +from textual.app import RenderResult +from textual.binding import Binding +from textual.widgets import Static + +from smorg.integrations.gcal.palette import BRAND_DOTS, RED, TODAY_BLUE +from smorg.integrations.gcal.source import CALENDAR_HOME, Event, EventKind, Response +from smorg.integrations.gcal.views import CalendarView +from smorg.shell.animation import FrameClock +from smorg.shell.cards import CHANGED_MARK, SELECTED_MARK, format_count +from smorg.shell.cursor import step_cursor +from smorg.shell.format import plain_lines +from smorg.shell.panel import PanelState +from smorg.shell.view_host import HostedView + +if TYPE_CHECKING: + from smorg.integrations.gcal.panel import CalendarPanel + +_ENTER_GLYPH = "⏎" +_COMPACT_MIN_COLUMNS = 100 +_ICON_INNER_WIDTH = 16 +_PIXEL = "██" +_BLANK_PIXEL = " " +_DIGIT_GAP = " " +_DIGIT_FONT: dict[str, tuple[str, str, str, str, str]] = { + "0": ("###", "#.#", "#.#", "#.#", "###"), + "1": (".#.", "##.", ".#.", ".#.", "###"), + "2": ("###", "..#", "###", "#..", "###"), + "3": ("###", "..#", "###", "..#", "###"), + "4": ("#.#", "#.#", "###", "..#", "..#"), + "5": ("###", "#..", "###", "..#", "###"), + "6": ("###", "#..", "###", "#.#", "###"), + "7": ("###", "..#", "..#", "..#", "..#"), + "8": ("###", "#.#", "###", "#.#", "###"), + "9": ("###", "#.#", "###", "..#", "###"), +} + +_DOTS_FPS = 10 +_BREATH_SECONDS = 1.6 +_DOT_GLYPH = "●" + + +@dataclass(frozen=True) +class _Destination: + label: str + view: CalendarView + + +_DESTINATIONS = ( + _Destination("today", CalendarView.DAY), + _Destination("week", CalendarView.WEEK), + _Destination("invites awaiting your reply", CalendarView.INVITES), + _Destination("upcoming", CalendarView.UPCOMING), +) + + +def next_meeting_label(event: Event, now: datetime) -> str: + """ "in 48 min" today, "tomorrow · 09:00" the day after, "Thu · 09:00" beyond that.""" + clock = event.start.strftime("%H:%M") + days_ahead = (event.start.date() - now.date()).days + if days_ahead == 0: + minutes = max(0, math.ceil((event.start - now).total_seconds() / 60)) + return f"in {minutes} min" + if days_ahead == 1: + return f"tomorrow · {clock}" + weekday = event.start.strftime("%a") + return f"{weekday} · {clock}" + + +def _digit_rows(digit: str) -> list[str]: + rows: list[str] = [] + for font_row in _DIGIT_FONT[digit]: + pixels: list[str] = [] + for character in font_row: + if character == "#": + pixels.append(_PIXEL) + else: + pixels.append(_BLANK_PIXEL) + rows.append("".join(pixels)) + return rows + + +def _format_icon(day: int) -> list[Text]: + number = f"{day:02d}" + left_rows = _digit_rows(number[0]) + right_rows = _digit_rows(number[1]) + top = Text("╭", style="dim") + top.append("─" * _ICON_INNER_WIDTH, style="dim") + top.append("◥", style=f"bold {RED}") + lines = [top] + for left, right in zip(left_rows, right_rows, strict=True): + row = Text("│", style="dim") + row.append(f" {left}{_DIGIT_GAP}{right} ", style=f"bold {TODAY_BLUE}") + row.append("│", style="dim") + lines.append(row) + bottom = Text("╰", style="dim") + bottom.append("─" * _ICON_INNER_WIDTH, style="dim") + bottom.append("╯", style="dim") + lines.append(bottom) + return lines + + +def _format_header(now: datetime) -> Text: + line = Text(now.strftime("%A, %B %-d"), style="bold") + line.append(" · ", style="dim") + line.append(now.strftime("%H:%M"), style="dim") + return line + + +def _format_countdown(event: Event | None, now: datetime) -> list[Text]: + if event is None: + return [Text("nothing scheduled", style="dim")] + title = Text(event.title, style="bold") + line = Text() + line.append(f"{CHANGED_MARK} ", style=f"bold {RED}") + line.append(next_meeting_label(event, now), style=f"bold {RED}") + span = f"{event.start.strftime('%H:%M')}–{event.end.strftime('%H:%M')}" + line.append(f" · {span}", style="dim") + if event.my_response is Response.NEEDS_ACTION: + line.append(" · awaiting your reply", style=RED) + return [title, line] + + +def _dot_style(index: int, elapsed: float) -> str: + phase = (elapsed / _BREATH_SECONDS + index / 4) % 1.0 + brightness = 0.5 + 0.5 * math.cos(2 * math.pi * phase) + if brightness > 0.66: + return "bold" + if brightness > 0.33: + return "" + return "dim" + + +def _format_dots(elapsed: float) -> Text: + row = Text() + for index, color in enumerate(BRAND_DOTS): + if index > 0: + row.append(" ") + brightness = _dot_style(index, elapsed) + style = f"{brightness} {color}".strip() + row.append(_DOT_GLYPH, style=style) + return row + + +def _description_for(view: CalendarView, panel: CalendarPanel) -> str: + today_events = [e for e in panel.events_on(panel.today()) if e.kind is EventKind.MEETING] + today_count = len(today_events) + week_events = [e for e in panel.events() if e.kind is EventKind.MEETING] + week_count = len(week_events) + if view is CalendarView.DAY: + return f"{format_count(today_count, 'meeting')} today, hour by hour" + if view is CalendarView.WEEK: + return f"{format_count(week_count, 'meeting')} across the loaded weeks, side by side" + if view is CalendarView.INVITES: + return "just what still needs a yes, no, or maybe" + return f"{format_count(week_count, 'meeting')} coming up, one line each" + + +def _format_destinations(panel: CalendarPanel, cursor: int, compact: bool) -> list[Text]: + lines: list[Text] = [] + invite_count = len(panel.invites()) + for index, destination in enumerate(_DESTINATIONS): + selected = index == cursor + head = Text() + if selected: + head.append(f"{SELECTED_MARK} ", style="bold") + head.append(destination.label, style="bold") + else: + head.append(f" {destination.label}") + if destination.view is CalendarView.INVITES: + head.append(f" ({invite_count})", style=RED) + if selected: + head.append(f" {_ENTER_GLYPH} to open", style="dim") + lines.append(head) + description = _description_for(destination.view, panel) + lines.append(Text(f" {description}", style="dim")) + if not compact and index < len(_DESTINATIONS) - 1: + lines.append(Text()) + return lines + + +def _center(lines: list[Text], width: int) -> list[Text]: + cell_lens = [line.cell_len for line in lines] + block_width = max(cell_lens, default=0) + indent = max(0, (width - block_width) // 2) + centered: list[Text] = [] + for line in lines: + row = Text(" " * indent) + row.append_text(line) + centered.append(row) + return centered + + +class CalendarMenu(Static, HostedView): + BINDINGS = [ + Binding("up", "previous_destination", "select destination", show=False), + Binding("down", "next_destination", "select destination", show=False), + Binding("enter", "open_destination", "open the selected view", show=False), + Binding("o", "open_home", "open Google Calendar in the browser", show=False), + ] + can_focus = True + DEFAULT_CSS = """ + CalendarMenu { height: 1fr; content-align: center middle; } + """ + + def __init__(self, panel: CalendarPanel) -> None: + super().__init__(markup=False) + self.panel = panel + self.cursor = 0 + self.elapsed = 0.0 + self.dots_clock = FrameClock(self, _DOTS_FPS, self._tick) + + def on_show(self) -> None: + self.dots_clock.start() + + def on_hide(self) -> None: + self.dots_clock.stop() + + def _tick(self, elapsed: float) -> None: + self.elapsed = elapsed + self.refresh() + + def render(self) -> RenderResult: + if self.panel.state is PanelState.READY: + return self.render_content(self.size.width) + return self.panel.body_text() + + def render_content(self, width: int) -> RenderableType: + compact = width < _COMPACT_MIN_COLUMNS + now = self.panel.now() + next_meeting = self.panel.next_meeting() + rows: list[Text] = [] + rows.extend(_center([_format_header(now)], width)) + if not compact: + rows.append(Text()) + rows.extend(_center(_format_icon(now.day), width)) + rows.append(Text()) + rows.extend(_center(_format_countdown(next_meeting, now), width)) + rows.extend(_center([_format_dots(self.elapsed)], width)) + rows.extend(_center(_format_destinations(self.panel, self.cursor, compact), width)) + return Group(*rows) + + def content_lines(self) -> list[str]: + return plain_lines(self.render_content(self.size.width), self.size.width) + + def refresh_content(self) -> None: + self.refresh() + + def action_previous_destination(self) -> None: + self.cursor = step_cursor(self.cursor, -1, len(_DESTINATIONS)) + self.refresh() + + def action_next_destination(self) -> None: + self.cursor = step_cursor(self.cursor, 1, len(_DESTINATIONS)) + self.refresh() + + def action_open_destination(self) -> None: + destination = _DESTINATIONS[self.cursor] + if destination.view not in self.panel.view_classes(): + self.notify("coming in the next milestone") + return + self.panel.show_view(destination.view) + + def action_open_home(self) -> None: + webbrowser.open(CALENDAR_HOME) diff --git a/tests/integrations/gcal/__init__.py b/tests/integrations/gcal/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integrations/gcal/fixtures/calendar_list.json b/tests/integrations/gcal/fixtures/calendar_list.json new file mode 100644 index 0000000..97f96e9 --- /dev/null +++ b/tests/integrations/gcal/fixtures/calendar_list.json @@ -0,0 +1,12 @@ +{ + "items": [ + {"id": "lucas@example.com", "summary": "Work", "backgroundColor": "#9fe1e7", + "accessRole": "owner", "selected": true, "primary": true}, + {"id": "c_team@group.calendar.google.com", "summary": "Infra team", + "backgroundColor": "#b99aff", "accessRole": "owner", "selected": true}, + {"id": "en.usa#holiday@group.v.calendar.google.com", "summary": "US Holidays", + "backgroundColor": "#16a765", "accessRole": "reader", "selected": true}, + {"id": "ignored@group.calendar.google.com", "summary": "Hidden", + "backgroundColor": "#000000", "accessRole": "reader", "selected": false} + ] +} diff --git a/tests/integrations/gcal/fixtures/events_work.json b/tests/integrations/gcal/fixtures/events_work.json new file mode 100644 index 0000000..aa77dac --- /dev/null +++ b/tests/integrations/gcal/fixtures/events_work.json @@ -0,0 +1,32 @@ +{ + "items": [ + {"id": "standup_20260914", "status": "confirmed", "summary": "Infra standup", + "htmlLink": "https://www.google.com/calendar/event?eid=standup", "updated": "2026-09-10T08:00:00.000Z", + "start": {"dateTime": "2026-09-14T12:00:00-04:00", "timeZone": "America/Toronto"}, + "end": {"dateTime": "2026-09-14T12:30:00-04:00", "timeZone": "America/Toronto"}, + "recurringEventId": "standup", "hangoutLink": "https://meet.google.com/abc-defg-hij", + "organizer": {"email": "erol@example.com", "displayName": "Erol Schmidt"}, + "attendees": [ + {"email": "erol@example.com", "displayName": "Erol Schmidt", "organizer": true, "responseStatus": "accepted"}, + {"email": "lucas@example.com", "self": true, "responseStatus": "needsAction"} + ], + "description": "Agenda
- mocks
- landing", + "attachments": [{"title": "notes.pdf", "fileUrl": "https://drive.google.com/x"}], + "eventType": "default"}, + {"id": "office_20260914", "status": "confirmed", "summary": "Office", + "htmlLink": "https://www.google.com/calendar/event?eid=office", "updated": "2026-09-01T08:00:00.000Z", + "start": {"date": "2026-09-14"}, "end": {"date": "2026-09-15"}, + "eventType": "workingLocation", "workingLocationProperties": {"type": "officeLocation"}}, + {"id": "declined_1", "status": "confirmed", "summary": "Skip me", + "htmlLink": "https://www.google.com/calendar/event?eid=declined", "updated": "2026-09-01T08:00:00.000Z", + "start": {"dateTime": "2026-09-15T10:00:00-07:00"}, "end": {"dateTime": "2026-09-15T11:00:00-07:00"}, + "attendees": [{"email": "lucas@example.com", "self": true, "responseStatus": "declined"}], + "eventType": "default"}, + {"id": "cancelled_1", "status": "cancelled", + "start": {"dateTime": "2026-09-15T10:00:00-07:00"}, "end": {"dateTime": "2026-09-15T11:00:00-07:00"}}, + {"id": "dentist_1", "status": "confirmed", + "htmlLink": "https://www.google.com/calendar/event?eid=dentist", "updated": "2026-09-01T08:00:00.000Z", + "start": {"dateTime": "2026-09-14T16:00:00-07:00"}, "end": {"dateTime": "2026-09-14T16:45:00-07:00"}, + "location": "Bay Dental, 3rd Ave", "eventType": "default"} + ] +} diff --git a/tests/integrations/gcal/helpers.py b/tests/integrations/gcal/helpers.py new file mode 100644 index 0000000..6e8d854 --- /dev/null +++ b/tests/integrations/gcal/helpers.py @@ -0,0 +1,172 @@ +"""Shared fixtures for Google Calendar panel and view tests.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo + +from textual.app import App, ComposeResult + +from smorg.core.contract import Item +from smorg.core.state import SeenState +from smorg.integrations.gcal.panel import CalendarPanel +from smorg.integrations.gcal.source import ( + CALENDARS_ID, + Attendee, + Calendar, + Calendars, + Event, + EventKind, + Response, +) +from smorg.shell.panel import PanelState + +PACIFIC = ZoneInfo("America/Los_Angeles") +NOW = datetime(2026, 9, 14, 11, 42, tzinfo=PACIFIC) +WORK = Calendar(id="work", name="Work", color="#9fe1e7", primary=True) +TEAM = Calendar(id="team", name="Infra team", color="#b99aff", primary=False) +PERSONAL = Calendar(id="personal", name="Personal", color="#4986e7", primary=False) + + +def calendars(*entries: Calendar) -> Calendars: + if not entries: + entries = (WORK, TEAM, PERSONAL) + return Calendars( + id=CALENDARS_ID, + updated_at=NOW, + url="https://calendar.google.com", + timezone="America/Los_Angeles", + calendars=entries, + ) + + +def at(day_offset: int, hour: int, minute: int = 0) -> datetime: + day = NOW.date() + timedelta(days=day_offset) + return datetime(day.year, day.month, day.day, hour, minute, tzinfo=PACIFIC) + + +def event( + identifier: str, + start: datetime, + end: datetime, + title: str | None = None, + calendar: Calendar = WORK, + kind: EventKind = EventKind.MEETING, + my_response: Response = Response.ACCEPTED, + attendees: tuple[Attendee, ...] = (), + meet_url: str = "", + location: str = "", + all_day: bool = False, + description: str = "", +) -> Event: + if title is None: + title = f"title of {identifier}" + return Event( + id=identifier, + updated_at=NOW, + url=f"https://www.google.com/calendar/event?eid={identifier}", + calendar_id=calendar.id, + title=title, + start=start, + end=end, + all_day=all_day, + kind=kind, + my_response=my_response, + attendees=attendees, + organizer="Erol Schmidt", + meet_url=meet_url, + location=location, + description=description, + attachments=(), + recurring=False, + ) + + +def attendee(name: str, response: Response, is_self: bool = False) -> Attendee: + email = name.casefold().replace(" ", ".") + "@example.com" + return Attendee(name=name, email=email, response=response, organizer=False, is_self=is_self) + + +def monday_week() -> tuple[Event, ...]: + """The fixture week the mockups used: Monday 14 with five meetings and an office day.""" + return ( + event( + "office", at(0, 0), at(1, 0), "Office", kind=EventKind.WORKING_LOCATION, all_day=True + ), + event( + "standup", + at(0, 9), + at(0, 9, 30), + "Infra standup", + calendar=TEAM, + meet_url="https://meet.google.com/a", + ), + event( + "erol", at(0, 10, 30), at(0, 11), "Lucas / Erol", meet_url="https://meet.google.com/b" + ), + event( + "design", + at(0, 12, 30), + at(0, 13), + "Design sync: calendar tab", + my_response=Response.NEEDS_ACTION, + meet_url="https://meet.google.com/c", + attendees=( + attendee("Erol Schmidt", Response.ACCEPTED), + attendee("Melissa Cao", Response.TENTATIVE), + attendee("Lucas Delvoye", Response.NEEDS_ACTION, is_self=True), + ), + ), + event( + "hours", + at(0, 14), + at(0, 15), + "Infra Office Hours", + calendar=TEAM, + my_response=Response.NEEDS_ACTION, + meet_url="https://meet.google.com/d", + ), + event( + "dentist", + at(0, 16), + at(0, 16, 45), + "Dentist", + calendar=PERSONAL, + location="Bay Dental, 3rd Ave", + my_response=Response.NONE, + ), + event("standup-tue", at(1, 9), at(1, 9, 30), "Infra standup", calendar=TEAM), + event("migration", at(1, 11), at(1, 12), "Migration review: pk swap"), + event("market", at(5, 9), at(5, 10), "Farmers market", calendar=PERSONAL), + ) + + +def panel_with(*items: Item, seen: SeenState | None = None) -> CalendarPanel: + panel = CalendarPanel() + panel.state = PanelState.READY + panel.items = items + if seen is None: + panel.seen = SeenState({}) + else: + panel.seen = seen + panel.integration_id = "gcal" + panel.now_provider = lambda: NOW + return panel + + +def week_panel() -> CalendarPanel: + return panel_with(calendars(), *monday_week()) + + +class PanelHarness(App[None]): + """The smallest app that can mount a `CalendarPanel` and hand it focus.""" + + def __init__(self, panel: CalendarPanel) -> None: + super().__init__() + self._panel = panel + + def compose(self) -> ComposeResult: + yield self._panel + + def on_mount(self) -> None: + self._panel.focus() diff --git a/tests/integrations/gcal/test_menu.py b/tests/integrations/gcal/test_menu.py new file mode 100644 index 0000000..3d238a9 --- /dev/null +++ b/tests/integrations/gcal/test_menu.py @@ -0,0 +1,59 @@ +import pytest + +from smorg.integrations.gcal.views import CalendarView +from smorg.integrations.gcal.views.menu import CalendarMenu, next_meeting_label + +from .helpers import NOW, PanelHarness, at, calendars, event, panel_with, week_panel + + +def _text(menu: CalendarMenu) -> str: + return "\n".join(menu.content_lines()) + + +def test_the_countdown_has_three_shapes(): + today = event("a", at(0, 12, 30), at(0, 13)) + tomorrow = event("b", at(1, 9), at(1, 9, 30)) + later = event("c", at(3, 9), at(3, 9, 30)) + assert next_meeting_label(today, NOW) == "in 48 min" + assert next_meeting_label(tomorrow, NOW) == "tomorrow · 09:00" + assert next_meeting_label(later, NOW) == "Thu · 09:00" + + +def test_the_panel_picks_the_next_meeting_skipping_all_day_and_working_location(): + panel = week_panel() + next_meeting = panel.next_meeting() + assert next_meeting is not None + assert next_meeting.id == "design" + assert [invite.id for invite in panel.invites()] == ["design", "hours"] + + +@pytest.mark.asyncio +async def test_the_landing_shows_the_date_icon_countdown_dots_and_destinations(monkeypatch): + panel = week_panel() + async with PanelHarness(panel).run_test(size=(120, 40)) as pilot: + await pilot.pause() + menu = panel.query_one(CalendarMenu) + text = _text(menu) + assert "Monday, September 14" in text + assert "██" in text + assert "Design sync: calendar tab" in text + assert "in 48 min" in text + assert "awaiting your reply" in text + assert "● ● ● ●" in text + assert "▸ today" in text + assert "invites awaiting your reply (2)" in text + + toasts: list[str] = [] + monkeypatch.setattr(menu, "notify", lambda message, **kwargs: toasts.append(message)) + await pilot.press("down", "enter") + assert panel.active_view is CalendarView.MENU + assert toasts == ["coming in the next milestone"] + + +@pytest.mark.asyncio +async def test_with_nothing_left_the_countdown_reaches_into_tomorrow(): + late = panel_with(calendars(), event("b", at(1, 9), at(1, 9, 30), "Standup")) + late.now_provider = lambda: NOW.replace(hour=17, minute=30) + async with PanelHarness(late).run_test(size=(120, 40)) as pilot: + await pilot.pause() + assert "tomorrow · 09:00" in _text(late.query_one(CalendarMenu)) diff --git a/tests/integrations/gcal/test_source.py b/tests/integrations/gcal/test_source.py new file mode 100644 index 0000000..1c733ca --- /dev/null +++ b/tests/integrations/gcal/test_source.py @@ -0,0 +1,191 @@ +import json +from datetime import datetime +from pathlib import Path +from urllib.parse import parse_qs +from zoneinfo import ZoneInfo + +import httpx +import pytest + +from smorg.auth.store import Credentials +from smorg.core.contract import AuthExpired +from smorg.integrations.gcal.source import ( + CALENDARS_ID, + Calendars, + Event, + EventKind, + Response, + fetch, +) + +FIXTURES = Path(__file__).parent / "fixtures" +CALENDAR_LIST = json.loads((FIXTURES / "calendar_list.json").read_text()) +TIMEZONE = {"kind": "calendar#setting", "id": "timezone", "value": "America/Los_Angeles"} +NO_EVENTS = {"items": []} + +CREDENTIALS = Credentials( + access_token="gcal-secret-token", refresh_token=None, expires_at=None, scope="" +) + + +class _Server: + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + self.timezone: tuple[int, object] = (200, TIMEZONE) + self.calendar_list: tuple[int, object] = (200, CALENDAR_LIST) + self.events: dict[str, tuple[int, object]] = {} + + def handle(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + path = request.url.raw_path.decode() + if path.startswith("/calendar/v3/users/me/settings/timezone"): + status, payload = self.timezone + elif path.startswith("/calendar/v3/users/me/calendarList"): + status, payload = self.calendar_list + elif path.startswith("/calendar/v3/calendars/"): + encoded = path.removeprefix("/calendar/v3/calendars/").split("/events")[0] + status, payload = self.events.get(encoded, (200, NO_EVENTS)) + else: + raise AssertionError(f"unexpected request to {request.url}") + return httpx.Response(status, json=payload) + + def client(self) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(self.handle)) + + +@pytest.fixture +def server() -> _Server: + return _Server() + + +def test_fetch_reads_the_timezone_and_only_the_selected_calendars(server): + items = fetch(CREDENTIALS, server.client()) + + calendars = items[0] + assert isinstance(calendars, Calendars) + assert calendars.id == CALENDARS_ID + assert calendars.timezone == "America/Los_Angeles" + assert [calendar.name for calendar in calendars.calendars] == [ + "Work", + "Infra team", + "US Holidays", + ] + assert calendars.calendars[0].primary is True + assert calendars.calendars[1].color == "#b99aff" + + +def test_a_calendar_id_with_a_hash_is_encoded_in_the_events_path(server): + fetch(CREDENTIALS, server.client()) + + paths = [request.url.raw_path.decode() for request in server.requests] + assert any( + "/calendars/en.usa%23holiday%40group.v.calendar.google.com/events" in p for p in paths + ) + + +def test_events_are_requested_for_the_two_week_window_as_single_instances(server): + fetch(CREDENTIALS, server.client()) + + event_requests = [r for r in server.requests if "/events" in r.url.path] + query = parse_qs(event_requests[0].url.query.decode()) + assert query["singleEvents"] == ["true"] + assert query["orderBy"] == ["startTime"] + assert query["maxResults"] == ["250"] + assert "timeMin" in query and "timeMax" in query + + +def test_a_401_is_auth_expired(server): + server.timezone = (401, {"error": {"code": 401}}) + + with pytest.raises(AuthExpired): + fetch(CREDENTIALS, server.client()) + + +EVENTS_WORK = json.loads((FIXTURES / "events_work.json").read_text()) +WORK = "lucas%40example.com" +PACIFIC = ZoneInfo("America/Los_Angeles") + + +def _events(server: _Server) -> list[Event]: + items = fetch(CREDENTIALS, server.client()) + return [item for item in items if isinstance(item, Event)] + + +def test_events_land_in_the_settings_timezone_with_every_field_mapped(server): + server.events[WORK] = (200, EVENTS_WORK) + + events = _events(server) + standup = next(event for event in events if event.id == "standup_20260914") + + assert standup.start == datetime(2026, 9, 14, 9, 0, tzinfo=PACIFIC) + assert standup.end == datetime(2026, 9, 14, 9, 30, tzinfo=PACIFIC) + assert standup.all_day is False + assert standup.kind is EventKind.MEETING + assert standup.my_response is Response.NEEDS_ACTION + assert standup.recurring is True + assert standup.meet_url == "https://meet.google.com/abc-defg-hij" + assert standup.organizer == "Erol Schmidt" + assert [attendee.response for attendee in standup.attendees] == [ + Response.ACCEPTED, + Response.NEEDS_ACTION, + ] + assert standup.attendees[1].is_self is True + assert standup.description == "Agenda\n- mocks\n- landing" + assert standup.attachments == ("notes.pdf",) + assert standup.calendar_id == "lucas@example.com" + + +def test_all_day_events_are_midnight_to_exclusive_midnight_and_keep_their_kind(server): + server.events[WORK] = (200, EVENTS_WORK) + + office = next(event for event in _events(server) if event.id == "office_20260914") + + assert office.all_day is True + assert office.start == datetime(2026, 9, 14, 0, 0, tzinfo=PACIFIC) + assert office.end == datetime(2026, 9, 15, 0, 0, tzinfo=PACIFIC) + assert office.kind is EventKind.WORKING_LOCATION + + +def test_declined_and_cancelled_instances_are_dropped_and_untitled_events_are_named(server): + server.events[WORK] = (200, EVENTS_WORK) + + ids = [event.id for event in _events(server)] + dentist = next(event for event in _events(server) if event.id == "dentist_1") + + assert "declined_1" not in ids + assert "cancelled_1" not in ids + assert dentist.title == "(no title)" + assert dentist.my_response is Response.NONE + assert dentist.location == "Bay Dental, 3rd Ave" + + +def test_a_404_calendar_is_skipped_while_the_others_load(server): + server.events[WORK] = (200, EVENTS_WORK) + server.events["c_team%40group.calendar.google.com"] = (404, {"error": {"code": 404}}) + + events = _events(server) + + assert len(events) == 3 + + +def test_events_are_sorted_by_start_across_calendars(server): + server.events[WORK] = (200, EVENTS_WORK) + server.events["c_team%40group.calendar.google.com"] = ( + 200, + { + "items": [ + { + "id": "early", + "status": "confirmed", + "summary": "Early", + "htmlLink": "https://www.google.com/calendar/event?eid=early", + "updated": "2026-09-01T08:00:00.000Z", + "start": {"dateTime": "2026-09-14T07:00:00-07:00"}, + "end": {"dateTime": "2026-09-14T07:30:00-07:00"}, + "eventType": "default", + } + ] + }, + ) + + assert [event.id for event in _events(server)][:2] == ["office_20260914", "early"] diff --git a/tests/test_cli.py b/tests/test_cli.py index 855550d..5e9d117 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -80,7 +80,7 @@ def test_version_flag_omits_dev_on_a_release_build(monkeypatch, capsys): def test_the_allowlist_is_what_this_build_registers(): - assert known_integration_ids() == ("github", "linear", "spotify") + assert known_integration_ids() == ("gcal", "github", "linear", "spotify") def test_connect_rejects_an_unknown_integration(capsys):