PyLink/utils.py

"""
utils.py - PyLink utilities module.

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

import argparse
import collections
import functools
import importlib
import ipaddress
import os
import re
import string

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

from . import conf, structures, world
from .log import log

PLUGIN_PREFIX = plugins.__name__ + '.'
PROTOCOL_PREFIX = protocols.__name__ + '.'
NORMALIZEWHITESPACE_RE = re.compile(r'\s+')

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, priority=100):
    """
    Binds a hook function to the given command name.

    A custom priority can also be given (defaults to 100), and hooks with
    higher priority values will be called first."""
    command = command.upper()
    world.hooks[command].append((priority, func))
    world.hooks[command].sort(key=lambda pair: pair[0], reverse=True)
    return func

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)
    if not all({nick, ident, host}):
        raise ValueError("Invalid user@host %r" % mask)
    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 plugin-defined persistent channels. The bot will leave them if they're empty,
        # and rejoin whenever someone else does.
        # This is stored as a nested dictionary:
        # {"plugin1": {"net1": IRCCaseInsensitiveSet({"#a", "#b"}), "net2": ...}, ...}
        self.dynamic_channels = {}

        # 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, ignore_empty=None):
        """
        Joins the given service bot to the given channel(s). "channels" can be
        an iterable of channel names or the name of a single channel (type 'str').

        The ignore_empty option sets whether we should skip joining empty
        channels and join them later when we see someone else join (for channels
        marked persistent). This option is automatically *disabled* on networks
        where we cannot monitor channels that we're not in (e.g. Clientbot).

        Before PyLink 2.0-alpha3, this function implicitly marked channels it
        receives to be persistent. This behaviour is no longer the case.
        """
        uid = self.uids.get(irc.name)
        if uid is None:
            return

        if isinstance(channels, str):
            channels = [channels]

        if irc.has_cap('visible-state-only'):
            # Disable dynamic channel joining on networks where we can't monitor channels for joins.
            ignore_empty = False
        elif ignore_empty is None:
            ignore_empty = not (irc.serverdata.get('join_empty_channels',
                                                   conf.conf['pylink'].get('join_empty_channels',
                                                                           False)))

        # Specify modes to join the services bot with.
        joinmodes = irc.get_service_option(self.name, 'joinmodes', default='')
        joinmodes = ''.join([m for m in joinmodes if m in irc.prefixmodes])

        for channel in channels:
            if irc.is_channel(channel):
                if channel in irc.channels:
                    if uid in irc.channels[channel].users:
                        log.debug('(%s/%s) Skipping join to %r - we are already present', irc.name, self.name, channel)
                        continue
                elif ignore_empty:
                    log.debug('(%s/%s) Skipping joining empty channel %r', irc.name, self.name, channel)
                    continue

                log.debug('(%s/%s) Joining channel %s with modes %r', irc.name, self.name, channel, joinmodes)

                if joinmodes:  # Modes on join were specified; use SJOIN to burst our service
                    irc.sjoin(irc.sid, channel, [(joinmodes, uid)])
                else:
                    irc.join(uid, channel)

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

    def part(self, irc, channels, reason=''):
        """
        Parts the given service bot from the given channel(s) if no plugins
        still register it as a persistent dynamic channel.

        "channels" can be an iterable of channel names or the name of a single
        channel (type 'str').
        """
        uid = self.uids.get(irc.name)
        if uid is None:
            return

        if isinstance(channels, str):
            channels = [channels]

        to_part = []
        persistent_channels = self.get_persistent_channels(irc)
        for channel in channels:
            if channel in irc.channels and uid in irc.channels[channel].users:
                if channel in persistent_channels:
                    log.debug('(%s/%s) Not parting %r because it is registered '
                              'as a dynamic channel: %r', irc.name, self.name, channel,
                              persistent_channels)
                    continue
                to_part.append(channel)
                irc.part(uid, channel, reason)
            else:
                log.debug('(%s/%s) Ignoring part to %r, we are not there', irc.name, self.name, channel)
                continue

        irc.call_hooks([uid, 'PYLINK_SERVICE_PART', {'channels': to_part, 'text': reason}])

    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 as e:
                self.reply(irc, 'Error: %s' % e)
                log.warning('(%s) Denying access to command %r for %s; msg: %s', irc.name, cmd,
                            irc.get_hostmask(source), e)
            except 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 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 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 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 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 add_persistent_channel(self, irc, namespace, channel, try_join=True):
        """
        Adds a persistent channel to the service bot on the given network and namespace.
        """
        namespace = self.dynamic_channels.setdefault(namespace, {})
        chanlist = namespace.setdefault(irc.name, structures.IRCCaseInsensitiveSet(irc))
        chanlist.add(channel)

        if try_join and irc.has_cap('can-manage-bot-channels'):
            self.join(irc, [channel])

    def remove_persistent_channel(self, irc, namespace, channel, try_part=True, part_reason=''):
        """
        Removes a persistent channel from the service bot on the given network and namespace.
        """
        chanlist = self.dynamic_channels[namespace][irc.name].remove(channel)

        if try_part and irc.connected.is_set() and irc.has_cap('can-manage-bot-channels'):
            self.part(irc, [channel], reason=part_reason)

    def get_persistent_channels(self, irc, namespace=None):
        """
        Returns a set of persistent channels for the IRC network, optionally filtering
        by namespace is one is given.
        """
        channels = structures.IRCCaseInsensitiveSet(irc)
        if namespace:
            chanlist = self.dynamic_channels.get(namespace, {}).get(irc.name, set())
            log.debug('(%s/%s) get_persistent_channels: adding channels '
                      '%r from namespace %r (single)', irc.name, self.name,
                      chanlist, namespace)
            channels |= chanlist
        else:
            for dch_namespace, dch_data in self.dynamic_channels.items():
                chanlist = dch_data.get(irc.name, set())
                log.debug('(%s/%s) get_persistent_channels: adding channels '
                          '%r from namespace %r', irc.name, self.name,
                          chanlist, dch_namespace)
                channels |= chanlist
        channels |= set(irc.serverdata.get(self.name+'_channels', []))
        channels |= set(irc.serverdata.get('channels', []))
        return channels

    def clear_persistent_channels(self, irc, namespace, try_part=True, part_reason=''):
        """
        Clears the persistent channels defined by a namespace.

        irc can be None to clear persistent channels for all networks in this namespace.
        """
        dch_data = self.dynamic_channels.get(namespace, {})

        if irc is not None:
            if irc.name in dch_data:
                chanlist = dch_data[irc.name]
                log.debug('(%s/%s) Clearing persistent channels %r from namespace %r',
                          irc.name, self.name, chanlist, namespace)

                del dch_data[irc.name]

                if try_part:
                    self.part(irc, chanlist, reason=part_reason)


        else:  # when irc is None
            del self.dynamic_channels[namespace]

            for netname, chanlist in dch_data.items():
                log.debug('(%s/%s) Globally clearing persistent channels %r from namespace %r',
                          netname, self.name, chanlist, namespace)
                if try_part and netname in world.networkobjects:
                    self.part(world.networkobjects[netname], chanlist,
                              reason=part_reason)


    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/jlu5/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.
                        # (On the same topic, real line wrapping on IRC is done in irc.msg() as of
                        #  2.0-beta1)
                        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:
                            # Show the last line.
                            _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.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

# From http://modern.ircdocs.horse/formatting.html
_strip_color_regex = re.compile(r'\x03(\d{1,2}(,\d{1,2})?)?')
_irc_formatting_chars = "\x02\x1D\x1F\x1E\x11\x16\x0F\x03"

def strip_irc_formatting(text):
    """Returns text with IRC formatting (colors, underlines, bold, italics, reverse) removed."""
    text = _strip_color_regex.sub('', text)
    for char in _irc_formatting_chars:
        text = text.replace(char, '')
    return text

_subrange_re = re.compile(r'(?P<start>(\d+))-(?P<end>(\d+))')
def remove_range(rangestr, mylist):
    """
    Removes a range string of (one-indexed) items from the list.
    Range strings are indices or ranges of them joined together with a ",":
    e.g. "5", "2", "2-10", "1,3,5-8"

    See test/test_utils.py for more complete examples.
    """
    if None in mylist:
        raise ValueError("mylist must not contain None!")

    # Split and filter out empty subranges
    ranges = filter(None, rangestr.split(','))
    if not ranges:
        raise ValueError("Invalid range string %r" % rangestr)

    for subrange in ranges:
        match = _subrange_re.match(subrange)
        if match:
            start = int(match.group('start'))
            end = int(match.group('end'))

            if end <= start:
                raise ValueError("Range start (%d) is <= end (%d) in range string %r" %
                                 (start, end, rangestr))
            elif 0 in (end, start):
                raise ValueError("Got range index 0 in range string %r, this function is one-indexed" %
                                 rangestr)

            # For our purposes, make sure the start and end are within the list
            mylist[start-1], mylist[end-1]

            # Replace the entire range with None's
            log.debug('utils.remove_range: removing items from %s to %s: %s', start, end, mylist[start-1:end])
            mylist[start-1:end] = [None] * (end-(start-1))

        elif subrange in string.digits:
            index = int(subrange)
            if index == 0:
                raise ValueError("Got index 0 in range string %r, this function is one-indexed" %
                                 rangestr)
            log.debug('utils.remove_range: removing item %s: %s', index, mylist[index-1])
            mylist[index-1] = None

        else:
            raise ValueError("Got invalid subrange %r in range string %r" %
                             (subrange, rangestr))

    return list(filter(lambda x: x is not None, mylist))

def get_hostname_type(address):
    """
    Returns whether the given address is an IPv4 address (1), IPv6 address (2), or neither
    (0; assumed to be a hostname instead).
    """
    try:
        ip = ipaddress.ip_address(address)
    except ValueError:
        return 0
    else:
        if isinstance(ip, ipaddress.IPv4Address):
            return 1
        elif isinstance(ip, ipaddress.IPv6Address):
            return 2
        else:
            raise ValueError("Got unknown value %r from ipaddress.ip_address()" % address)

_duration_re = re.compile(r"^((?P<week>\d+)w)?((?P<day>\d+)d)?((?P<hour>\d+)h)?((?P<minute>\d+)m)?((?P<second>\d+)s)?$")
def parse_duration(text):
    """
    Takes in a duration string and returns the equivalent amount of seconds.

    Time strings are in the following format:
    - '123'         => 123 seconds
                       (positive integers are treated as # of seconds)
    - '1w2d3h4m5s'  => 1 week, 2 days, 3 hours, 4 minutes, and 5 seconds
                       (must be in decreasing order by unit)
    - '72h'         => 72 hours
    - '1h5s'        => 1 hour and 5 seconds
    and so on...
    """
    # If we get an already valid number, just return it
    if text.isdigit():
        return int(text)

    match = _duration_re.match(text)
    if not match:
        raise ValueError("Failed to parse duration string %r" % text)
    result = 0
    matched = 0

    if match.group('week'):
        result += int(match.group('week'))   *  7 * 24 * 60 * 60
        matched += 1
    if match.group('day'):
        result += int(match.group('day'))    * 24 * 60 * 60
        matched += 1
    if match.group('hour'):
        result += int(match.group('hour'))   * 60 * 60
        matched += 1
    if match.group('minute'):
        result += int(match.group('minute')) * 60
        matched += 1
    if match.group('second'):
        result += int(match.group('second'))
        matched += 1

    if not matched:
        raise ValueError("Failed to parse duration string %r" % text)

    return result

@functools.lru_cache(maxsize=1024)
def _glob2re(glob):
    """Converts an IRC-style glob to a regular expression."""
    patt = ['^']

    for char in glob:
        if char == '*' and patt[-1] != '*':  # Collapse ** into *
            patt.append('.*')
        elif char == '?':
            patt.append('.')
        else:
            patt.append(re.escape(char))

    patt.append('$')
    return ''.join(patt)

def match_text(glob, text, filterfunc=str.lower):
    """
    Returns whether glob matches text. If filterfunc is specified, run filterfunc on glob and text
    before preforming matches.
    """
    if filterfunc:
        glob = filterfunc(glob)
        text = filterfunc(text)

    return re.match(_glob2re(glob), text)

def merge_iterables(A, B):
    """
    Merges the values in two iterables. A and B must be of the same type, and one of the following:

    - list: items are combined as A + B
    - set:  items are combined as A | B
    - dict: items are combined as {**A, **B}
    """
    if type(A) != type(B):
        raise ValueError("inputs must be the same type")

    if isinstance(A, list):
        return A + B
    elif isinstance(A, set):
        return A | B
    elif isinstance(A, dict):
        return {**A, **B}