forked from khuxkm/cosmicbot
94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
import teambot,tasks,rss,time,sys,subprocess,re
|
|
|
|
feed_url = "https://cosmic.voyage/rss.xml"
|
|
bot_op = "khuxkm@sudoers.tilde.team"
|
|
|
|
def unhighlight_nick (nick):
|
|
return "_{}_".format(nick)
|
|
|
|
def datetimejoin(joiner, parts):
|
|
return joiner.join([str(x).rjust(2, '0') for x in parts])
|
|
|
|
def newpost(it):
|
|
return dict(
|
|
guid = it.guid,
|
|
title = it.title,
|
|
link = it.link,
|
|
date = datetimejoin('-', it.published_parsed[0:3]) + ' ' + datetimejoin(':', it.published_parsed[3:6])
|
|
)
|
|
|
|
class CosmicBot(teambot.Handler):
|
|
def __init__(self,bot):
|
|
super(CosmicBot,self).__init__(bot)
|
|
self.prefix = "!"
|
|
self.tasks = tasks.TaskPool()
|
|
self.tasks.add_coroutine(self.check_rss,12,dict(url=feed_url,known=[],channel="#cosmic"))
|
|
self.tasks.load_state(0)
|
|
self.commands = dict()
|
|
self.register_command("botlist",self.on_botlist)
|
|
self.register_command("roster",self.on_roster)
|
|
self.register_command("latest",self.on_latest)
|
|
self.register_command("fortune",self.on_fortune)
|
|
self.register_command("admin",self.on_admin,True)
|
|
def register_command(self,name,action,is_admin=False):
|
|
self.commands[name] = dict(action=action,is_admin=is_admin)
|
|
def on_connection_established(self):
|
|
self.tasks.run()
|
|
def check_rss(self,state,base_state):
|
|
newtrans = rss.fetchNew(state["url"],[x["guid"] for x in state["known"]])
|
|
if newtrans:
|
|
state["known"].extend(newpost(x) for x in newtrans)
|
|
for trans in newtrans:
|
|
self.say(state["channel"],"Transmission received: {transmission.title} ({transmission.link})".format(transmission=trans))
|
|
time.sleep(1)
|
|
return state
|
|
def on_pubmsg(self,channel,nick,text):
|
|
if not text.startswith(self.prefix):
|
|
return
|
|
args = text[len(self.prefix):].strip().split()
|
|
cmd = args.pop(0)
|
|
is_admin = self.event.source.userhost == bot_op
|
|
if cmd in self.commands:
|
|
if (not self.commands[cmd]["is_admin"]) or (self.commands[cmd]["is_admin"] and is_admin):
|
|
try:
|
|
self.commands[cmd]["action"](channel,nick,*args)
|
|
except Exception as e:
|
|
print("access violation "+str(e))
|
|
def on_botlist(self,channel,nick):
|
|
self.say(channel,nick+": Maintainer: khuxkm@cosmic.voyage | Utility bot")
|
|
def on_admin(self,channel,nick,subcmd,*args):
|
|
if subcmd=="down":
|
|
self.tasks.stop()
|
|
self.tasks.save_state(0)
|
|
self._bot.die("Stopping...")
|
|
sys.exit(0)
|
|
elif subcmd=="check":
|
|
self.tasks.states[0] = self.check_rss(self.tasks.states[0],self.tasks.base_state)
|
|
def on_roster(self,channel,nick,*namecnt):
|
|
output = subprocess.check_output(["/usr/local/bin/roster",' '.join(namecnt)]).decode("ascii").split("\n")
|
|
output = filter(None,output)
|
|
for line in output:
|
|
line = re.sub("\s+"," ",line).split(" ",1)
|
|
self.say(channel,"{}: {} (by {})".format(nick,line[1],unhighlight_nick(line[0])))
|
|
def on_latest(self,channel,nick,count=5):
|
|
count = int(count)
|
|
if count < 1:
|
|
count = 1 # ...nice try, smartass
|
|
if count > 5:
|
|
count = 5 # don't spam the channel
|
|
self.say(channel, "{}: Latest {} {}. (See cosmic.voyage for more!)".format(nick, count, (count == 1 and "entry" or "entries")))
|
|
output = subprocess.check_output(["/usr/local/bin/latest",str(count)]).decode("ascii").split("\n")
|
|
output = filter(None,output)
|
|
for line in output:
|
|
self.say(channel,"{}: {}".format(nick,line))
|
|
def on_fortune(self,channel,nick):
|
|
output = subprocess.check_output(["/usr/games/fortune"]).decode("ascii").split("\n")
|
|
output = filter(None,output)
|
|
for line in output:
|
|
self.say(channel,"{}".format(line))
|
|
|
|
if __name__=="__main__":
|
|
channels = "#cosmic".split()
|
|
bot = teambot.TeamBot(channels,"cosmicbot","localhost",chandler=CosmicBot)
|
|
bot.start()
|