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)