PyLink/utils.py

"""
utils.py - PyLink utilities module.

This module contains various utility functions related to IRC and/or the PyLink
framework.
"""

import string
import re
import importlib
import os
import collections
import argparse

from .log import log
from . import world, conf

# Load the protocol and plugin packages.
from pylinkirc import protocols, plugins

PLUGIN_PREFIX = plugins.__name__ + '.'
PROTOCOL_PREFIX = protocols.__name__ + '.'
NORMALIZEWHITESPACE_RE = re.compile(r'\s+')
_proto_utils_class = None  # Set by classes.py when loaded

class NotAuthorizedError(Exception):
    """
    Exception raised by the PyLink permissions system when a user fails access requirements.
    """
    pass

class InvalidArgumentsError(TypeError):
    """
    Exception raised (by IRCParser and potentially others) when a bot command is given invalid arguments.
    """

class ProtocolError(RuntimeError):
    """
    Exception raised when a network protocol violation is encountered in some way.
    """

def add_cmd(func, name=None, **kwargs):
    """Binds an IRC command function to the given command name."""
    world.services['pylink'].add_cmd(func, name=name, **kwargs)
    return func

def add_hook(func, command):
    """Binds a hook function to the given command name."""
    command = command.upper()
    world.hooks[command].append(func)
    return func

# DEPRECATED
def isNick(s, nicklen=None):
    """Returns whether the string given is a valid nick.

    Deprecated since 2.0: use irc.is_nick() instead."""

    log.warning('utils.isNick() is deprecated since PyLink 2.0, use irc.is_nick() instead.')
    return _proto_utils_class.is_nick(s, nicklen=nicklen)

# DEPRECATED
def isChannel(s):
    """Returns whether the string given is a valid channel name.

    Deprecated since 2.0: use irc.is_channel() instead."""

    log.warning('utils.isChannel() is deprecated since PyLink 2.0, use irc.is_channel() instead.')
    return _proto_utils_class.is_channel(s)

# DEPRECATED
def isServerName(s):
    """Returns whether the string given is a valid server name.

    Deprecated since 2.0: use irc.is_server_name() instead."""

    log.warning('utils.isServerName() is deprecated since PyLink 2.0, use irc.is_server_name() instead.')
    return _proto_utils_class.is_server_name(s)

# DEPRECATED
def isHostmask(text):
    """Returns whether the given text is a valid hostmask.

    Deprecated since 2.0: use irc.is_hostmask() instead."""
    log.warning('utils.isHostmask() is deprecated since PyLink 2.0, use irc.is_hostmask() instead.')
    return _proto_utils_class.is_hostmask(text)

def expand_path(path):
    """
    Returns a path expanded with environment variables and home folders (~) expanded, in that order."""
    return os.path.expanduser(os.path.expandvars(path))
expandpath = expand_path  # Consistency with os.path

def reset_module_dirs():
    """
    (Re)sets custom protocol module and plugin directories to the ones specified in the config.
    """
    # Note: This assumes that the first element of the package path is the default one.
    plugins.__path__ = [plugins.__path__[0]] + [expandpath(path) for path in conf.conf['pylink'].get('plugin_dirs', [])]
    log.debug('reset_module_dirs: new pylinkirc.plugins.__path__: %s', plugins.__path__)
    protocols.__path__ = [protocols.__path__[0]] + [expandpath(path) for path in conf.conf['pylink'].get('protocol_dirs', [])]
    log.debug('reset_module_dirs: new pylinkirc.protocols.__path__: %s', protocols.__path__)
resetModuleDirs = reset_module_dirs

def load_plugin(name):
    """
    Imports and returns the requested plugin.
    """
    return importlib.import_module(PLUGIN_PREFIX + name)
loadPlugin = load_plugin

def get_protocol_module(name):
    """
    Imports and returns the protocol module requested.
    """
    return importlib.import_module(PROTOCOL_PREFIX + name)
getProtocolModule = get_protocol_module

def split_hostmask(mask):
    """
    Returns a nick!user@host hostmask split into three fields: nick, user, and host.
    """
    nick, identhost = mask.split('!', 1)
    ident, host = identhost.split('@', 1)
    return [nick, ident, host]
splitHostmask = split_hostmask

class ServiceBot():
    """
    PyLink IRC Service class.
    """

    def __init__(self, name, default_help=True, default_list=True, manipulatable=False, default_nick=None, desc=None):
        # Service name and default nick
        self.name = name
        self.default_nick = default_nick

        # Tracks whether the bot should be manipulatable by the 'bots' plugin and other commands.
        self.manipulatable = manipulatable

        # We make the command definitions a dict of lists of functions. Multiple
        # plugins are actually allowed to bind to one function name; this just causes
        # them to be called in the order that they are bound.
        self.commands = collections.defaultdict(list)

        # This tracks the UIDs of the service bot on different networks, as they are
        # spawned.
        self.uids = {}

        # Track what channels other than those defined in the config
        # that the bot should join by default.
        self.extra_channels = collections.defaultdict(set)

        # Service description, used in the default help command if one is given.
        self.desc = desc

        # List of command names to "feature"
        self.featured_cmds = set()

        # Maps command aliases to the respective primary commands
        self.alias_cmds = {}

        if default_help:
            self.add_cmd(self.help)

        if default_list:
            self.add_cmd(self.listcommands, 'list')

    def spawn(self, irc=None):
        """
        Spawns instances of this service on all connected networks.
        """
        # Spawn the new service by calling the PYLINK_NEW_SERVICE hook,
        # which is handled by coreplugin.
        if irc is None:
            for irc in world.networkobjects.values():
                irc.call_hooks([None, 'PYLINK_NEW_SERVICE', {'name': self.name}])
        else:
            raise NotImplementedError("Network specific plugins not supported yet.")

    def join(self, irc, channels, autojoin=True):
        """
        Joins the given service bot to the given channel(s).
        """

        if isinstance(irc, str):
            netname = irc
        else:
            netname = irc.name

        # Ensure type safety: pluralize strings if only one channel was given, then convert to set.
        if isinstance(channels, str):
            channels = [channels]
        channels = set(channels)

        if autojoin:
            log.debug('(%s/%s) Adding channels %s to autojoin', netname, self.name, channels)
            self.extra_channels[netname] |= channels

        # If the network was given as a string, look up the Irc object here.
        try:
            irc = world.networkobjects[netname]
        except KeyError:
            log.debug('(%s/%s) Skipping join(), IRC object not initialized yet', netname, self.name)
            return

        try:
            u = self.uids[irc.name]
        except KeyError:
            log.debug('(%s/%s) Skipping join(), UID not initialized yet', irc.name, self.name)
            return

        # Specify modes to join the services bot with.
        joinmodes = irc.serverdata.get("%s_joinmodes" % self.name) or conf.conf.get(self.name, {}).get('joinmodes') or ''
        joinmodes = ''.join([m for m in joinmodes if m in irc.prefixmodes])

        for chan in channels:
            if isChannel(chan):
                if chan in irc.channels and u in irc.channels[chan].users:
                    log.debug('(%s) Skipping join of services %s to channel %s - it is already present', irc.name, self.name, chan)
                    continue
                log.debug('(%s) Joining services %s to channel %s with modes %r', irc.name, self.name, chan, joinmodes)
                if joinmodes:  # Modes on join were specified; use SJOIN to burst our service
                    irc.proto.sjoin(irc.sid, chan, [(joinmodes, u)])
                else:
                    irc.proto.join(u, chan)

                irc.call_hooks([irc.sid, 'PYLINK_SERVICE_JOIN', {'channel': chan, 'users': [u]}])
            else:
                log.warning('(%s) Ignoring invalid autojoin channel %r.', irc.name, chan)

    def reply(self, irc, text, notice=None, private=None):
        """Replies to a message as the service in question."""
        servuid = self.uids.get(irc.name)
        if not servuid:
            log.warning("(%s) Possible desync? UID for service %s doesn't exist!", irc.name, self.name)
            return

        irc.reply(text, notice=notice, source=servuid, private=private)

    def error(self, irc, text, notice=None, private=None):
        """Replies with an error, as the service in question."""
        servuid = self.uids.get(irc.name)
        if not servuid:
            log.warning("(%s) Possible desync? UID for service %s doesn't exist!", irc.name, self.name)
            return

        irc.error(text, notice=notice, source=servuid, private=private)

    def call_cmd(self, irc, source, text, called_in=None):
        """
        Calls a PyLink bot command. source is the caller's UID, and text is the
        full, unparsed text of the message.
        """
        irc.called_in = called_in or source
        irc.called_by = source

        cmd_args = text.strip().split(' ')
        cmd = cmd_args[0].lower()
        cmd_args = cmd_args[1:]
        if cmd not in self.commands:
            # XXX: we really need abstraction for this kind of config fetching...
            show_unknown_cmds = irc.serverdata.get('%s_show_unknown_commands' % self.name,
                                                   conf.conf.get(self.name, {}).get('show_unknown_commands',
                                                   conf.conf['pylink'].get('show_unknown_commands', True)))

            if cmd and show_unknown_cmds and not cmd.startswith('\x01'):
                # Ignore empty commands and invalid command errors from CTCPs.
                self.reply(irc, 'Error: Unknown command %r.' % cmd)
            log.info('(%s/%s) Received unknown command %r from %s', irc.name, self.name, cmd, irc.get_hostmask(source))
            return

        log.info('(%s/%s) Calling command %r for %s', irc.name, self.name, cmd, irc.get_hostmask(source))
        for func in self.commands[cmd]:
            try:
                func(irc, source, cmd_args)
            except (NotAuthorizedError, InvalidArgumentsError) as e:
                self.reply(irc, 'Error: %s' % e)
            except Exception as e:
                log.exception('Unhandled exception caught in command %r', cmd)
                self.reply(irc, 'Uncaught exception in command %r: %s: %s' % (cmd, type(e).__name__, str(e)))

    def add_cmd(self, func, name=None, featured=False, aliases=None):
        """Binds an IRC command function to the given command name."""
        if name is None:
            name = func.__name__
        name = name.lower()

        # Mark as a featured command if requested to do so.
        if featured:
            self.featured_cmds.add(name)

        # If this is an alias, store the primary command in the alias_cmds dict
        if aliases is not None:
            for alias in aliases:
                if name == alias:
                    log.error('Refusing to alias command %r (in plugin %r) to itself!', name, func.__module__)
                    continue

                self.add_cmd(func, name=alias)  # Bind the alias as well.
                self.alias_cmds[alias] = name

        self.commands[name].append(func)
        return func

    def get_nick(self, irc, fails=0):
        """
        If the 'fails' argument is set to zero, this method returns the preferred nick for this
        service bot on the given network. The following fields are checked in the given order:
        # 1) Network specific nick settings for this service (servers:<netname>:servicename_nick)
        # 2) Global settings for this service (servicename:nick)
        # 3) The service's hardcoded default nick.
        # 4) The literal service name.

        If the 'fails' argument is set to a non-zero value, a list of *alternate* (fallback) nicks
        will be fetched from these fields in this order:
        # 1) Network specific altnick settings for this service (servers:<netname>:servicename_altnicks)
        # 2) Global altnick settings for this service (servicename:altnicks)

        If such an alternate nicks list exists, an alternate nick will be chosen based on the value
        of the 'fails' argument:
        - If nick fetching fails once, return the 1st alternate nick from the list,
        - If nick fetching fails twice, return the 2nd alternate nick from the list, ...

        Otherwise, if the alternate nicks list doesn't exist, or if there is no corresponding value
        for the current 'fails' value, the preferred nick plus the 'fails' number of underscores (_)
        will be used instead.
        - fails=1 => preferred_nick_
        - fails=2 => preferred_nick__

        If the resulting nick is too long for the given network, ProtocolError will be raised.
        """
        sbconf = conf.conf.get(self.name, {})
        nick = irc.serverdata.get("%s_nick" % self.name) or sbconf.get('nick') or self.default_nick or self.name

        if fails >= 1:
            altnicks = irc.serverdata.get("%s_altnicks" % self.name) or sbconf.get('altnicks') or []
            try:
                nick = altnicks[fails-1]
            except IndexError:
                nick += ('_' * fails)

        if irc.maxnicklen > 0 and len(nick) > irc.maxnicklen:
            raise ProtocolError("Nick %r too long for network (maxnicklen=%s)" % (nick, irc.maxnicklen))

        assert nick
        return nick

    def get_ident(self, irc):
        """
        Returns the preferred ident for this service bot on the given network. The following fields are checked in the given order:
        # 1) Network specific ident settings for this service (servers:<netname>:servicename_ident)
        # 2) Global settings for this service (servicename:ident)
        # 3) The service's hardcoded default nick.
        # 4) The literal service name.
        """
        sbconf = conf.conf.get(self.name, {})
        return irc.serverdata.get("%s_ident" % self.name) or sbconf.get('ident') or self.default_nick or self.name

    def get_host(self, irc):
        """
        Returns the preferred hostname for this service bot on the given network. The following fields are checked in the given order:
        # 1) Network specific hostname settings for this service (servers:<netname>:servicename_host)
        # 2) Global settings for this service (servicename:host)
        # 3) The PyLink server hostname.
        """
        sbconf = conf.conf.get(self.name, {})
        return irc.serverdata.get("%s_host" % self.name) or sbconf.get('host') or irc.hostname()

    def get_realname(self, irc):
        """
        Returns the preferred real name for this service bot on the given network. The following fields are checked in the given order:
        # 1) Network specific realname settings for this service (servers:<netname>:servicename_realname)
        # 2) Global settings for this service (servicename:realname)
        # 3) The globally configured real name (pylink:realname).
        # 4) The literal service name.
        """
        sbconf = conf.conf.get(self.name, {})
        return irc.serverdata.get("%s_realname" % self.name) or sbconf.get('realname') or conf.conf['pylink'].get('realname') or self.name

    def _show_command_help(self, irc, command, private=False, shortform=False):
        """
        Shows help for the given command.
        """
        def _reply(text):
            """
            reply() wrapper to handle the private argument.
            """
            self.reply(irc, text, private=private)

        def _reply_format(next_line):
            """
            Formats and outputs the given line.
            """
            next_line = next_line.strip()
            next_line = NORMALIZEWHITESPACE_RE.sub(' ', next_line)
            _reply(next_line)

        if command not in self.commands:
            _reply('Error: Unknown command %r.' % command)
            return
        else:
            funcs = self.commands[command]
            if len(funcs) > 1:
                _reply('The following \x02%s\x02 plugins bind to the \x02%s\x02 command: %s'
                       % (len(funcs), command, ', '.join([func.__module__ for func in funcs])))
            for func in funcs:
                doc = func.__doc__
                mod = func.__module__
                if doc:
                    lines = doc.splitlines()
                    # Bold the first line, which usually just tells you what
                    # arguments the command takes.
                    args_desc = '\x02%s %s\x02' % (command, lines[0])

                    _reply(args_desc.strip())
                    if not shortform:
                        # Note: we handle newlines in docstrings a bit differently. Per
                        # https://github.com/GLolol/PyLink/issues/307, only double newlines (and
                        # combinations of more) have the effect of showing a new line on IRC.
                        # Single newlines are stripped so that word wrap can be applied in source
                        # code without affecting the output on IRC.
                        # TODO: we should probably verify that the output line doesn't exceed IRC
                        # line length limits...
                        next_line = ''
                        for linenum, line in enumerate(lines[1:], 1):
                            stripped_line = line.strip()
                            log.debug("_show_command_help: Current line (%s): %r", linenum, stripped_line)
                            log.debug("_show_command_help: Last line (%s-1=%s): %r", linenum, linenum-1, lines[linenum-1].strip())

                            if stripped_line:
                                # If this line has content, join it with the previous one.
                                next_line += line.rstrip()
                                next_line += ' '
                            elif linenum > 0 and not lines[linenum-1].strip():
                                # The line before us was empty, so treat this one as a legitimate
                                # newline/break.
                                log.debug("_show_command_help: Adding an extra break...")
                                _reply(' ')
                            else:
                                # Otherwise, output it to IRC.
                                _reply_format(next_line)
                                next_line = ''  # Reset the next line buffer
                        else:
                            _reply_format(next_line)
                else:
                    _reply("Error: Command %r doesn't offer any help." % command)

                # Regardless of whether help text is available, mention aliases.
                if not shortform:
                    if command in self.alias_cmds:
                        _reply(' ')
                        _reply('This command is an alias for \x02%s\x02.' % self.alias_cmds[command])
                    aliases = set(alias for alias, primary in self.alias_cmds.items() if primary == command)
                    if aliases:
                        _reply(' ')
                        _reply('Available aliases: \x02%s\x02' % ', '.join(aliases))

    def help(self, irc, source, args):
        """<command>

        Gives help for <command>, if it is available."""
        try:
            command = args[0].lower()
        except IndexError:
            # No argument given: show service description (if present), 'list' output, and a list
            # of featured commands.
            if self.desc:
                self.reply(irc, self.desc)
                self.reply(irc, " ")  # Extra newline to unclutter the output text

            self.listcommands(irc, source, args)
            return
        else:
            self._show_command_help(irc, command)

    def listcommands(self, irc, source, args):
        """[<plugin name>]

        Returns a list of available commands this service has to offer. The optional
        plugin name argument also allows you to filter commands by plugin (case
        insensitive)."""

        try:
            plugin_filter = args[0].lower()
        except IndexError:
            plugin_filter = None

        # Don't show CTCP handlers or aliases in the public command list.
        cmds = sorted(cmd for cmd in self.commands.keys() if '\x01' not in cmd and cmd not in self.alias_cmds)

        if plugin_filter is not None:
            # Filter by plugin, if the option was given.
            new_cmds = []

            # Add the pylinkirc.plugins prefix to the module name, so it can be used for matching.
            plugin_module = PLUGIN_PREFIX + plugin_filter

            for cmd_definition in cmds:
                for cmdfunc in self.commands[cmd_definition]:
                    if cmdfunc.__module__.lower() == plugin_module:
                        new_cmds.append(cmd_definition)

            # Replace the old command list.
            cmds = new_cmds

        if cmds:
            self.reply(irc, 'Available commands include: %s' % ', '.join(cmds))
            self.reply(irc, 'To see help on a specific command, type \x02help <command>\x02.')
        elif not plugin_filter:
            self.reply(irc, 'This service doesn\'t provide any public commands.')
        else:
            self.reply(irc, 'This service doesn\'t provide any public commands from the plugin %s.' % plugin_filter)

        # If there are featured commands, list them by showing the help for each.
        # These definitions are sent in private to prevent flooding in channels.
        if self.featured_cmds and not plugin_filter:
            self.reply(irc, " ", private=True)
            self.reply(irc, 'Featured commands include:', private=True)
            for cmd in sorted(self.featured_cmds):
                if cmd in cmds:
                    # Only show featured commands that are both defined and loaded.
                    # TODO: perhaps plugin unload should remove unused featured command
                    # definitions automatically?
                    self._show_command_help(irc, cmd, private=True, shortform=True)
            self.reply(irc, 'End of command listing.', private=True)

def register_service(name, *args, **kwargs):
    """Registers a service bot."""
    name = name.lower()
    if name in world.services:
        raise ValueError("Service name %s is already bound!" % name)

    # Allow disabling service spawning either globally or by service.
    elif name != 'pylink' and not (conf.conf.get(name, {}).get('spawn_service',
            conf.conf['pylink'].get('spawn_services', True))):
        return world.services['pylink']

    world.services[name] = sbot = ServiceBot(name, *args, **kwargs)
    sbot.spawn()
    return sbot
registerService = register_service

def unregister_service(name):
    """Unregisters an existing service bot."""
    name = name.lower()

    if name not in world.services:
        # Service bot doesn't exist; ignore.
        return

    sbot = world.services[name]
    for ircnet, uid in sbot.uids.items():
        ircobj = world.networkobjects[ircnet]
        # Special case for the main PyLink client. If we're unregistering that,
        # clear the irc.pseudoclient entry.
        if name == 'pylink':
            ircobj.pseudoclient = None

        ircobj.proto.quit(uid, "Service unloaded.")

    del world.services[name]
unregisterService = unregister_service

def wrap_arguments(prefix, args, length, separator=' ', max_args_per_line=0):
    """
    Takes a static prefix and a list of arguments, and returns a list of strings
    with the arguments wrapped across multiple lines. This is useful for breaking up
    long SJOIN or MODE strings so they aren't cut off by message length limits.
    """
    strings = []

    assert args, "wrap_arguments: no arguments given"

    buf = prefix

    args = list(args)

    while args:
        assert len(prefix+args[0]) <= length, \
            "wrap_arguments: Argument %r is too long for the given length %s" % (args[0], length)

        # Add arguments until our buffer is up to the length limit.
        if (len(buf + args[0]) + 1) <= length and ((not max_args_per_line) or len(buf.split(' ')) < max_args_per_line):
            if buf != prefix:  # Only add a separator if this isn't the first argument of a line
                buf += separator
            buf += args.pop(0)
        else:
            # Once this is full, add the string to the list and reset the buffer.
            strings.append(buf)
            buf = prefix
    else:
        strings.append(buf)

    return strings
wrapArguments = wrap_arguments

class IRCParser(argparse.ArgumentParser):
    """
    Wrapper around argparse.ArgumentParser, without quitting on usage errors.
    """
    REMAINDER = argparse.REMAINDER

    def print_help(self, *args, **kwargs):
        # XXX: find a way to somehow route this through IRC
        raise InvalidArgumentsError("Use help <commandname> to receive help for PyLink commands.")

    def error(self, message, *args, **kwargs):
        raise InvalidArgumentsError(message)
    _print_message = error  # XXX: ugly

    def exit(self, *args):
        return