add an IRC bot

contrib/boobot.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# Copyright(C) 2012  Romain Bignon
+#
+# This file is part of weboob.
+#
+# weboob is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# weboob is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with weboob. If not, see <http://www.gnu.org/licenses/>.
+
+
+import re
+import sys
+from threading import Thread
+from ircbot import SingleServerIRCBot
+from time import sleep
+
+from weboob.core import Weboob
+from weboob.tools.storage import StandardStorage
+
+IRC_CHANNEL = '#weboob'
+IRC_NICKNAME = 'boobot'
+IRC_SERVER = 'irc.freenode.org'
+STORAGE_FILE = 'boobot.storage'
+
+class MyThread(Thread):
+    def __init__(self, bot):
+        Thread.__init__(self)
+        self.weboob = Weboob(storage=StandardStorage(STORAGE_FILE))
+        self.weboob.load_backends()
+        self.bot = bot
+        self.bot.weboob = self.weboob
+
+    def run(self):
+        while not self.bot.joined:
+            sleep(1)
+
+        sleep(1)
+        self.weboob.repeat(600, self.check_dlfp)
+
+        self.weboob.loop()
+
+    def check_dlfp(self):
+        print 'Checking DLFP...'
+        for backend, msg in self.weboob.do('iter_unread_messages', backends=['dlfp']):
+            for word in ['weboob', 'videoob', 'havesex', 'havedate', u'sàt', u'salut à toi']:
+                if word in msg.content.lower():
+                    url = msg.signature[msg.signature.find('https://linuxfr'):]
+                    self.bot.send_message('[DLFP] %s talks about %s: %s' % (msg.sender, word, url))
+                    break
+            backend.set_message_read(msg)
+        print 'Checked.'
+
+    def stop(self):
+        self.weboob.want_stop()
+
+class TestBot(SingleServerIRCBot):
+    def __init__(self, channel, nickname, server, port=6667):
+        SingleServerIRCBot.__init__(self, [(server, port)], nickname, nickname + "`")
+        self.channel = channel
+        self.joined = False
+        self.weboob = None
+
+    def on_welcome(self, c, e):
+        c.join(self.channel)
+        self.joined = True
+
+    def send_message(self, msg):
+        self.connection.privmsg(self.channel, msg.encode("UTF-8"))
+
+    def on_pubmsg(self, channel, event):
+        text = ' '.join(event.arguments())
+        for m in re.findall('([\w\d_\-]+@\w+)', text):
+            id, backend_name = m.split('@', 1)
+            if backend_name in self.weboob.backend_instances:
+                backend = self.weboob.backend_instances[backend_name]
+                for cap in backend.iter_caps():
+                    func = 'obj_info_%s' % cap.__name__[4:].lower()
+                    if hasattr(self, func):
+                        try:
+                            getattr(self, func)(backend, id)
+                        except Exception, e:
+                            self.send_message('Oops: [%s] %s' % (type(e).__name__, e))
+                        break
+
+    def obj_info_video(self, backend, id):
+        v = backend.get_video(id)
+        if v:
+            self.send_message(u'Video: %s (%s)' % (v.title, v.duration))
+
+    def obj_info_housing(self, backend, id):
+        h = backend.get_housing(id)
+        if h:
+            self.send_message(u'Housing: %s (%sm² / %s%s)' % (h.title, h.area, h.cost, h.currency))
+
+def main():
+    bot = TestBot(IRC_CHANNEL, IRC_NICKNAME, IRC_SERVER)
+
+    thread = MyThread(bot)
+    thread.start()
+
+    try:
+        bot.start();
+    except KeyboardInterrupt:
+        print "Stopped."
+
+    thread.stop()
+
+if __name__ == "__main__":
+    sys.exit(main())