Skip to content

Instantly share code, notes, and snippets.

@blha303
Last active August 31, 2021 23:05
Show Gist options
  • Save blha303/e3cf2d93c932a082c5eb to your computer and use it in GitHub Desktop.
Save blha303/e3cf2d93c932a082c5eb to your computer and use it in GitHub Desktop.
IRC Single Message Bot. Sends a single IRC message to a channel, then quits. Useful for job notifications maybe
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!"))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment