signal-bot/signal_bot/registry.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Command lookup: trigger words to handlers.

Reimplements the idea behind the unmerged ``rules.py`` branch: triggers are
data, not a chain of if-statements, so aliases (including the Unicode-styled
ones people actually type) cost one tuple entry each.
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable, Iterable
from dataclasses import dataclass
from typing import TYPE_CHECKING

from signal_bot.folding import fold
from signal_bot.quoting import quote_of

if TYPE_CHECKING:
    from signal_bot.config import Config
    from signal_bot.models import IncomingMessage
    from signal_bot.store import Store
    from signal_bot.transport_base import Transport

COMMAND_PREFIX = "!"


@dataclass(frozen=True, slots=True)
class CommandContext:
    """Everything a command handler is allowed to touch."""

    message: IncomingMessage
    group_id: str
    transport: Transport
    store: Store
    config: Config
    command_name: str
    now: float
    day: str

    async def reply(self, text: str) -> None:
        """Answer the message that triggered this command, in its group.

        There is no configurable reply target. A command answers where it
        was asked, which is the only routing that stays correct once the
        bot watches a test group and a production group at once -- and it
        quotes the asker, so a busy group can tell which "!cat" this is.
        """
        await self.transport.send_text(text, self.group_id, quote_of(self.message))

    async def reply_image(self, image_base64: str, caption: str = "") -> None:
        """Answer the triggering message with an image, quoting it."""
        await self.transport.send_image(
            image_base64, self.group_id, quote_of(self.message), caption
        )


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


@dataclass(frozen=True, slots=True)
class Command:
    """A named command, every phrase that triggers it, and how to explain it."""

    name: str
    triggers: frozenset[str]
    handler: Handler
    # Group-facing Polish, one line, for !pomoc. Required at registration so
    # that a command cannot exist without one -- a parallel help table would
    # go stale exactly the way the README's command list did.
    description: str
    # The spelling !pomoc shows. It cannot come from `triggers`: those are
    # folded skeletons in an unordered set, so "!papiez" is all that survives
    # of "!papież". This is the first raw trigger instead, which is the
    # canonical Polish one for every entry in aliases.json.
    primary: str


class DuplicateTriggerError(ValueError):
    """Raised when two commands claim the same trigger word."""


class CommandRefusedError(Exception):
    """A gate declined the command, so nothing happened and nothing is recorded.

    The dispatcher records one use per invocation, and it cannot see inside a
    handler. Without this signal a *refused* run looked exactly like a
    successful one, which slid the cooldown window forward every time somebody
    was told to wait -- so waiting the stated time never helped.
    """


class CommandRegistry:
    """Maps trigger words to commands."""

    def __init__(self) -> None:
        """Start with no commands registered."""
        self._by_trigger: dict[str, Command] = {}
        self._commands: list[Command] = []

    def register(
        self, name: str, triggers: Iterable[str], handler: Handler, description: str
    ) -> Command:
        """Register one command under every one of its triggers.

        Raises:
            DuplicateTriggerError: A trigger is already claimed.
        """
        raw = list(triggers)
        normalised = frozenset(fold(t) for t in raw)
        for trigger in sorted(normalised):
            if trigger in self._by_trigger:
                msg = f"trigger {trigger!r} is already registered"
                raise DuplicateTriggerError(msg)
        command = Command(
            name=name,
            triggers=normalised,
            handler=handler,
            description=description,
            primary=raw[0],
        )
        for trigger in normalised:
            self._by_trigger[trigger] = command
        self._commands.append(command)
        return command

    def resolve(self, text: str | None) -> Command | None:
        """Return the command a message body triggers, or None."""
        if not text:
            return None
        # Folded, not just lowercased: a Cyrillic "o" is a different
        # codepoint from a Latin one and used to resolve to nothing at all.
        candidate = fold(text)
        if not candidate.startswith(COMMAND_PREFIX):
            return None
        return self._by_trigger.get(candidate)

    @property
    def commands(self) -> list[Command]:
        """Every registered command, in registration order."""
        return list(self._commands)