215 lines
6.3 KiB
Python
215 lines
6.3 KiB
Python
"""Command registry and dispatcher."""
|
|
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import cast
|
|
|
|
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
|
|
admin: bool = False
|
|
|
|
|
|
# Registry maps command names to definitions
|
|
_registry: dict[str, CommandDefinition] = {}
|
|
|
|
|
|
def unregister(name: str) -> None:
|
|
"""Unregister a command and all its aliases.
|
|
|
|
Args:
|
|
name: The main command name to unregister
|
|
"""
|
|
# Find the definition by name
|
|
defn = _registry.get(name)
|
|
if defn is None:
|
|
return
|
|
|
|
# Remove main name
|
|
_registry.pop(name, None)
|
|
|
|
# Remove all aliases
|
|
for alias in defn.aliases:
|
|
_registry.pop(alias, None)
|
|
|
|
# Also remove any other keys pointing to this same definition
|
|
keys_to_remove = [k for k, v in _registry.items() if v is defn]
|
|
for key in keys_to_remove:
|
|
del _registry[key]
|
|
|
|
|
|
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 aliases (with recursion guard)
|
|
expansion_count = 0
|
|
max_expansions = 10
|
|
while command in player.aliases:
|
|
if expansion_count >= max_expansions:
|
|
player.writer.write("Too many nested aliases (max 10).\r\n")
|
|
await player.writer.drain()
|
|
return
|
|
expansion = player.aliases[command]
|
|
# Combine expansion with remaining args
|
|
raw_input = f"{expansion} {args}" if args else expansion
|
|
# Re-split to get new command and args
|
|
parts = raw_input.split(maxsplit=1)
|
|
command = parts[0].lower()
|
|
args = parts[1] if len(parts) > 1 else ""
|
|
expansion_count += 1
|
|
|
|
# Resolve command by exact match or prefix
|
|
result = resolve_prefix(command)
|
|
|
|
if result is None:
|
|
# No matches - try verb dispatch as fallback
|
|
if args:
|
|
from mudlib.verbs import find_object
|
|
|
|
# Split args into target name and any extra arguments
|
|
target_parts = args.split(maxsplit=1)
|
|
target_name = target_parts[0]
|
|
extra_args = target_parts[1] if len(target_parts) > 1 else ""
|
|
|
|
# Try to find the object
|
|
obj = find_object(target_name, player)
|
|
if obj is not None and obj.has_verb(command):
|
|
handler = obj.get_verb(command)
|
|
assert handler is not None # has_verb checked above
|
|
await handler(player, extra_args)
|
|
return
|
|
|
|
# Still no match - show unknown command
|
|
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:
|
|
prefix = ", ".join(cast(list[str], names[:-1]))
|
|
msg = f"{prefix}, 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
|
|
|
|
# Check admin permission
|
|
if defn.admin and not getattr(player, "is_admin", False):
|
|
player.writer.write("You don't have permission to do that.\r\n")
|
|
await player.writer.drain()
|
|
return
|
|
|
|
# Execute the handler
|
|
await defn.handler(player, args)
|