aly / __init__.py

from irc.bot import ServerSpec import irc import sys import json try: with open('smbotconfig.json') as f: conf = json.load(f) except: print "Error in config or no config found, returning to defaults. Please edit smbotconfig.json (http://jsonlint.com/) and restart the bot" with open('smbotconfig.json', 'w') as f: f.write(json.dumps(dict(channel="#blha303", nickname="SingleMessageTest", server="irc.esper.net") ) ) sys.exit(1) class SingleMessageClient(object): reactor_class = irc.client.Reactor def __init__(self): self.reactor = self.reactor_class() self.connection = self.reactor.server() self.dcc_connections = [] self.reactor.add_global_handler("all_events", self._dispatcher, -10) self.reactor.add_global_handler("dcc_disconnect", self._dcc_disconnect, -10) def _dispatcher(self, connection, event): """ Dispatch events to on_<event.type> method, if present. """ do_nothing = lambda c, e: None method = getattr(self, "on_" + event.type, do_nothing) method(connection, event) def _dcc_disconnect(self, c, e): self.dcc_connections.remove(c) def connect(self, *args, **kwargs): """Connect using the underlying connection""" self.connection.connect(*args, **kwargs) def dcc_connect(self, address, port, dcctype="chat"): """Connect to a DCC peer. Arguments: address -- IP address of the peer. port -- Port to connect to. Returns a DCCConnection instance. """ dcc = self.reactor.dcc(dcctype) self.dcc_connections.append(dcc) dcc.connect(address, port) return dcc def dcc_listen(self, dcctype="chat"): """Listen for connections from a DCC peer. Returns a DCCConnection instance. """ dcc = self.reactor.dcc(dcctype) self.dcc_connections.append(dcc) dcc.listen() return dcc def start(self): """Start the IRC client.""" while 1: if not self.connection.is_connected(): break self.reactor.process_once(0.2) class SingleMessageBot(SingleMessageClient): def __init__(self, msg, channel=conf.get('channel'), nickname=conf.get('nickname'), realname=conf.get('realname', conf.get('nickname', None)), server=conf.get('server'), port=conf.get('port', 6667), password=conf.get('password', None), **connect_params): super(SingleMessageBot, self).__init__() self.__connect_params = connect_params self.channel = channel self.server = ServerSpec(server, port, password) self.msg = msg self._nickname = nickname self._realname = realname for i in ["join", "kick", "mode", "namreply", "nick", "part", "quit"]: self.connection.add_global_handler(i, getattr(self, "_on_" + i), -20) def _connect(self): """ Establish a connection to the server at the front of the server_list. """ try: self.connect(self.server.host, self.server.port, self._nickname, self.server.password, ircname=self._realname, **self.__connect_params) except irc.client.ServerConnectionError: pass def _on_join(self, c, e): ch = e.target nick = e.source.nick if nick == c.get_nickname(): self.channels[ch] = Channel() self.channels[ch].add_user(nick) def _on_kick(self, c, e): nick = e.arguments[0] channel = e.target if nick == c.get_nickname(): del self.channels[channel] else: self.channels[channel].remove_user(nick) def _on_mode(self, c, e): modes = irc.modes.parse_channel_modes(" ".join(e.arguments)) t = e.target if irc.client.is_channel(t): ch = self.channels[t] for mode in modes: if mode[0] == "+": f = ch.set_mode else: f = ch.clear_mode f(mode[1], mode[2]) else: # Mode on self... XXX pass def _on_namreply(self, c, e): """ e.arguments[0] == "@" for secret channels, "*" for private channels, "=" for others (public channels) e.arguments[1] == channel e.arguments[2] == nick list """ ch_type, channel, nick_list = e.arguments if channel == '*': # User is not in any visible channel # http://tools.ietf.org/html/rfc2812#section-3.2.5 return for nick in nick_list.split(): nick_modes = [] if nick[0] in self.connection.features.prefix: nick_modes.append(self.connection.features.prefix[nick[0]]) nick = nick[1:] for mode in nick_modes: self.channels[channel].set_mode(mode, nick) self.channels[channel].add_user(nick) def _on_nick(self, c, e): before = e.source.nick after = e.target for ch in self.channels.values(): if ch.has_user(before): ch.change_nick(before, after) def _on_part(self, c, e): nick = e.source.nick channel = e.target if nick == c.get_nickname(): del self.channels[channel] else: self.channels[channel].remove_user(nick) def _on_quit(self, c, e): nick = e.source.nick for ch in self.channels.values(): if ch.has_user(nick): ch.remove_user(nick) def die(self, msg="Bye, cruel world!"): self.connection.disconnect(msg) def get_version(self): """Returns the bot version. Used when answering a CTCP VERSION request. """ return "Python irc.bot ({version})".format( version=irc.client.VERSION_STRING) def jump_server(self, msg="Changing servers"): """Connect to a new server, possibly disconnecting from the current. The bot will skip to next server in the server_list each time jump_server is called. """ if self.connection.is_connected(): self.connection.disconnect(msg) self.server_list.append(self.server_list.pop(0)) self._connect() def on_ctcp(self, c, e): """Default handler for ctcp events. Replies to VERSION and PING requests and relays DCC requests to the on_dccchat method. """ nick = e.source.nick if e.arguments[0] == "VERSION": c.ctcp_reply(nick, "VERSION " + self.get_version()) elif e.arguments[0] == "PING": if len(e.arguments) > 1: c.ctcp_reply(nick, "PING " + e.arguments[1]) elif e.arguments[0] == "DCC" and e.arguments[1].split(" ", 1)[0] == "CHAT": self.on_dccchat(c, e) def on_dccchat(self, c, e): pass def start(self): """Start the bot.""" self._connect() super(SingleMessageBot, self).start() def on_welcome(self, c, e): c.join(self.channel) c.privmsg(self.channel, self.msg) c.close() def main(msg): b = SingleMessageBot(msg) b.start() return 0 if __name__ == "__main__": sys.exit(main(" ".join(sys.argv[1:]) if len(sys.argv) > 1 else "This is a test of SingleMessageBot!"))