Source code for commands.typed

# ../commands/

#: .. todo:: Add the ability to define prefixes for messages.
#: .. todo:: Add callback to the Node class. It could be called when a sub-command is required.

# =============================================================================
# =============================================================================
# Python Imports
#   Inspect
import inspect
from inspect import Parameter
import itertools
import textwrap

# Source.Python
#   Auth
from auth.manager import auth_manager
#   Core
from core import AutoUnload
from core import Tokenize
#   Commands
from commands import commands_logger
from commands import CommandReturn
from commands.server import server_command_manager
from commands.client import client_command_manager
from commands.say import say_command_manager
#   Filters
from filters.players import parse_filter
#   Messages
from messages import SayText2
from messages import TextMsg
from messages import HudDestination
#   Translations
from translations.strings import TranslationStrings

# =============================================================================
# =============================================================================

# =============================================================================
# =============================================================================
    # There is no syntax for this in Python, but a PEP. If it ever gets added,
    # this implementation will already support it (unless
    # <Parameter object>.name is empty).


# =============================================================================
# =============================================================================
logger = commands_logger.typed

# =============================================================================
# =============================================================================
[docs]class ValidationError(Exception):
[docs] def __init__(self, message='', language=None, **tokens): self.message = message self.language = language self.tokens = tokens
[docs]class ArgumentError(ValidationError): pass
[docs]class ArgumentNumberMismatch(ArgumentError): pass
[docs]class InvalidArgumentValue(ArgumentError): pass
[docs]class SubCommandError(ValidationError): pass
[docs]class SubCommandNotFound(SubCommandError): pass
[docs]class SubCommandExpectedError(SubCommandError): pass
# ============================================================================= # >> CLASSES # =============================================================================
[docs]class Node(object):
[docs] def __init__(self, commands, description): self.description = description self.commands = commands
@property def signature(self): """Return the signature of the node. :rtype: str """ raise NotImplementedError('Must be implemented by a sub class.')
[docs]class Store(Node, dict):
[docs] def __init__(self, commands=None, description=None): super().__init__(commands, description)
@property def signature(self): """Return the signature of the node. :rtype: str """ return ' '.join(self.commands) + ' <sub-command>' @property def help_text(self): """Return the help text of the node. :rtype: str """ wrapper = textwrap.TextWrapper( 40, subsequent_indent=' ', break_long_words=True) result = '' for node in sorted(self.values(), key=lambda node: node.signature): sig_lines = wrapper.wrap(node.signature) desc_lines = wrapper.wrap(node.description or '') for sig, desc in itertools.zip_longest( sig_lines, desc_lines, fillvalue=''): result += '\n ' + sig.ljust(40) + ' ' + desc return result
[docs]class CommandNode(Node):
[docs] def __init__(self, commands, params, callback, description, permission, fail_callback, requires_registration): """Initialize the object. .. seealso:: :meth:`CommandParser.add_command` """ super().__init__(commands, description) self.params = params self.callback = callback # Explicitly check against True, because <permission> can also be a # non-empty string if permission == True: permission = '.'.join(commands) self.permission = permission self.fail_callback = fail_callback self.requires_registration = requires_registration self.command_to_register = commands[0]
@property def signature(self): """Return the full signature of a command. :rtype: str """ result = ' '.join(self.commands) for param in self.params: result += ' ' + self._param_to_str(param) return result @staticmethod def _param_to_str(param): """Represent a parameter as a string. :rtype: str """ result = '{open_char}{arg_name}{type_name}{default}{close_char}' if (param.kind == param.VAR_POSITIONAL or param.default is not param.empty): open_char = '[' close_char = ']' else: open_char = '<' close_char = '>' if param.default is not param.empty: default = '=' + str(param.default) else: default = '' if param.annotation is not param.empty: type_name = ':' + param.annotation.__name__ else: type_name = '' if param.kind == param.VAR_POSITIONAL: arg_name = '*' + else: arg_name = return result.format(open_char=open_char, arg_name=arg_name, type_name=type_name, default=default, close_char=close_char)
# TODO: This base class is not quite correct.
[docs]class CommandParser(Store):
[docs] def add_command(self, commands, params, callback, description=None, permission=None, fail_callback=None): """Add a command to the parser. :param str/list/tuple commands: Command to register. :param iterable params: Parameters of the command. :param callable callback: The callback for the command. :param str description: Description of the command. :param str permission: Required permission to use the command. :param callable fail_callback: Callback that gets called if authorization failed. :rtype: CommandNode """ commands = self._validate_commands(commands) for param in params: if param.kind not in SUPPORTED_KINDS: raise TypeError( 'Unsupported argument type "{}" for argument.'.format( param.kind, command = CommandNode(tuple(commands), tuple(params), callback, description, permission, fail_callback, commands[0] not in self) parsed_commands = [] store = self while commands: command_name = commands.pop(0).lower() parsed_commands.append(command_name) if command_name in store: store = store[command_name] if (isinstance(store, CommandNode) or (not commands and isinstance(store, Store))): # We can't support multiple callbacks, because they might # have different signatures. raise ValueError('Command already exists.') else: if commands: # We need to split these two lines to prevent recursive # dicts new_store = store[command_name] = Store( tuple(parsed_commands)) store = new_store else: store[command_name] = command return command
[docs] def remove_command(self, commands): """Remove a command. :param str/list/tuple: Command to remove. .. seealso:: :meth:`_remove_command` .. seealso:: :meth:`_validate_commands` """ return self._remove_command(self._validate_commands(commands))
def _remove_command(self, commands): """Remove a command. :param list commands: Command to remove. :raise ValueError: Raised if the node does not exist. :return: Return whether the base command needs to be unregistered. :rtype: bool """ store = self for command_name in commands[:-1]: try: store = store[command_name] except KeyError: raise ValueError('Command does not exist.') del store[commands[-1]] if len(commands) > 1: if store: return False return self._remove_command(commands[:-1]) return True
[docs] def get_node(self, commands): """Return a node. :param str/list/tuple: Node to seach. :raise ValueError: Raised if the node does not exist. :rtype: Node .. seealso:: :meth:`_validate_command` """ store = self for command_name in self._validate_commands(commands): try: store = store[command_name] except (TypeError, KeyError): # The TypeError happens if "store" is already a CommandNode raise ValueError('Node does not exist.') return store
[docs] def set_node_description(self, commands, description): """Set the description of a node. :param str/list/tuple: Node to seach. :raise ValueError: Raised if the node does not exist. """ self.get_node(commands).description = description
[docs] def get_command(self, commands): """Return a command. :param str/list/tuple: Command to search. :rtype: CommandNode .. seealso:: :meth:`get_node` """ command = self.get_node(commands) assert isinstance(command, CommandNode) return command
def _validate_commands(self, commands): """Validate a string, list or tuple of commands. :param str/list/tuple: The command name or command path to validate. :raise TypeError: Raised if ``commands`` is not a str, list or tuple. :raise ValueError: Raised if ``commands`` is an empty list or tuple. .. seealso:: :meth:`_validate_command` """ if isinstance(commands, str): self._validate_command(commands) commands = [commands] elif not isinstance(commands, (list, tuple)): raise TypeError('<commands> must be str, list or tuple.') elif not commands: raise ValueError('<commands> must not be empty.') else: for command_name in commands: self._validate_command(command_name) return list(commands) def _validate_command(self, command_name): """Validate a command name. :param str command_name: The command name to check. :raise ValueError: Raised if the command name is invalid. """ if not command_name: raise ValueError('Command name cannot be empty.') if ' ' in command_name: raise ValueError('Command cannot contain spaces.')
[docs] def clean_command(self, command, args): """Clean a command and its passed arguments. :param CommandNode command: :param iterable args: The arguments to clean. :rtype: list :raise InvalidArgumentValue: Raised if the value is invalid for an argument. :raise ArgumentNumberMismatch: Raised if too many/less arguments have been passed. """ result = [] params = list(command.params) param = None for arg in args: if param is None or param.kind is not param.VAR_POSITIONAL: try: param = params.pop(0) except IndexError: raise ArgumentNumberMismatch( 'Too many arguments:\n {}'.format(command.signature)) if param.annotation is not param.empty: try: arg = param.annotation(arg) except ValidationError: raise except: raise InvalidArgumentValue( '"{}" is an invalid value for "{}:{}".'.format( arg,, param.annotation.__name__)) result.append(arg) for param in params: if param.kind is param.VAR_POSITIONAL: break if param.default is param.empty: raise ArgumentNumberMismatch( 'Not enough arguments:\n {}'.format(command.signature)) result.append(param.default) return result
[docs] def parse_command(self, command): """Parse a :class:`Command` object. Splits the command into the actual command and its arguments. :param Command command: A command to parse. :raise SubCommandNotFound: Raised if a sub command was not found. :raise SubCommandExpected: Raised if a sub command was expected, but more arguments have been passed. :rtype: tuple """ args = Tokenize(command.command_string) store = self while args and isinstance(store, Store): sub_command = args.pop(0).lower() try: store = store[sub_command] except KeyError: raise SubCommandNotFound( 'Sub command "{}" not found.'.format(sub_command)) if isinstance(store, Store): raise SubCommandExpectedError( 'A sub-command is required:{}'.format(store.help_text)) return (store, args)
[docs]class CommandInfo(object): """Stores command information for typed commands."""
[docs] def __init__(self, command, typed_command_cls, index=None, team_only=None): """Initializes the instance. :param Command command: The actual Command instance. :param _TypedCommand typed_command_cls: Command this instance belongs to. :param int index: The index of the player that issued the command. None, if it's a server command. :param bool team_only: Indicates whether the command was issued in team chat. None, if it's a server or client command. """ self.command = command self.typed_command_cls = typed_command_cls self.index = index self.team_only = team_only
[docs] def reply(self, msg, language=None, **tokens): """Reply to the command issuer. :param str/TranslationStrings msg: Message to send. :param str language: Language to be used. :param tokens: Translation tokens for message. """ self.typed_command_cls.send_message(self, msg, language, **tokens)
[docs] def is_private_command(self): """Return ``True`` if it's a private command. :rtype: bool """ return self.command[0].startswith(PRIVATE_COMMAND_PREFIX)
@property def auto_command_return(self): """Determine the probably most desired ``CommandReturn`` value. For server commands it will always return ``CommandReturn.CONTINUE``. Client commands will always use ``CommandReturn.BLOCK``. For say commands it's determined by checking the prefix. If the command starts with a slash (``/``) it's handled as a private command. Thus, ``CommandReturn.BLOCK`` is returned. Otherwise ``CommandReturn.CONTINUE`` is used. :rtype: CommandReturn """ return self.typed_command_cls.get_auto_command_return(self)
# We can't integrate this into SayCommand, ServerCommand and ClientCommand, # because multiple callbacks are not supported by this system (because of the # possibility of different function signatures). But multiple callbacks are # important e.g. for ClientCommand, because that can actually hook client # commands and multiple plugins might want to hook a client command. class _TypedCommand(AutoUnload): """Decorator class to create typed commands.""" def __init__(self, commands, permission=None, fail_callback=None): """Register a typed command callback. :param str/list/tuple commands: (Sub-) command to register. :param str/bool permission: A permission that is required to execute the command. If True, the permission string will be generated from the given command. :param callable fail_callback: Called when the executer does not have the required permission. """ self.commands = commands self.permission = permission self.fail_callback = fail_callback self.command = None def __call__(self, callback): """Finish registering a typed command callback. :param callable callback: A callback that get called when the command has been issued. :raise ValueError: Raised if the callback does not accept at least one argument (command info). :return: Return the passed callback. :rtype: callable """ params = tuple(inspect.signature(callback).parameters.values()) if not params: raise ValueError( 'Callback must at least accept 1 argument (command_info).') self.command = self.parser.add_command( self.commands, params[1:], callback, inspect.getdoc(callback), self.permission, self.fail_callback) if self.command.requires_registration: self.manager.register_commands( self.command.command_to_register, self.on_command) return callback def _unload_instance(self): try: if self.parser.remove_command(self.commands): self.manager.unregister_commands( self.command.command_to_register, self.on_command) except ValueError: pass @classmethod def on_command(cls, command, *args): """Called when a (base) command has been executed. Parse the command, clean its arguments and execute the callback. """ info = CommandInfo(command, cls, *args) try: command_node, args = cls.parser.parse_command(info.command) result = cls.on_clean_command(info, command_node, args) except ValidationError as e: info.reply(e.message, e.language, **e.tokens) else: if result is None: return info.auto_command_return return result return CommandReturn.CONTINUE @classmethod def on_clean_command(cls, command_info, command_node, args): """Called when the arguments of the parsed command should be cleaned. :rtype: CommandReturn """ cleaned_args = cls.parser.clean_command(command_node, args) return command_node.callback(command_info, *cleaned_args) @property def parser(self): """Return the parser to use. :rtype: CommandParser """ raise NotImplementedError('Needs to be implemented by a sub class.') @property def manager(self): """Return the manager that registers the commands. :rtype: _BaseCommandManager """ raise NotImplementedError('Needs to be implemented by a sub class.') @staticmethod def send_message(command_info, message, language=None, **tokens): """Send a message.""" raise NotImplementedError('Needs to be implemented by a sub class.') @classmethod def get_auto_command_return(cls, info): """Return the most desired ``CommandReturn`` value. :rtype: CommandReturn """ raise NotImplementedError('Needs to be implemented by a sub class.')
[docs]class TypedServerCommand(_TypedCommand): """Decorator class to create typed server commands.""" parser = CommandParser() manager = server_command_manager @staticmethod
[docs] def send_message(command_info, message, language=None, **tokens): # Translate the message if it's a :class:`TranslationStrings` object. if isinstance(message, TranslationStrings): message = message.get_string(language, **tokens) logger.log_message(message)
[docs] def get_auto_command_return(cls, info): return CommandReturn.CONTINUE
class _TypedPlayerCommand(_TypedCommand): """Decorator class to create typed player based commands.""" @classmethod def on_clean_command(cls, command_info, command_node, args): if (command_node.permission is None or auth_manager.is_player_authorized( command_info.index, command_node.permission)): return super().on_clean_command(command_info, command_node, args) return cls.handle_fail_callback(command_info, command_node, args) @classmethod def handle_fail_callback(cls, command_info, command_node, args): if command_node.fail_callback is not None: return command_node.fail_callback(command_info, args) # TODO: Send "Required permission: <permission>" or write it to the # logs? cls.send_message( command_info, 'You are not authorized to use this command.\n' + 'Required permission: {}'.format(command_node.permission)) return CommandReturn.CONTINUE
[docs]class TypedClientCommand(_TypedPlayerCommand): """Decorator class to create typed client commands.""" parser = CommandParser() manager = client_command_manager @staticmethod
[docs] def send_message(command_info, message, language=None, **tokens): TextMsg(message, HudDestination.CONSOLE).send(command_info.index, **tokens)
[docs] def get_auto_command_return(cls, info): return CommandReturn.BLOCK
[docs]class TypedSayCommand(_TypedPlayerCommand): """Decorator class to create typed say commands.""" parser = CommandParser() manager = say_command_manager @staticmethod
[docs] def send_message(command_info, message, language=None, **tokens): SayText2(message).send(command_info.index, **tokens)
[docs] def get_auto_command_return(cls, info): if info.is_private_command(): return CommandReturn.BLOCK return CommandReturn.CONTINUE
# ============================================================================= # >> COMMAND ANNOTATIONS # =============================================================================
[docs]def filter_str(expr): """.. seealso:: :func:`filters.players.parse_filter`""" # A simple wrapper for parse_filter for a better looking command signature return parse_filter(expr)