Source code for evennia.server.portal.irc

This connects to an IRC network/channel and launches an 'bot' onto it.
The bot then pipes what is being said between the IRC channel and one or
more Evennia channels.

import re

from twisted.application import internet
from twisted.internet import protocol, reactor
from twisted.words.protocols import irc

from evennia.server.session import Session
from evennia.utils import ansi, logger, utils

# IRC colors

IRC_BOLD = "\002"
IRC_COLOR = "\003"
IRC_RESET = "\017"
IRC_ITALIC = "\026"
IRC_INVERT = "\x16"

IRC_RED = "4"
IRC_DRED = "5"
IRC_DCYAN = "10"
IRC_CYAN = "11"
IRC_BLUE = "12"
IRC_DGREY = "14"
IRC_GREY = "15"

# obsolete test:

# test evennia->irc:
# |rred |ggreen |yyellow |bblue |mmagenta |ccyan |wwhite |xdgrey
# |Rdred |Gdgreen |Ydyellow |Bdblue |Mdmagenta |Cdcyan |Wlgrey |Xblack
# |[rredbg |[ggreenbg |[yyellowbg |[bbluebg |[mmagentabg |[ccyanbg |[wlgreybg |[xblackbg

# test irc->evennia
# Use Ctrl+C <num> to produce mIRC colors in e.g. irssi

        (r"|n", IRC_COLOR + IRC_NORMAL),  # normal mode
        (r"|H", IRC_RESET),  # un-highlight
        (r"|/", "\n"),  # line break
        (r"|t", "    "),  # tab
        (r"|-", "    "),  # fixed tab
        (r"|_", " "),  # space
        (r"|*", IRC_INVERT),  # invert
        (r"|^", ""),  # blinking text
        (r"|h", IRC_BOLD),  # highlight, use bold instead
        (r"|r", IRC_COLOR + IRC_RED),
        (r"|g", IRC_COLOR + IRC_GREEN),
        (r"|y", IRC_COLOR + IRC_YELLOW),
        (r"|b", IRC_COLOR + IRC_BLUE),
        (r"|m", IRC_COLOR + IRC_MAGENTA),
        (r"|c", IRC_COLOR + IRC_CYAN),
        (r"|w", IRC_COLOR + IRC_WHITE),  # pure white
        (r"|x", IRC_COLOR + IRC_DGREY),  # dark grey
        (r"|R", IRC_COLOR + IRC_DRED),
        (r"|G", IRC_COLOR + IRC_DGREEN),
        (r"|Y", IRC_COLOR + IRC_DYELLOW),
        (r"|B", IRC_COLOR + IRC_DBLUE),
        (r"|M", IRC_COLOR + IRC_DMAGENTA),
        (r"|C", IRC_COLOR + IRC_DCYAN),
        (r"|W", IRC_COLOR + IRC_GREY),  # light grey
        (r"|X", IRC_COLOR + IRC_BLACK),  # pure black
        (r"|[r", IRC_COLOR + IRC_NORMAL + "," + IRC_DRED),
        (r"|[g", IRC_COLOR + IRC_NORMAL + "," + IRC_DGREEN),
        (r"|[y", IRC_COLOR + IRC_NORMAL + "," + IRC_DYELLOW),
        (r"|[b", IRC_COLOR + IRC_NORMAL + "," + IRC_DBLUE),
        (r"|[m", IRC_COLOR + IRC_NORMAL + "," + IRC_DMAGENTA),
        (r"|[c", IRC_COLOR + IRC_NORMAL + "," + IRC_DCYAN),
        (r"|[w", IRC_COLOR + IRC_NORMAL + "," + IRC_GREY),  # light grey background
        (r"|[x", IRC_COLOR + IRC_NORMAL + "," + IRC_BLACK),  # pure black background
# ansi->irc
RE_ANSI_COLOR = re.compile(r"|".join([re.escape(key) for key in IRC_COLOR_MAP.keys()]), re.DOTALL)
RE_MXP = re.compile(r"\|lc(.*?)\|lt(.*?)\|le", re.DOTALL)
RE_ANSI_ESCAPES = re.compile(r"(%s)" % "|".join(("{{", "%%", "\\\\")), re.DOTALL)
# irc->ansi
    re.escape(val) for val in sorted(IRC_COLOR_MAP.values(), key=len, reverse=True) if val.strip()
_CLR_LIST = _CLR_LIST[-2:] + _CLR_LIST[:-2]
RE_IRC_COLOR = re.compile(r"|".join(_CLR_LIST), re.DOTALL)
ANSI_COLOR_MAP = dict((tup[1], tup[0]) for tup in IRC_COLOR_MAP.items() if tup[1].strip())

[docs]def parse_ansi_to_irc(string): """ Parse |-type syntax and replace with IRC color markers Args: string (str): String to parse for ANSI colors. Returns: parsed_string (str): String with replaced ANSI colors. """ def _sub_to_irc(ansi_match): return IRC_COLOR_MAP.get(, "") in_string = utils.to_str(string) parsed_string = [] parts = RE_ANSI_ESCAPES.split(in_string) + [" "] for part, sep in zip(parts[::2], parts[1::2]): pstring = RE_ANSI_COLOR.sub(_sub_to_irc, part) parsed_string.append("%s%s" % (pstring, sep[0].strip())) # strip mxp parsed_string = RE_MXP.sub(r"\2", "".join(parsed_string)) return parsed_string
[docs]def parse_irc_to_ansi(string): """ Parse IRC mIRC color syntax and replace with Evennia ANSI color markers Args: string (str): String to parse for IRC colors. Returns: parsed_string (str): String with replaced IRC colors. """ def _sub_to_ansi(irc_match): return ANSI_COLOR_MAP.get(, "") in_string = utils.to_str(string) pstring = RE_IRC_COLOR.sub(_sub_to_ansi, in_string) return pstring
# IRC bot
[docs]class IRCBot(irc.IRCClient, Session): """ An IRC bot that tracks activity in a channel as well as sends text to it when prompted """ lineRate = 1 # assigned by factory at creation nickname = None logger = None factory = None channel = None sourceURL = ""
[docs] def signedOn(self): """ This is called when we successfully connect to the network. We make sure to now register with the game as a full session. """ self.join( self.stopping = False = self address = "%s@%s" % (, self.init_session("ircbot", address, self.factory.sessionhandler) # we link back to our bot and log in self.uid = int(self.factory.uid) self.logged_in = True self.factory.sessionhandler.connect(self) logger.log_info( "IRC bot '%s' connected to %s at %s:%s." % (self.nickname,,, self.port) )
[docs] def disconnect(self, reason=""): """ Called by sessionhandler to disconnect this protocol. Args: reason (str): Motivation for the disconnect. """ self.sessionhandler.disconnect(self) self.stopping = True self.transport.loseConnection()
[docs] def at_login(self): pass
[docs] def privmsg(self, user, channel, msg): """ Called when the connected channel receives a message. Args: user (str): User name sending the message. channel (str): Channel name seeing the message. msg (str): The message arriving from channel. """ if channel == self.nickname: # private message user = user.split("!", 1)[0] self.data_in(text=msg, type="privmsg", user=user, channel=channel) elif not msg.startswith("***"): # channel message user = user.split("!", 1)[0] user = ansi.raw(user) self.data_in(text=msg, type="msg", user=user, channel=channel)
[docs] def action(self, user, channel, msg): """ Called when an action is detected in channel. Args: user (str): User name sending the message. channel (str): Channel name seeing the message. msg (str): The message arriving from channel. """ if not msg.startswith("**"): user = user.split("!", 1)[0] self.data_in(text=msg, type="action", user=user, channel=channel)
[docs] def get_nicklist(self): """ Retrieve name list from the channel. The return is handled by the catch methods below. """ if not self.nicklist: self.sendLine("NAMES %s" %
[docs] def irc_RPL_NAMREPLY(self, prefix, params): """ "Handles IRC NAME request returns (nicklist)""" channel = params[2].lower() if channel != return self.nicklist += params[3].split(" ")
[docs] def irc_RPL_ENDOFNAMES(self, prefix, params): """Called when the nicklist has finished being returned.""" channel = params[1].lower() if channel != return self.data_in( text="", type="nicklist", user="server", channel=channel, nicklist=self.nicklist ) self.nicklist = []
[docs] def pong(self, user, time): """ Called with the return timing from a PING. Args: user (str): Name of user time (float): Ping time in secs. """ self.data_in(text="", type="ping", user="server",, timing=time)
[docs] def data_in(self, text=None, **kwargs): """ Data IRC -> Server. Keyword Args: text (str): Ingoing text. kwargs (any): Other data from protocol. """ self.sessionhandler.data_in(self, bot_data_in=[parse_irc_to_ansi(text), kwargs])
[docs] def send_channel(self, *args, **kwargs): """ Send channel text to IRC channel (visible to all). Note that we don't handle the "text" send (it's rerouted to send_default which does nothing) - this is because the IRC bot is a normal session and would otherwise report anything that happens to it to the IRC channel (such as it seeing server reload messages). Args: text (str): Outgoing text """ text = args[0] if args else "" if text: text = parse_ansi_to_irc(text) self.say(, text)
[docs] def send_privmsg(self, *args, **kwargs): """ Send message only to specific user. Args: text (str): Outgoing text. Keyword Args: user (str): the nick to send privately to. """ text = args[0] if args else "" user = kwargs.get("user", None) if text and user: text = parse_ansi_to_irc(text) self.msg(user, text)
[docs] def send_request_nicklist(self, *args, **kwargs): """ Send a request for the channel nicklist. The return (handled by `self.irc_RPL_ENDOFNAMES`) will be sent back as a message with type `nicklist'. """ self.get_nicklist()
[docs] def send_ping(self, *args, **kwargs): """ Send a ping. The return (handled by `self.pong`) will be sent back as a message of type 'ping'. """
[docs] def send_reconnect(self, *args, **kwargs): """ The server instructs us to rebuild the connection by force, probably because the client silently lost connection. """ self.factory.reconnect()
[docs] def send_default(self, *args, **kwargs): """ Ignore other types of sends. """ pass
[docs]class IRCBotFactory(protocol.ReconnectingClientFactory): """ Creates instances of IRCBot, connecting with a staggered increase in delay """ # scaling reconnect time initialDelay = 1 factor = 1.5 maxDelay = 60
[docs] def __init__( self, sessionhandler, uid=None, botname=None, channel=None, network=None, port=None, ssl=None, ): """ Storing some important protocol properties. Args: sessionhandler (SessionHandler): Reference to the main Sessionhandler. Keyword Args: uid (int): Bot user id. botname (str): Bot name (seen in IRC channel). channel (str): IRC channel to connect to. network (str): Network address to connect to. port (str): Port of the network. ssl (bool): Indicates SSL connection. """ self.sessionhandler = sessionhandler self.uid = uid self.nickname = str(botname) = str(channel) = str(network) self.port = port self.ssl = ssl = None self.nicklists = {}
[docs] def buildProtocol(self, addr): """ Build the protocol and assign it some properties. Args: addr (str): Not used; using factory data. """ protocol = IRCBot() protocol.factory = self protocol.nickname = self.nickname = = protocol.port = self.port protocol.ssl = self.ssl protocol.nicklist = [] return protocol
[docs] def startedConnecting(self, connector): """ Tracks reconnections for debugging. Args: connector (Connector): Represents the connection. """ logger.log_info("(re)connecting to %s" %
[docs] def clientConnectionFailed(self, connector, reason): """ Called when Client failed to connect. Args: connector (Connection): Represents the connection. reason (str): The reason for the failure. """ self.retry(connector)
[docs] def clientConnectionLost(self, connector, reason): """ Called when Client loses connection. Args: connector (Connection): Represents the connection. reason (str): The reason for the failure. """ if not ( or ( and self.retry(connector)
[docs] def reconnect(self): """ Force a reconnection of the bot protocol. This requires de-registering the session and then reattaching a new one, otherwise you end up with an ever growing number of bot sessions. """ = True self.sessionhandler.server_disconnect( self.start()
[docs] def start(self): """ Connect session to sessionhandler. """ if self.port: if self.ssl: try: from twisted.internet import ssl service = reactor.connectSSL(, int(self.port), self, ssl.ClientContextFactory() ) except ImportError: logger.log_err("To use SSL, the PyOpenSSL module must be installed.") else: service = internet.TCPClient(, int(self.port), self)