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 and every phrase that triggers it."""

    name: str
    triggers: frozenset[str]
    handler: Handler


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


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) -> Command:
        """Register one command under every one of its triggers.

        Raises:
            DuplicateTriggerError: A trigger is already claimed.
        """
        normalised = frozenset(fold(t) for t in triggers)
        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)
        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)