Last active
August 31, 2021 23:05
-
-
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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