hermes-chatto-plugin/vendor/common/chattolib/bot.py

"""A small, opinionated framework for building bots on Chatto.

Chatto 0.5.0 introduced *bot accounts*: user identities flagged as bots that
authenticate with a **key** (e.g. ``cht_BK_...``) rather than a
username/password. The key is used **directly as a bearer token** — there is
no ``/auth/login`` round-trip — and it resolves to a ``User`` with a
capability grant set.

This module turns :class:`chattolib.client.ChattoClient` into the standard
library for bots by adding three things on top of the raw client:

* **Key-based login** — :meth:`Bot.login` builds an authenticated client from
  a bot key in one call.
* **An event dispatcher** — :meth:`Bot.run` opens the realtime stream and
  routes incoming events (messages, reactions, mentions, presence, typing,
  room changes) to the async handlers you register.
* **Flattened verbs** — :meth:`Bot.say`, :meth:`Bot.reply`,
  :meth:`Bot.react`, :method:`Bot.set_status`, :meth:`Bot.join_room`,
  :meth:`Bot.create_room`, and friends, so a bot's "brain" reads like
  natural language instead of protobuf plumbing.

In the protocol-4 realtime channel, live events are thin, caller-scoped hints
(identifiers rather than full resources). Where a handler needs the full
resource (e.g. the body of a posted message), the bot hydrates it on demand
through the corresponding ConnectRPC.

A minimal bot::

    import asyncio
    from chattolib.bot import Bot

    async def on_message(event):
        if event.body.startswith("!hello"):
            await event.bot.reply(event, "hi!")

    async def main():
        async with await Bot.login("cht_BK_...") as bot:
            bot.on("message", on_message)
            await bot.run()

    asyncio.run(main())

Requires the ``chattolib[realtime]`` extra for the live stream.
"""

from __future__ import annotations

import asyncio
import contextlib
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any

from chattolib import _pb  # noqa: F401 — installs the generated pb import path
from chattolib._pb.chatto.api.v1 import presence_pb2
from chattolib.client import ChattoClient
from chattolib.exceptions import ChattoError
from chattolib.realtime import (
    ChattoRealtimeCloseError,
    ChattoRealtimeError,
    RealtimeConnection,
    RealtimeEvent,
    stream_events,
)
from chattolib.types import (
    Message,
    PresencePreference,
    PresenceStatus,
    Room,
    RoomGroup,
    RoomWithViewerState,
    User,
)


def _presence(value: Any) -> PresenceStatus:
    """Coerce a protobuf presence value (int or enum) to a PresenceStatus."""
    name = value.name if hasattr(value, "name") else presence_pb2.PresenceStatus.Name(int(value))
    # The protobuf name is the *value* of our StrEnum (e.g. "PRESENCE_STATUS_ONLINE"),
    # so look it up by value, not by member name.
    for member in PresenceStatus:
        if member.value == name:
            return member
    return PresenceStatus.UNSPECIFIED


__all__ = [
    "Bot",
    "BotError",
    "BotEvent",
    "BotMessageEvent",
    "BotPresenceEvent",
    "BotPresencePreferenceEvent",
    "BotReactionEvent",
    "BotRoomEvent",
    "BotRoomInvitationEvent",
    "BotTypingEvent",
    "BotUserEvent",
]

# Oneof ``event`` cases that map onto a :class:`BotRoomEvent`.
_ROOM_EVENT_KINDS = frozenset(
    {
        "room_created",
        "room_updated",
        "room_deleted",
        "room_archived",
        "room_unarchived",
        "room_universal_changed",
        "room_slow_mode_changed",
        "room_threading_mode_changed",
        "user_joined_room",
        "user_left_room",
    }
)
# Oneof ``event`` cases that map onto a :class:`BotUserEvent`.
_USER_EVENT_KINDS = frozenset(
    {"user_account_created", "user_profile_changed", "user_account_deleted"}
)
# Oneof ``event`` cases that map onto a :class:`BotRoomInvitationEvent`
# (invitation lifecycle, since 0.5.0b20; delivered only to sender+recipient).
_INVITATION_EVENT_KINDS = frozenset(
    {"room_invitation_created", "room_invitation_accepted", "room_invitation_declined"}
)


class BotError(ChattoError):
    """Raised for bot-framework-level errors (bad handlers, bad keys, ...)."""


# ---------------------------------------------------------------------------
# Event types
# ---------------------------------------------------------------------------


@dataclass
class BotEvent:
    """Base class for every event the dispatcher delivers to a handler.

    ``bot`` is the owning :class:`Bot`, so a handler can act on the event
    (e.g. ``await event.bot.reply(event, "...")``) without closing over
    globals.
    """

    bot: Bot
    kind: str

    async def _noop(self) -> None:  # pragma: no cover - interface
        ...


@dataclass
class BotMessageEvent(BotEvent):
    """A new message in a room the bot can see.

    ``message`` is the full :class:`Message`, hydrated from the server when
    this event was dispatched. ``actor`` is the author's :class:`User` when
    it could be resolved; otherwise ``None``.
    """

    message: Message
    actor: User | None = None

    @property
    def room_id(self) -> str:
        return self.message.room_id

    @property
    def body(self) -> str | None:
        return self.message.body

    @property
    def is_mention(self) -> bool:
        """True when this message mentions the bot (best-effort)."""
        me = self.bot.user
        if me is None or not self.message.body:
            return False
        return f"@{me.login}" in self.message.body or me.display_name in self.message.body


@dataclass
class BotReactionEvent(BotEvent):
    """A reaction was added to (or removed from) a message."""

    room_id: str
    message_event_id: str
    emoji: str
    user_id: str
    added: bool = True


@dataclass
class BotPresenceEvent(BotEvent):
    """A user's presence status changed."""

    user_id: str
    status: PresenceStatus


@dataclass
class BotPresencePreferenceEvent(BotEvent):
    """The bot's own private availability *preference* changed (0.5.0b6).

    Delivered only to the account whose own choice changed. The wire payload is
    an empty hint: to read the new choice, hydrate it from the bot via
    ``await event.bot.get_presence_preference()`` (the current
    :class:`~chattolib.types.PresencePreference`).
    """


@dataclass
class BotTypingEvent(BotEvent):
    """A user started (or stopped) typing in a room/thread."""

    room_id: str
    thread_root_event_id: str | None = None


@dataclass
class BotRoomEvent(BotEvent):
    """A room lifecycle change (created, updated, archived, member join/...)."""

    room: Room | None = None
    detail: str = ""  # e.g. "created", "updated", "archived", "user_joined_room"


@dataclass
class BotUserEvent(BotEvent):
    """A user account was created, changed, or deleted."""

    user: User | None = None
    removed: bool = False


@dataclass
class BotRoomInvitationEvent(BotEvent):
    """A room invitation was created, accepted, or declined (since 0.5.0b20).

    Delivered only to the invitation's sender and recipient. ``detail`` is
    ``"created"``, ``"accepted"``, or ``"declined"``. The payload is a thin
    hint (IDs only); hydrate the full invitation via
    ``await event.bot.client.get_room_invitation(event.invitation_id)`` when
    the recipient is the bot itself.
    """

    room_id: str = ""
    invitation_id: str = ""
    detail: str = ""


# ---------------------------------------------------------------------------
# The Bot
# ---------------------------------------------------------------------------


Handler = Callable[[BotEvent], Awaitable[None]]


class Bot:
    """A Chatto bot: an authenticated client plus an event dispatcher and a
    set of convenience verbs.

    Create one with :meth:`login` (from a bot key) or :meth:`from_client`
    (wrapping an already-authenticated :class:`ChattoClient`).
    """

    def __init__(
        self,
        client: ChattoClient,
        *,
        base_url: str | None = None,
    ) -> None:
        self._client = client
        self._base_url = base_url or client.base_url
        self._user: User | None = None
        self._handlers: dict[str, list[Handler]] = {}
        self._connection: RealtimeConnection | None = None
        self._running = False

    # -- construction ------------------------------------------------------

    @classmethod
    async def login(
        cls,
        key: str,
        *,
        base_url: str | None = None,
    ) -> Bot:
        """Authenticate a bot with its **key**.

        The key (e.g. ``cht_BK_...``) is used directly as the bearer token —
        no username/password. ``base_url`` defaults to the public Chatto
        server; pass it to target a self-hosted or preview deployment.
        """
        if base_url is None:
            base_url = ChattoClient.DEFAULT_BASE_URL
        client = ChattoClient(token=key, base_url=base_url)
        bot = cls(client, base_url=base_url)
        await bot._probe_identity()
        return bot

    @classmethod
    def from_client(cls, client: ChattoClient) -> Bot:
        """Wrap an already-authenticated client (e.g. a human login)."""
        return cls(client)

    async def _probe_identity(self) -> None:
        """Resolve the bot's own identity and warn (not fail) if not a bot."""
        try:
            self._user = await self._client.me()
        except ChattoError:
            self._user = None
            return
        if self._user is not None and not self._user.is_bot:
            # Not fatal — a human client can drive the same verbs — but surface
            # it so a misconfigured key is obvious.
            import warnings

            warnings.warn(
                f"Bot key authenticated as {self._user.login!r}, which is not "
                "flagged is_bot. It will still work, but this is usually a "
                "misconfigured key.",
                stacklevel=2,
            )

    # -- identity ----------------------------------------------------------

    @property
    def client(self) -> ChattoClient:
        """The underlying :class:`ChattoClient` for any RPC not wrapped here."""
        return self._client

    @property
    def base_url(self) -> str:
        return self._base_url

    @property
    def user(self) -> User | None:
        """The bot's own :class:`User` profile, once known."""
        return self._user

    @property
    def login_name(self) -> str | None:
        return self._user.login if self._user else None

    @property
    def display_name(self) -> str | None:
        return self._user.display_name if self._user else None

    # -- lifecycle ---------------------------------------------------------

    async def __aenter__(self) -> Bot:
        return self

    async def __aexit__(self, *exc: Any) -> None:
        await self.close()

    async def close(self) -> None:
        if self._connection is not None:
            with contextlib.suppress(Exception):
                await self._connection.close()
            self._connection = None
        self._running = False
        await self._client.close()

    # -- event registration ------------------------------------------------

    def on(self, kind: str, handler: Handler) -> Handler:
        """Register ``handler`` for events of ``kind``.

        ``kind`` is one of: ``message``, ``reaction``, ``presence``,
        ``presence_preference``, ``typing``, ``room``, ``user``, ``*`` (all
        events). ``handler`` is an
        ``async def`` taking a single :class:`BotEvent`. Returns the handler
        so ``on`` can be used as a decorator.
        """
        self._handlers.setdefault(kind, []).append(handler)
        return handler

    def off(self, kind: str, handler: Handler) -> None:
        with contextlib.suppress(ValueError):
            self._handlers[kind].remove(handler)

    async def _dispatch(self, event: BotEvent) -> None:
        handlers = list(self._handlers.get(event.kind, [])) + list(self._handlers.get("*", []))
        for handler in handlers:
            try:
                await handler(event)
            except Exception:  # noqa: BLE001 - one bad handler must not kill the loop
                import logging

                logging.exception("bot handler for %r raised", event.kind)

    # -- verbs: messaging --------------------------------------------------

    async def say(self, room_id: str, body: str = "", *, join_if_needed: bool = True) -> Message:
        """Post a message to a room. Returns the created :class:`Message`.

        With ``join_if_needed`` (the default), the bot joins the room first if
        it isn't already a member — so a bot that wants to be present
        everywhere can simply ``await bot.say(room_id, ...)`` without a
        separate join step. Pass ``join_if_needed=False`` to instead surface
        the server's ``permission_denied`` if the bot can't post.
        """
        try:
            return await self._client.post_message(room_id, body)
        except ChattoError as exc:
            if not join_if_needed or "not a member" not in str(exc):
                raise
            await self._client.join_room(room_id)
            return await self._client.post_message(room_id, body)

    async def reply(
        self,
        target: BotMessageEvent | Message,
        body: str,
        *,
        also_send_to_channel: bool = False,
    ) -> Message:
        """Reply to a message (or a :class:`BotMessageEvent`) in its room/thread."""
        if isinstance(target, BotMessageEvent):
            target = target.message
        return await self._client.post_message(
            target.room_id,
            body,
            thread_root_event_id=target.thread_root_event_id,
            in_reply_to=target.id,
            also_send_to_channel=also_send_to_channel,
        )

    async def react(self, room_id: str, message_event_id: str, emoji: str) -> bool:
        """Add an emoji reaction to a message."""
        return await self._client.add_reaction(room_id, message_event_id, emoji)

    async def unreact(self, room_id: str, message_event_id: str, emoji: str) -> bool:
        """Remove one of the bot's emoji reactions from a message."""
        return await self._client.remove_reaction(room_id, message_event_id, emoji)

    # -- verbs: presence & status -----------------------------------------

    async def set_presence(self, status: PresenceStatus) -> PresenceStatus:
        """Set the bot's presence (``ONLINE`` / ``AWAY`` / ``DO_NOT_DISTURB``)."""
        return await self._client.set_presence(status)

    async def get_presence_preference(self) -> PresencePreference | None:
        """Read the bot's saved availability choice (or ``None`` if unset)."""
        return await self._client.get_presence_preference()

    async def set_presence_preference(
        self, status: PresenceStatus, *, expected_revision: str = ""
    ) -> PresencePreference:
        """Save the bot's availability choice for every device on the server.

        Unlike :meth:`set_presence` this persists across disconnects and
        accepts ``OFFLINE`` (meaning "Appear Offline"). Echo a prior
        ``revision`` back as ``expected_revision`` to guard against a
        concurrent change.
        """
        return await self._client.set_presence_preference(
            status, expected_revision=expected_revision
        )

    async def refresh_presence(self) -> PresencePreference | None:
        """Refresh the bot's liveness (call every ~30 s to stay "live")."""
        return await self._client.refresh_presence()

    async def set_status(self, emoji: str, text: str) -> dict[str, Any]:
        """Set the bot's custom status (e.g. a "working on X" note)."""
        return await self._client.set_custom_status(emoji, text)

    async def clear_status(self) -> dict[str, Any]:
        """Clear the bot's custom status."""
        return await self._client.delete_custom_status()

    # -- verbs: rooms ------------------------------------------------------

    async def join_room(self, room_id: str) -> Room:
        """Join a room. Returns the joined :class:`Room`."""
        return await self._client.join_room(room_id)

    async def join_room_group(self, group_id: str) -> list[str]:
        """Join **all** the rooms in a room group in one call.

        Mirrors the Chatto UI's one-click "join group" action. Returns the
        list of room IDs the bot is now a member of as a result.
        """
        return await self._client.join_room_group(group_id)

    async def list_room_groups(self) -> list[RoomGroup]:
        """List the room groups (and the rooms each contains) the bot can see."""
        return await self._client.list_room_groups()

    async def join_all_rooms(self) -> list[str]:
        """Join every room the bot can see, grouped the way the UI does.

        Joins each room group in one call (so a bot becomes a member of all
        the rooms in a group at once), then joins any ungrouped rooms
        individually. Returns the room IDs the bot is now a member of.
        """
        joined: list[str] = []
        groups = await self.list_room_groups()
        grouped_room_ids: set[str] = set()
        for group in groups:
            for rws in group.rooms:
                if rws.room is not None:
                    grouped_room_ids.add(rws.room.id)
            try:
                joined.extend(await self.join_room_group(group.id))
            except ChattoError:
                # A group the bot can't join is skipped, not fatal.
                continue
        # Rooms that are not part of any group.
        for rws in await self.list_rooms():
            if rws.room is None or rws.room.id in grouped_room_ids:
                continue
            if rws.viewer_state.is_member:
                continue
            try:
                joined.append((await self.join_room(rws.room.id)).id)
            except ChattoError:
                continue
        return joined

    async def leave_room(self, room_id: str) -> None:
        """Leave a room."""
        await self._client.leave_room(room_id)

    async def create_room(
        self,
        name: str,
        group_id: str,
        *,
        description: str = "",
        universal: bool = False,
    ) -> Room:
        """Create a room in a room group."""
        return await self._client.create_room(
            name, group_id, description=description, universal=universal
        )

    async def list_rooms(self) -> list[RoomWithViewerState]:
        """List the rooms the bot is a member of."""
        return await self._client.list_rooms()

    async def mark_read(self, room_id: str) -> None:
        """Mark a room as read (clears its unread state for the bot)."""
        await self._client.mark_room_as_read(room_id)

    # -- the run loop ------------------------------------------------------

    async def run(
        self,
        *,
        resume_cursor: str | None = None,
        until: asyncio.Event | None = None,
    ) -> None:
        """Connect the realtime stream and dispatch events until it closes.

        Reconnects automatically (resuming from the last received
        ``resume_cursor``) when the server sends a reconnectable close frame.
        When the server terminates the session (or sends a non-reconnectable
        close), the loop stops. Pass ``until`` to stop on an external signal.
        """
        self._running = True
        cursor = resume_cursor
        while self._running:
            if until is not None and until.is_set():
                break
            try:
                async for frame in stream_events(self._client, resume_cursor=cursor):
                    if until is not None and until.is_set():
                        break
                    if isinstance(frame, RealtimeEvent):
                        if frame.cursor:
                            cursor = frame.cursor
                        await self._handle_live(frame)
            except ChattoRealtimeCloseError as close:
                if not close.reconnect:
                    # Session terminated or otherwise unrecoverable: stop.
                    self._running = False
                    break
                # Reconnectable close: loop again, resuming from the cursor.
                continue
            except ChattoRealtimeError:
                raise
        self._running = False

    # -- live-event dispatch -----------------------------------------------

    async def _handle_live(self, frame: RealtimeEvent) -> None:
        kind = frame.kind
        if kind == "presence_changed":
            await self._dispatch(
                BotPresenceEvent(
                    bot=self,
                    kind="presence",
                    user_id=frame.actor_id or "",
                    status=_presence(frame.payload.status),
                )
            )
        elif kind == "viewer_presence_preference_changed":
            await self._dispatch(BotPresencePreferenceEvent(bot=self, kind="presence_preference"))
        elif kind == "user_typing":
            payload = frame.payload
            await self._dispatch(
                BotTypingEvent(
                    bot=self,
                    kind="typing",
                    room_id=payload.room_id,
                    thread_root_event_id=payload.thread_root_event_id or None,
                )
            )
        elif kind in ("reaction_added", "reaction_removed"):
            payload = frame.payload
            await self._dispatch(
                BotReactionEvent(
                    bot=self,
                    kind="reaction",
                    room_id=payload.room_id,
                    message_event_id=payload.message_event_id,
                    emoji=payload.emoji,
                    user_id=frame.actor_id or "",
                    added=kind == "reaction_added",
                )
            )
        elif kind == "message_posted":
            await self._on_message_posted(frame)
        elif kind in _ROOM_EVENT_KINDS:
            await self._on_room_event(frame)
        elif kind in _USER_EVENT_KINDS:
            await self._on_user_event(frame)
        elif kind in _INVITATION_EVENT_KINDS:
            payload = frame.payload
            await self._dispatch(
                BotRoomInvitationEvent(
                    bot=self,
                    kind="room_invitation",
                    room_id=getattr(payload, "room_id", ""),
                    invitation_id=getattr(payload, "invitation_id", ""),
                    detail=frame.kind.removeprefix("room_invitation_"),
                )
            )

    async def _on_message_posted(self, frame: RealtimeEvent) -> None:
        # Protocol-4 events are thin hints; hydrate the full message so the
        # handler gets a usable :class:`Message` (including the body).
        payload = frame.payload
        msg: Message | None = None
        with contextlib.suppress(ChattoError):
            msg = await self._client.get_message(payload.room_id, frame.id)
        if msg is None:
            return
        await self._dispatch(BotMessageEvent(bot=self, kind="message", message=msg, actor=None))

    async def _on_room_event(self, frame: RealtimeEvent) -> None:
        payload = frame.payload
        room_id = getattr(payload, "room_id", "")
        if not room_id:
            return
        extra: dict[str, Any] = {}
        if frame.kind == "room_created":
            name = getattr(payload, "name", "")
            if name:
                extra["name"] = name
        room = Room.parse({"id": room_id, **extra})
        await self._dispatch(BotRoomEvent(bot=self, kind="room", room=room, detail=frame.kind))

    async def _on_user_event(self, frame: RealtimeEvent) -> None:
        user_id = getattr(frame.payload, "user_id", "") or frame.actor_id or ""
        removed = frame.kind == "user_account_deleted"
        await self._dispatch(
            BotUserEvent(
                bot=self,
                kind="user",
                user=User.parse({"id": user_id}) if user_id else None,
                removed=removed,
            )
        )