Skip to content

Combining handlers

This bot showcases combining two Handler subclasses — DataMessageHandler and GroupUpdateHandler — on a single class via multiple inheritance, so they can share state directly.

import os

from examples.timestamps import local_datetime_str_from_timestamp
from signalbot import (
    Config,
    DataMessageContext,
    DataMessageHandler,
    GroupUpdateContext,
    GroupUpdateHandler,
    SendMessage,
    SignalBot,
    text_triggered,
)


class GroupActivityHandler(DataMessageHandler, GroupUpdateHandler):
    """Tracks the most recent group update and answers a query about it.

    `handle_group_update` records who changed the group and when;
    `handle_data_message` answers the `last-change` command by reading that
    same cache. Combining both handlers on one instance is what lets them
    share `_last_update` directly instead of through a database or some other
    out-of-process channel.
    """

    def __init__(self) -> None:
        super().__init__()
        self._last_update: dict[str, tuple[str, str]] = {}

    async def handle_group_update(self, context: GroupUpdateContext) -> None:
        group_info = context.message.group_info
        if group_info.group_id is None:
            return

        who = context.message.source_name or context.message.source_number
        when = local_datetime_str_from_timestamp(context.message.timestamp)
        self._last_update[group_info.group_id] = (who or "someone", when)

    @text_triggered("last-change")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        group_info = context.message.group_info
        group_id = group_info.group_id if group_info is not None else None

        if group_id is None or group_id not in self._last_update:
            await context.send(SendMessage(text="No group changes seen yet."))
            return

        who, when = self._last_update[group_id]
        message = f"{who} last changed this group at {when}."
        await context.send(SendMessage(text=message))


if __name__ == "__main__":
    bot = SignalBot(Config(phone_number=os.environ["PHONE_NUMBER"]))
    bot.register(GroupActivityHandler(), contacts=False, groups=True)
    bot.start()