Skip to content

Instantly share code, notes, and snippets.

@dbr
Created April 23, 2009 04:03
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dbr/100295 to your computer and use it in GitHub Desktop.
Save dbr/100295 to your computer and use it in GitHub Desktop.
Simple message queue based Jabber bot
#!/usr/bin/env python
import yaml
from jabberbot import JabberBot
from beanstalkc import Connection as BSC
def connect_beanstalk(host = "127.0.0.1", port = "11300", tube = None):
try:
c = BSC()
except:
return None
else:
if tube is not None:
c.use(tube)
c.ignore("default")
c.watch(tube)
return c
class YayBot(JabberBot):
displayed_jobs = []
bsc = connect_beanstalk(tube = "jabber")
def idle_proc(self):
if self.bsc is None:
self.bsc = connect_beanstalk()
if self.bsc is not None:
self.log("Connected successfully to beanstalkd")
else:
return
try:
job = self.bsc.reserve(timeout=0)
except:
import traceback
self.log("Error talking to beanstalkd: %s" % (traceback.format_exc()))
self.bsc = connect_beanstalk()
else:
if job is not None:
self.displayed_jobs.append(job)
self.log("Got a job from the tube")
self.log("Job: %s" % (job.body))
msg_info = yaml.load(job.body)
self.send(msg_info['target'], msg_info['msg'])
def bot_ok(self, mess, args):
"""Acnowledges sent messages (removes them from queue)"""
for job in self.displayed_jobs:
self.log("Deleting job %s" % (job.jid))
try:
job.delete()
except:
import traceback
self.log("Error deleting job:\n%s" % (traceback.format_exc()))
def bot_quit(self, mess, args):
"""Disconnect (run twice to quit, or run "quit now")"""
if args == "now":
self.quit()
return "Bye"
else:
return """Run "quit now" to disconnect the bot from jabber"""
def main():
"Connect to the server and run the bot forever"
jabber_bot = YayBot("jabberusername", "jabberpassword")
jabber_bot.serve_forever()
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment