Allows connecting to an IRC server over the AX26 protocol. 2.2KB

    import kiss2 import ax26 import time import _thread from twisted.words.protocols import irc from twisted.internet import protocol, reactor CALLSIGN = "CALLSIGNHERE-IRC" NICKNAME = "ax26-irc-gateway" SERVER = "" CHANNEL = "#redditnet" k = kiss2.TCPKISS(host='localhost', port=8001) k.start() ax = ax26.Server(k, CALLSIGN) bot = None def blockingMethod(botfunc): #Not really global bot bot = botfunc backlog = "" class IRCClient(irc.IRCClient): nickname = NICKNAME def signedOn(self): self.join( print ("Signed on as %s." % (self.nickname,)) def joined(self, channel): print ("Joined %s." % (channel,)) global bot bot = self def privmsg(self, user, channel, msg): global backlog backlog += "<" + user + "> " + msg + "\r\n" def threadSafeMsg(self, channel, message): reactor.callFromThread(self.msg, channel, message) class IRCClientFactory(protocol.ClientFactory): protocol = IRCClient def __init__(self, channel): = channel def clientConnectionLost(self, connector, reason): print ("Lost connection (%s), reconnecting." % (reason,)) connector.connect() def clientConnectionFailed(self, connector, reason): print ("Could not connect: %s" % (reason,)) #ax26 client = "" def send_msg(msg): global client if client != "": ax.send_data(client, msg.encode()) class handler: def __init__(self, client_callsign): global client client = client_callsign def on_recv(self, data): text = data.decode().strip() bot.threadSafeMsg(channel, text) def on_disconnect(self): client = "" def ax26(): global backlog while (True): if backlog != "": backlog_temp = backlog backlog = "" send_msg(backlog_temp[:-2]) #Remove last \r\n time.sleep(5); _thread.start_new_thread(ax26, ()) def ax26_server(): ax.serve_forever(handler) _thread.start_new_thread(ax26_server, ()) #Setup IRC reactor.connectTCP(SERVER, 6667, IRCClientFactory(channel))