80 lines
3.8 KiB
Python
80 lines
3.8 KiB
Python
import inspect
|
|
import logging
|
|
from typing import Callable
|
|
|
|
from ..ratelimit import FailureLimiter
|
|
from .access import is_allowed
|
|
from .ai import ask_command, explain_command
|
|
from .builtin import help_command, host_command, problems_command, status_command
|
|
from .downtimecmd import cancel_command, confirm_command, downtime_command
|
|
from .constants import (AUDIT_ARGS_MAX, OUTCOME_DENIED, OUTCOME_ERROR, OUTCOME_OK, OUTCOME_RATE_LIMITED,
|
|
OUTCOME_UNKNOWN, PREFIXES, RATE_LIMIT_MAX, RATE_LIMIT_WINDOW, RATE_LIMITED_COMMANDS)
|
|
from .context import AuditEvent, CommandContext, CommandDeps, Registry
|
|
from .models import IncomingMessage, Reply
|
|
from .mutecmd import mute_command
|
|
from .parser import parse
|
|
from .render import plain
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
ERROR_TEXT = "Das hat gerade nicht geklappt. Bitte später erneut versuchen."
|
|
RATE_TEXT = "Zu viele Anfragen. Bitte kurz warten."
|
|
UNKNOWN_TEXT = "Unbekannter Befehl. {p}help zeigt alle Befehle."
|
|
|
|
|
|
def default_registry() -> Registry:
|
|
return {"help": help_command, "status": status_command, "problems": problems_command,
|
|
"host": host_command, "explain": explain_command, "ask": ask_command,
|
|
"mute": mute_command, "downtime": downtime_command, "confirm": confirm_command,
|
|
"cancel": cancel_command}
|
|
|
|
|
|
class CommandHandler:
|
|
def __init__(self, deps: CommandDeps, allowlist: Callable[[str], str | None],
|
|
audit: Callable[[AuditEvent], object], registry: Registry | None = None,
|
|
limiter: FailureLimiter | None = None, bot_name: Callable[[str], str | None] | None = None,
|
|
rate_limited: frozenset[str] = RATE_LIMITED_COMMANDS):
|
|
self._deps = deps
|
|
self._allowlist = allowlist
|
|
self._audit = audit
|
|
self.registry = registry if registry is not None else default_registry()
|
|
self._limiter = limiter or FailureLimiter(RATE_LIMIT_MAX, RATE_LIMIT_WINDOW)
|
|
self._bot_name = bot_name
|
|
self._rate_limited = rate_limited
|
|
|
|
async def _record(self, message: IncomingMessage, name: str, args: str, outcome: str) -> None:
|
|
event = AuditEvent(message.backend, message.sender_id, name, args[:AUDIT_ARGS_MAX], outcome)
|
|
try:
|
|
result = self._audit(event)
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
except Exception:
|
|
log.exception("command audit failed")
|
|
|
|
async def handle(self, message: IncomingMessage) -> Reply | None:
|
|
bot = self._bot_name(message.backend) if self._bot_name else None
|
|
command = parse(message.backend, message.text, bot)
|
|
if command is None:
|
|
return None
|
|
if not is_allowed(message.backend, message.sender_id, self._allowlist(message.backend)):
|
|
await self._record(message, command.name, command.rest, OUTCOME_DENIED)
|
|
return None
|
|
handler = self.registry.get(command.name)
|
|
if handler is None:
|
|
await self._record(message, command.name, command.rest, OUTCOME_UNKNOWN)
|
|
return plain(UNKNOWN_TEXT.format(p=PREFIXES[message.backend]))
|
|
if command.name in self._rate_limited:
|
|
key = f"{message.backend}:{message.sender_id}"
|
|
if self._limiter.blocked(key):
|
|
await self._record(message, command.name, command.rest, OUTCOME_RATE_LIMITED)
|
|
return plain(RATE_TEXT)
|
|
self._limiter.record(key)
|
|
try:
|
|
reply = await handler(CommandContext(message, command, self._deps, self.registry))
|
|
except Exception:
|
|
log.exception("command %s failed", command.name)
|
|
await self._record(message, command.name, command.rest, OUTCOME_ERROR)
|
|
return plain(ERROR_TEXT)
|
|
await self._record(message, command.name, command.rest, OUTCOME_OK)
|
|
return reply
|