| 1 | """Minimal command routing table for slash commands.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from dataclasses import dataclass |
| 6 | from typing import TYPE_CHECKING, Any, Awaitable, Callable |
| 7 | |
| 8 | if TYPE_CHECKING: |
| 9 | from nanobot.bus.events import InboundMessage, OutboundMessage |
| 10 | from nanobot.session.manager import Session |
| 11 | |
| 12 | Handler = Callable[["CommandContext"], Awaitable["OutboundMessage | None"]] |
| 13 | |
| 14 | |
| 15 | @dataclass |
| 16 | class CommandContext: |
| 17 | """Everything a command handler needs to produce a response.""" |
| 18 | |
| 19 | msg: InboundMessage |
| 20 | session: Session | None |
| 21 | key: str |
| 22 | raw: str |
| 23 | args: str = "" |
| 24 | loop: Any = None |
| 25 | |
| 26 | |
| 27 | class CommandRouter: |
| 28 | """Pure dict-based command dispatch. |
| 29 | |
| 30 | Three tiers checked in order: |
| 31 | 1. *priority* — exact-match commands handled before the dispatch lock |
| 32 | (e.g. /stop, /restart). |
| 33 | 2. *exact* — exact-match commands handled inside the dispatch lock. |
| 34 | 3. *prefix* — longest-prefix-first match (e.g. "/team "). |
| 35 | 4. *interceptors* — fallback predicates (e.g. team-mode active check). |
| 36 | """ |
| 37 | |
| 38 | def __init__(self) -> None: |
| 39 | self._priority: dict[str, Handler] = {} |
| 40 | self._exact: dict[str, Handler] = {} |
| 41 | self._prefix: list[tuple[str, Handler]] = [] |
| 42 | self._interceptors: list[Handler] = [] |
| 43 | |
| 44 | def priority(self, cmd: str, handler: Handler) -> None: |
| 45 | self._priority[cmd] = handler |
| 46 | |
| 47 | def exact(self, cmd: str, handler: Handler) -> None: |
| 48 | self._exact[cmd] = handler |
| 49 | |
| 50 | def prefix(self, pfx: str, handler: Handler) -> None: |
| 51 | self._prefix.append((pfx, handler)) |
| 52 | self._prefix.sort(key=lambda p: len(p[0]), reverse=True) |
| 53 | |
| 54 | def intercept(self, handler: Handler) -> None: |
| 55 | self._interceptors.append(handler) |
| 56 | |
| 57 | def is_priority(self, text: str) -> bool: |
| 58 | return text.strip().lower() in self._priority |
| 59 | |
| 60 | def is_dispatchable_command(self, text: str) -> bool: |
| 61 | """Check whether *text* matches any non-priority command tier (exact or prefix). |
| 62 | |
| 63 | Does NOT check priority or interceptor tiers. |
| 64 | If this returns True, ``dispatch()`` is guaranteed to match a handler. |
| 65 | """ |
| 66 | cmd = text.strip().lower() |
| 67 | if cmd in self._exact: |
| 68 | return True |
| 69 | for pfx, _ in self._prefix: |
| 70 | if cmd.startswith(pfx): |
| 71 | return True |
| 72 | return False |
| 73 | |
| 74 | async def dispatch_priority(self, ctx: CommandContext) -> OutboundMessage | None: |
| 75 | """Dispatch a priority command. Called from run() without the lock.""" |
| 76 | handler = self._priority.get(ctx.raw.lower()) |
| 77 | if handler: |
| 78 | return await handler(ctx) |
| 79 | return None |
| 80 | |
| 81 | async def dispatch(self, ctx: CommandContext) -> OutboundMessage | None: |
| 82 | """Try exact, prefix, then interceptors. Returns None if unhandled.""" |
| 83 | cmd = ctx.raw.lower() |
| 84 | |
| 85 | if handler := self._exact.get(cmd): |
| 86 | return await handler(ctx) |
| 87 | |
| 88 | for pfx, handler in self._prefix: |
| 89 | if cmd.startswith(pfx): |
| 90 | ctx.args = ctx.raw[len(pfx):] |
| 91 | return await handler(ctx) |
| 92 | |
| 93 | for interceptor in self._interceptors: |
| 94 | result = await interceptor(ctx) |
| 95 | if result is not None: |
| 96 | return result |
| 97 | |
| 98 | return None |
| 99 |