175 lines
5.5 KiB
Python
175 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import subprocess, re, os, inspect, logging
|
|
|
|
class Bot:
|
|
def __init__(self, cmd, initcmd = [], user=None):
|
|
|
|
if user: os.environ["USER"] = user
|
|
if os.environ["IRCSERVER"]: del os.environ["IRCSERVER"]
|
|
|
|
self.proc = subprocess.Popen(cmd,
|
|
stdin = subprocess.PIPE,
|
|
stdout = subprocess.PIPE,
|
|
stderr = open("/dev/null", mode="w"))
|
|
|
|
for u in initcmd: self.write(u)
|
|
for u in dir(self):
|
|
if u[:3] == "on_":
|
|
self.write('/on ^' + u[3:]
|
|
+ ' * { echo trigger_event.' + u + ' $* }')
|
|
|
|
r = re.compile("trigger_event\.(on_[^ ]+)")
|
|
for line in self.proc.stdout:
|
|
try:
|
|
line = line.decode('utf-8')
|
|
u = r.match(line)
|
|
if u:
|
|
m = getattr(self, u.groups()[0])
|
|
args = re.split(" +", line.rstrip(),
|
|
maxsplit = len(inspect.signature(m).parameters))
|
|
args.pop(0)
|
|
m(*args)
|
|
except Exception as e:
|
|
logging.error("[" + type(e).__name__ + "] " + str(e))
|
|
|
|
def write(self, msg):
|
|
self.proc.stdin.write(bytes(msg + "\n", "utf-8"))
|
|
self.proc.stdin.flush()
|
|
|
|
|
|
logging.basicConfig(format='%(asctime)s %(message)s',
|
|
level=logging.INFO)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
############################
|
|
# Various extended classes #
|
|
############################
|
|
|
|
|
|
# This is the main extended class, which should be used as a starting
|
|
# point for implementing bots (if you need to fire delayed callbacks,
|
|
# use the ExtenddThreadedBot instead).
|
|
class ExtendedBot(Bot):
|
|
"""
|
|
An extended class logging all 'general_notice' and 'connect' info.
|
|
It should be used rather than the low-level bot because it will help
|
|
locating various issues (unless you want to implement these methods
|
|
by yourself for other purposes).
|
|
"""
|
|
def on_general_notice(self, server, first_word, msg):
|
|
logging.info("[%s] -%s- %s" % (server, first_word, msg))
|
|
def on_connect(self, server, port, host):
|
|
logging.info("Connected to %s (port %s) at host %s"
|
|
% (server, port, host))
|
|
|
|
|
|
# If you don't need delayed callbacks, you can safely remove the lines
|
|
# below (including the 'import threading' one) and use the ExtendedBot
|
|
# instead.
|
|
import threading
|
|
class ExtendedThreadedBot(ExtendedBot):
|
|
"""
|
|
A child class of ExtendedBot adding enough threading support for
|
|
allowing a self.delay method for a callback function.
|
|
"""
|
|
def __init__(self, *args, **kwargs):
|
|
self.S_write = threading.Semaphore()
|
|
super(ExtendedThreadedBot, self).__init__(*args, **kwargs)
|
|
|
|
def write(self, msg):
|
|
"""
|
|
Protected version of self.write
|
|
(protected against multithreading simultaneous calls)
|
|
"""
|
|
self.S_write.acquire()
|
|
super(ExtendedThreadedBot, self).write(msg)
|
|
self.S_write.release()
|
|
|
|
def delay(self, time, callback):
|
|
"""
|
|
Set a callback function allowed to call self.write
|
|
(which is protected by a Semaphore)
|
|
"""
|
|
threading.Timer(time, callback).start()
|
|
|
|
|
|
|
|
#########################
|
|
# Various examples bots #
|
|
#########################
|
|
|
|
|
|
# This bot does nothing except replying to private messages
|
|
class Test1(ExtendedBot):
|
|
def on_msg(self, nick, msg):
|
|
self.write("/msg " + nick + " Hi, " + nick + "!")
|
|
|
|
|
|
# This bot writes "I like Shakespeare" whenever someone says
|
|
# something about Shakespare in the current channel; it also
|
|
# quits IRC if someone sends a private message starting with
|
|
# the two words: Just leave!
|
|
class Test2(ExtendedBot):
|
|
def on_msg(self, nick, msg):
|
|
if msg[:11].lower() == "just leave!":
|
|
self.write("/quit I leave!")
|
|
def on_public(self, nick, channel, msg):
|
|
if "shakespeare" in msg.lower():
|
|
self.write("/say I like Shakespeare!")
|
|
|
|
|
|
# This bot uses the 'delay' feature from ExtendedThreadedBot
|
|
class Test3(ExtendedThreadedBot):
|
|
def on_msg(self, nick, msg):
|
|
if msg[:11].lower() == "just leave!":
|
|
self.write("/quit I leave!")
|
|
def on_public(self, nick, channel, msg):
|
|
msg = msg.lower()
|
|
if "shakespeare" in msg:
|
|
self.write("/say I like Shakespeare!")
|
|
elif "dickens" in msg:
|
|
self.delay(5,
|
|
lambda: self.write("/say I also like Dickens!"))
|
|
|
|
|
|
|
|
|
|
# Choose one bot above and run it:
|
|
|
|
## Running an Epic5 bot with SSL
|
|
## =============================
|
|
Test3([ "/home/pi/.local/bin/epic5",
|
|
"-d", "-q", "-c", "#test",
|
|
"-n", "TestNick",
|
|
"irc.efnet.org:6697::::type=IRC-SSL" ],
|
|
user="testus") # export the USER environment variable
|
|
|
|
## Running an ircII bot with SSL
|
|
## =============================
|
|
## Test2([ "ircII",
|
|
## "-d", "-q", "TestNick", "-c", "#test",
|
|
## "SSLIRC/irc.freenode.net:6697" ],
|
|
## initcmd = ["/set novice off"],
|
|
## user = "TestWarrior") # export the USER environment variable
|
|
|
|
## Running a BitchX bot without SSL
|
|
## ================================
|
|
## Test2([ "BitchX", # No SSL
|
|
## "-d", "-q", "TestNick", "-c", "#test",
|
|
## "irc.freenode.net" ],
|
|
## user = "TestWarrior") # export the USER environment variable
|
|
|
|
|
|
## Running a ScrollZ bot without SSL
|
|
## =================================
|
|
## Test2([ "scrollz", # No SSL
|
|
## "-d", "-q", "TestNick", "-c", "#test",
|
|
## "irc.freenode.net" ],
|
|
## user = "TestWarrior") # export the USER environment variable
|