Commit 0a8a9bab authored by Sébastien Dufromentel's avatar Sébastien Dufromentel
Browse files

Both bots on the same process

parent 848e0721
......@@ -21,73 +21,114 @@
##
######
import sys
import ssl
import sys
import json
import time
import irc.bot
import threading
import collections
import urllib.request
bots, chan, name = {}, "#play.it", "BabelBot"
servers = { "fr": "irc.geeknode.org", "en": "irc.oftc.net" }
chan = "#play.it"
ips = collections.defaultdict(lambda:
"2a00:5884:8300::01"+hex(len(ips))[2:].zfill(2))
ignore = set(irc.strings.IRCFoldedCase(n) for n in
("BabelBot", "C", "ChanServ"))
trans = "https://libretranslate.de/translate"
origin = sys.argv[1]
target = tuple(x for x in servers.keys() if x != origin)[0]
def botnick(usernick, lang):
if usernick.endswith("[m]"):
return irc.strings.IRCFoldedCase(usernick[:-1]+"-"+lang+"]")
return irc.strings.IRCFoldedCase(usernick+"["+lang+"]")
def translate(message, origin, target):
try: # Don't interrupt.
if message == "\o/":
return message
#TODO replace nick/nick[lang] correctly.
with urllib.request.urlopen(trans, urllib.parse.urlencode({ "source":
"auto", "target": target, "q": message }).encode()) as f:
return json.loads(f.read())["translatedText"]
except Exception: # Probably a server error.
return "�[ "+message+" ]�"
def check(main, *args):
if main.conn: # It's alive.
for nick, bot in main.bots.items():
if not bot.conn and time.time()-bot.stamp > 1:
main.bots.pop(nick).die() # Cleaning first…
main.bots[nick] = Bot(botnick(nick, main.origin),
nick, main.server, target=main.target)
elif time.time()-main.stamp > 2: # Need to relaunch it.
main.die() # Let's try this again with another bot.
return Babel(*args, bots=main.bots, server=main.server,
origin=main.origin, target=main.target)
return main # And that's done.
def ip():
base = len(bots)
if origin == "fr":
base = 0xff-base
return "2a00:5884:8300::01"+hex(base)[2:].zfill(2)
class data(object):
def __init__(self, **args):
for name, arg in args.items():
setattr(self, name, arg)
class Bot(irc.bot.SingleServerIRCBot):
def __init__(self, name, url):
irc.bot.SingleServerIRCBot.__init__(self, [(url, 6697)], name, name,
connect_factory=irc.connection.Factory(bind_address=(ip(), 6697),
wrapper=ssl.SSLContext().wrap_socket, ipv6=True))
self.pendings = []
def __init__(self, nick, name, url, **args):
self.conn, self.stamp = False, time.time()
irc.bot.SingleServerIRCBot.__init__(self, [(url, 6697)], nick, name,
connect_factory=irc.connection.Factory(bind_address=(ips[nick],
6697), wrapper=ssl.SSLContext().wrap_socket, ipv6=True))
threading.Thread(None, self.start, daemon=True).start()
for name, arg in args.items():
setattr(self, name, arg)
def on_welcome(self, conn, ev):
conn.join(chan)
self.conn = conn
while len(self.pendings) > 0:
self.say(chan, self.pendings.pop(0))
conn.join(chan)
def say(self, chan, message):
if not hasattr(self, "conn"):
return self.pendings.append(message)
self.conn.privmsg(chan, message)
def on_privmsg(self, conn, ev):
pass #TODO answer that I'm just a bot.
class Babel(Bot):
def on_join(self, conn, nick):
if hasattr(nick, "source"):
nick = nick.source.nick
nick = irc.strings.IRCFoldedCase(nick)
if nick != name and not nick.endswith("["+target+"]"):
if nick not in bots.keys():
bots[nick] = Bot(nick+"["+origin+"]", servers[target])
threading.Thread(None, bots[nick].start, daemon=True).start()
def on_endofnames(self, conn, ev):
def joining():
for nick in tuple(self.channels[chan].users()):
self.on_join(conn, nick)
time.sleep(0.2)
threading.Thread(None, joining, daemon=True).start()
for nick in tuple(self.channels[chan].users()):
self.on_join(conn, data(source=data(nick=nick)))
def on_join(self, conn, ev):
nick = irc.strings.IRCFoldedCase(ev.source.nick)
if nick not in ignore.union(self.bots.keys()):
if not nick.endswith(self.target+"]"):
self.bots[nick] = Bot(botnick(nick, self.origin),
nick, self.server, target=self.target)
def on_nick(self, conn, ev):
pass #TODO quit + join?
def on_quit(self, conn, ev):
nick = irc.strings.IRCFoldedCase(ev.source.nick)
if nick in self.bots.keys():
self.bots.pop(nick).die()
def on_part(self, conn, ev):
self.on_quit(conn, ev)
def on_kick(self, conn, ev):
self.on_quit(conn, ev)
def on_pubmsg(self, conn, ev):
nick = irc.strings.IRCFoldedCase(ev.source.nick)
if nick in bots.keys():
try:
post = urllib.parse.urlencode({ "source": "auto",
"target": target, "q": ev.arguments[0] })
with urllib.request.urlopen(trans, post.encode()) as f:
response = json.loads(f.read())
if "translatedText" in response.keys():
bots[nick].say(ev.target, response["translatedText"])
except Exception:
bots[nick].say(ev.target, "�[ "+ev.arguments[0]+" ]�")
if nick in self.bots.keys():
message = translate(ev.arguments[0], self.origin, self.target)
if self.bots[nick].conn:
self.bots[nick].conn.privmsg(chan, message)
frbot = data(bots={}, conn=False, stamp=-1, die=lambda: None,
origin="en", target="fr", server="irc.geeknode.org")
enbot = data(bots={}, conn=False, stamp=-1, die=lambda: None,
origin="fr", target="en", server="irc.oftc.net")
Babel(name, servers[origin]).start()
while True:
frbot = check(frbot, "BabelBot[fr]",
"Cool translating bot", "irc.oftc.net")
enbot = check(enbot, "BabelBot[en]",
"Chouette bot traducteur", "irc.geeknode.org")
time.sleep(0.5)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment