# 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 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 recipient: str transport: Transport store: Store config: Config command_name: str now: float day: str async def reply(self, text: str) -> None: """Send a text reply to the configured reply target.""" await self.transport.send_text(text, self.recipient) 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(t.strip().lower() 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 candidate = text.strip().lower() 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)