148 lines
4.1 KiB
Python
148 lines
4.1 KiB
Python
"""Command registry and dispatcher."""
|
|
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
|
|
from mudlib.player import Player
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Type alias for command handlers
|
|
CommandHandler = Callable[[Player, str], Awaitable[None]]
|
|
|
|
|
|
@dataclass
|
|
class CommandDefinition:
|
|
"""Metadata wrapper for a registered command."""
|
|
|
|
name: str
|
|
handler: CommandHandler
|
|
aliases: list[str] = field(default_factory=list)
|
|
mode: str = "normal"
|
|
help: str = ""
|
|
hidden: bool = False
|
|
|
|
|
|
# Registry maps command names to definitions
|
|
_registry: dict[str, CommandDefinition] = {}
|
|
|
|
|
|
def register(defn: CommandDefinition) -> None:
|
|
"""Register a command definition with its aliases.
|
|
|
|
Args:
|
|
defn: The command definition to register
|
|
"""
|
|
# Check for collision on the main name
|
|
if defn.name in _registry:
|
|
existing = _registry[defn.name]
|
|
if existing is not defn:
|
|
logger.warning(
|
|
"Command collision: '%s' already registered for '%s', "
|
|
"overwriting with '%s'",
|
|
defn.name,
|
|
existing.name,
|
|
defn.name,
|
|
)
|
|
|
|
_registry[defn.name] = defn
|
|
|
|
# Check for collisions on each alias
|
|
for alias in defn.aliases:
|
|
if alias in _registry:
|
|
existing = _registry[alias]
|
|
if existing is not defn:
|
|
logger.warning(
|
|
"Command collision: '%s' already registered for '%s', "
|
|
"overwriting with '%s'",
|
|
alias,
|
|
existing.name,
|
|
defn.name,
|
|
)
|
|
_registry[alias] = defn
|
|
|
|
|
|
def resolve_prefix(command: str) -> CommandDefinition | list[str] | None:
|
|
"""Resolve a command by exact match or shortest unique prefix.
|
|
|
|
Args:
|
|
command: Command name or prefix to resolve
|
|
|
|
Returns:
|
|
CommandDefinition if unique match found
|
|
list of command names if ambiguous
|
|
None if no match
|
|
"""
|
|
# Try exact match first (fast path)
|
|
defn = _registry.get(command)
|
|
if defn is not None:
|
|
return defn
|
|
|
|
# Try prefix matching
|
|
prefix_matches = [key for key in _registry if key.startswith(command)]
|
|
|
|
# Deduplicate by CommandDefinition identity
|
|
unique_defns = {}
|
|
for key in prefix_matches:
|
|
defn_obj = _registry[key]
|
|
unique_defns[id(defn_obj)] = defn_obj
|
|
|
|
if len(unique_defns) == 0:
|
|
# No matches
|
|
return None
|
|
elif len(unique_defns) == 1:
|
|
# Unique match
|
|
return next(iter(unique_defns.values()))
|
|
else:
|
|
# Multiple matches - return list of names
|
|
return sorted([d.name for d in unique_defns.values()])
|
|
|
|
|
|
async def dispatch(player: Player, raw_input: str) -> None:
|
|
"""Parse input, find command, call handler.
|
|
|
|
Args:
|
|
player: The player executing the command
|
|
raw_input: The raw input string from the player
|
|
"""
|
|
raw_input = raw_input.strip()
|
|
|
|
if not raw_input:
|
|
return
|
|
|
|
# Split into command and arguments
|
|
parts = raw_input.split(maxsplit=1)
|
|
command = parts[0].lower()
|
|
args = parts[1] if len(parts) > 1 else ""
|
|
|
|
# Resolve command by exact match or prefix
|
|
result = resolve_prefix(command)
|
|
|
|
if result is None:
|
|
# No matches
|
|
player.writer.write(f"Unknown command: {command}\r\n")
|
|
await player.writer.drain()
|
|
return
|
|
elif isinstance(result, list):
|
|
# Multiple matches - show disambiguation
|
|
names = result
|
|
if len(names) == 2:
|
|
msg = f"{names[0]} or {names[1]}?\r\n"
|
|
else:
|
|
msg = f"{', '.join(names[:-1])}, or {names[-1]}?\r\n"
|
|
player.writer.write(msg)
|
|
await player.writer.drain()
|
|
return
|
|
else:
|
|
# Unique match
|
|
defn = result
|
|
|
|
# Check mode restriction
|
|
if defn.mode != "*" and defn.mode != player.mode:
|
|
player.writer.write("You can't do that right now.\r\n")
|
|
await player.writer.drain()
|
|
return
|
|
|
|
# Execute the handler
|
|
await defn.handler(player, args)
|