houdini/Houdini/Commands.py

import inspect
import config


from Houdini import Handlers
from Houdini import Plugins
from Houdini import ConflictResolution

from Houdini.Converters import _ArgumentDeserializer, _listener


class UnknownCommandException(Exception):
    """Raised when a command is executed that doesn't exist"""


class _Command(_ArgumentDeserializer):

    def __init__(self, name, callback, **kwargs):
        super().__init__(name, callback, **kwargs)

        self.alias = kwargs.get('alias', [])
        self.parent = kwargs.get('parent', None)


class _CommandGroup(_Command):
    __slots__ = ['commands']

    def __init__(self, name, callback, **kwargs):
        super().__init__(name, callback, **kwargs)

        self.commands = {}

    async def __call__(self, p, data):
        if not data:
            if self.instance:
                return await self.callback(self.instance, p)
            return await self.callback(p)

        await invoke_command_objects(self.commands, p, data)

    def command(self, name=None, **kwargs):
        return command(name, parent=self, **kwargs)

    def group(self, name=None, **kwargs):
        return group(name, parent=self, **kwargs)


def command(name=None, **kwargs):
    return _listener(_Command, name, string_delimiter=config.commands['StringDelimiters'],
                     string_separator=' ', **kwargs)


def group(name=None, **kwargs):
    return _listener(_CommandGroup, name, string_delimiter=config.commands['StringDelimiters'],
                     string_separator=' ', **kwargs)


cooldown = Handlers.cooldown
check = Handlers.check

player_attribute = Handlers.player_attribute
player_data_attribute = Handlers.player_data_attribute
player_in_room = Handlers.player_in_room


def is_command(command_object):
    return issubclass(type(command_object), _Command)


def commands_from_plugin(commands, plugin):
    command_objects = inspect.getmembers(plugin, is_command)
    if not isinstance(plugin, Plugins.IPlugin):
        raise TypeError('Commands can only be loaded from plugins')

    for command_name, command_object in command_objects:
        command_object.instance = plugin

        if type(command_object.alias) == str:
            command_object.alias = [command_object.alias]
        command_object.alias.append(command_object.name)

        parent_commands = commands if command_object.parent is None else command_object.parent.commands

        for name in command_object.alias:
            if name in parent_commands:
                conflict_command = parent_commands[name][0]
                if config.commands['ConflictMode'] == ConflictResolution.Exception:
                    raise NameError('Command name conflict: \'{}\' from plugin \'{}\' '
                                    'conflicts with \'{}\' from module \'{}\''
                                    .format(name, plugin.__class__.__name__, conflict_command.name,
                                            conflict_command.plugin.__class__.__name__))
                elif config.commands['ConflictMode'] == ConflictResolution.Append:
                    parent_commands[name].append(command_object)
                elif config.commands['ConflictMode'] == ConflictResolution.Silent:
                    plugin.server.logger.warn('Command \'{}\' from plugin \'{}\' disabled due to conflict with \'{}\''
                                              .format(name, plugin.__class__.__name__,
                                                      conflict_command.plugin.__class__.__name__))
            else:
                parent_commands[name] = [command_object]


if type(config.commands['Prefix']) == str:
    config.commands['Prefix'] = [config.commands['Prefix']]


def has_command_prefix(command_string):
    for prefix in config.commands['Prefix']:
        if command_string.startswith(prefix):
            return True
    return False


def get_command_prefix(command_string):
    for prefix in config.commands['Prefix']:
        if command_string.startswith(prefix):
            return prefix


async def invoke_command_string(commands, p, command_string):
    prefix = get_command_prefix(command_string)
    no_prefix = command_string[len(prefix):]
    data = no_prefix.split(' ')

    await invoke_command_objects(commands, p, data)


async def invoke_command_objects(commands, p, data):
    command_identifier = data.pop(0)
    if command_identifier not in commands:
        raise UnknownCommandException('Command \'{}\' does not exist'.format(command_identifier))

    command_objects = commands[command_identifier]
    for command_object in command_objects:
        await command_object(p, data)