signal-bot/signal_bot/command.py
Krzysztof kuhy Rudnicki 72e90acb47
Some checks are pending
CI / check (push) Waiting to run
CI / web (push) Waiting to run
Let the operator add aliases from the Commands tab
A custom alias is data the command table is built from, so it cannot be a
live switch like the ones already on the tab. Instead the management page
writes aliases_custom.json in the bot's state directory and then asks the
bot to reload, in one step: whenever the answer is anything but `reloaded`
-- a word another command has, a name no command has, a bot still
connecting -- the file is restored byte for byte and the bot's own reason is
shown next to the field. What is on disk is therefore always what the bot
serves, and a typo cannot sit in the file poisoning every later reload.

The registry appends the operator's words to the shipped list at
registration, which keeps `!pomoc` on the shipped spelling and routes the
new words through the same clash check as aliases.json. An alias for a name
no command has fails the whole build (ConfigError -> `kept`), because a
silently ignored entry is how a file drifts. The tagger treats custom
aliases as reserved words too, so a person cannot be given one.

registry.py crossed the 250-line cap and now re-exports the record and the
error types from command.py; the bot's state directory lookup, previously
duplicated in pack.py and the suggestions export, is state_dir.py.

Proven end to end: !e2ekot is unknown, the file is written and reload.sh
runs, !e2ekot posts a cat, a file naming a command that does not exist is
refused with 409 and the previous table still answers.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-04 16:20:05 +02:00

105 lines
4.4 KiB
Python

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What a command *is*: the record, its handler type, and the ways it says no.
Split from ``registry.py`` so the lookup and the vocabulary each fit on one
screen; ``registry`` re-exports every name here, so callers see one module.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from signal_bot.context import CommandContext
COMMAND_PREFIX = "!"
Handler = Callable[[CommandContext], Awaitable[None]]
# Turns the words after a trigger into whatever the handler is allowed to see.
# For the pack that means pseudonym keys and nothing else: the raw remainder
# stops here, so no handler, log line or command_uses row ever holds a name
# somebody typed. A command without one gets the words unchanged, which is
# what every command before this feature did.
ArgResolver = Callable[[tuple[str, ...]], tuple[str, ...]]
@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
# The only groups this command exists in, or None for everywhere. Not a
# gate inside the handler: the dispatcher treats a command that is not
# allowed here as no command at all, so it is never answered, never
# counted as a command and never logged by name in a group it does not
# belong to. A silent handler would still leak all three.
groups: frozenset[str] | None = None
# Whether words after the trigger are arguments rather than nonsense.
# Off by default, so "!cat blah" keeps resolving to nothing exactly as
# it does today.
takes_args: bool = False
# Applied to those words before they reach CommandContext.args. The
# scrubbing seam -- see ArgResolver.
arg_resolver: ArgResolver | None = None
def allowed_in(self, group_id: str) -> bool:
"""Whether this command exists in one group."""
return self.groups is None or group_id in self.groups
class DuplicateTriggerError(ValueError):
"""Raised when two commands that can both run claim the same trigger word.
"Can both run" is the whole test. Two albums scoped to different groups
may each define "!alfa" -- they are different people, and no message can
ever mean both, because a command a group is not scoped to is not a
command there at all. A collision is only a collision where the group
sets overlap, and then it is fatal at startup rather than a coin toss.
"""
class CommandRefusedError(Exception):
"""A gate declined the command, so nothing happened and no *use* 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.
The refusal is now recorded in the command *history*, which is a different
table and a different question: how often a gate actually bites is what
says whether it is tuned right. ``gate`` names which one refused, because
"this command is refused 40% of the time" and "this command is popular"
are the two readings of the same raw count.
It defaults to empty so that raising it bare stays legal -- every existing
raise site said only "no", and a required argument would turn a behaviour
that works into a signature change across three decorators.
"""
def __init__(self, gate: str = "") -> None:
"""Record which gate refused, if the caller says."""
super().__init__(gate)
self.gate = gate
@dataclass(frozen=True, slots=True)
class Resolved:
"""A command a message fired, and the words that followed it."""
command: Command
args: tuple[str, ...]