Permalink
Browse files

Updated bot:

  - USes config file
  - OAuth2
  - SQLite FTS3
  • Loading branch information...
1 parent 490e499 commit 196fe95f9e754ed1f660796e439ae1a1a868aadc @saghul committed Sep 4, 2010
Showing with 119 additions and 90 deletions.
  1. +14 −6 README
  2. +105 −84 twitter-bot.py
View
20 README
@@ -4,17 +4,25 @@ TwitterBot by saghul
TwitterBot is an easy way of creating a Twitter bot which twitts everything
tagged with a givven tag.
-For getting TwitterBot to work you need to create a SQLite database:
+For getting TwitterBot to work you need to create a SQLite database using FTS3:
$ > sqlite3 twitterbot.db
-sqlite> CREATE TABLE repeated_twitts(id integer unique not null);
+sqlite> CREATE VIRTUAL TABLE twitts USING fts3(id, content);
sqlite> .quit
And add a cron job with the necessary parameters:
-*/5 * * * * /home/saghul/twitterbot/twitter-bot.py -u asteriskbot -p your_password -t asterisk
+*/5 * * * * /home/saghul/work/git/twitterbot/twitter-bot.py -t asterisk > /dev/null 2>&1
- * -u: Username
- * -p: Password
- * -t: Hashtag
+ -t: Hashtag
+
+You'll need to edit the config.ini file, because Twitter now only
+allows OAuth based authentication
+
+Dependencies:
+------------
+ * python-application: http://pypi.python.org/pypi/python-application/
+ * python-feedparser: http://feedparser.org
+ * python-oauth2: http://pypi.python.org/pypi/oauth2/
+ * python-twitter (trunk version): http://pypi.python.org/pypi/python-twitter/
View
@@ -1,103 +1,124 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
+# Copyright (C) 2010 Saúl ibarra Corretgé <saghul@gmail.com>
+#
-
-from pysqlite2 import dbapi2 as sqlite
-import urllib
-import urllib2
import feedparser
-import twitter
-import getopt
-import sys
import os
+import re
+import sys
+import twitter
+import urllib
+import urllib2
+from application.configuration import ConfigSection
+from optparse import OptionParser
+from pysqlite2 import dbapi2 as sqlite
-TWITTERBOT_DB = "%s/twitterbot.db" % os.path.dirname(sys.argv[0])
-
-class TwitterBot(object):
-
- def __init__(self, user, passw):
- self.user = user
- self.api = twitter.Api(username=user, password=passw)
-
- def startBot(self, searchtag):
- try:
- con = sqlite.connect(TWITTERBOT_DB)
- con.isolation_level = None
- twitts = self._search(searchtag)
-
- for twitt in reversed(twitts['entries']):
- twitt_id = twitt.id.split(':')[2]
- twitt_author = twitt.author.split(' ')[0]
-
- if self.user == twitt_author:
- # I don't want to RT my own twitts!
- continue
-
- db_id = con.execute("SELECT id FROM repeated_twitts WHERE id = %s LIMIT 1" % twitt_id)
-
- if db_id.fetchall():
- # We already twitted this!
- continue
-
- try:
- self.api.PostUpdate("@%s: %s" % (twitt_author, twitt.title))
-
- except:
- pass
- else:
- con.execute("INSERT INTO repeated_twitts(id) VALUES(%s)" % twitt_id)
+TWITTERBOT_CFG = "%s/config.ini" % (os.path.dirname(sys.argv[0]) or '.')
+TWITTERBOT_DB = "%s/twitterbot.db" % (os.path.dirname(sys.argv[0]) or '.')
- con.close()
+class Config(ConfigSection):
+ __cfgfile__ = TWITTERBOT_CFG
+ __section__ = 'twitterbot'
- except sqlite.Error, e:
- print "[ERROR]", e
- sys.exit(1)
+ consumer_key = ''
+ consumer_secret = ''
+ access_token_key = ''
+ access_token_secret = ''
- def _doRequest(self, url, data=None):
- try:
- opener = urllib2.build_opener()
- opener.addheaders = [ ( 'User-agent', 'Mozilla/5.0' ) ]
- req = urllib2.Request(url, data)
- return opener.open(req)
- except urllib2.HTTPError, err:
- print str(err)
-
- def _search(self, tag, lang='en'):
- data = urllib.urlencode({'tag' : tag, 'lang' : lang})
- f = self._doRequest('http://search.twitter.com/search.atom', data)
- d = feedparser.parse(f.read())
- return d
+class TwitterBot(object):
+ _rt_regex = re.compile(r"^(RT @\w+ )*(?P<tweet>.*)$")
+ _via_regex = re.compile(r"^(?P<tweet>.*)\(via @\w+\)$")
+
+ def __init__(self):
+ self._api = twitter.Api(username=Config.consumer_key,
+ password=Config.consumer_secret,
+ access_token_key=Config.access_token_key,
+ access_token_secret=Config.access_token_secret)
+ try:
+ user = self._api.VerifyCredentials()
+ except twitter.TwitterError, e:
+ raise RuntimeError(str(e))
+ else:
+ self.user = user.GetId()
+
+ def start(self, searchtag):
+ try:
+ con = sqlite.connect(TWITTERBOT_DB)
+ con.isolation_level = None
+ twitts = self.search_tag(searchtag)
+ for twitt in reversed(twitts['entries']):
+ twitt_id = twitt.id.split(':')[2]
+ twitt_author = twitt.author.split(' ')[0]
+ twitt_content = twitt.title
+
+ if self.user == twitt_author:
+ # I don't want to RT my own twitts!
+ continue
+
+ db_id = con.execute("SELECT id FROM twitts WHERE id MATCH %s LIMIT 1" % twitt_id)
+ if db_id.fetchall():
+ # We already twitted this!
+ continue
+
+ # Avoid duplicated twitts because of retwitting
+ tmp = twitt_content
+ if tmp.find('RT @') != -1:
+ tmp = tmp[tmp.find('RT @'):]
+ m = self._rt_regex.match(tmp) or self._via_regex.match(tmp)
+ if m:
+ data = m.groupdict()
+ tmp = data['tweet']
+ if not tmp:
+ continue
+ db_content = con.execute("SELECT id FROM twitts WHERE content MATCH '%s'" % tmp)
+ if db_content.fetchall():
+ continue
+ try:
+ self._api.PostUpdate("RT @%s: %s" % (twitt_author, twitt_content))
+ except twitter.TwitterError:
+ pass
+ else:
+ con.execute("INSERT INTO twitts(id, content) VALUES('%s', '%s')" % (twitt_id, twitt_content))
+ con.close()
+ except sqlite.Error, e:
+ print "[SQLite error: %s]" % str(e)
+ sys.exit(1)
+
+ def search_tag(self, tag, lang='en'):
+ url = 'http://search.twitter.com/search.atom'
+ data = urllib.urlencode({'tag' : tag, 'lang' : lang})
+ try:
+ opener = urllib2.build_opener()
+ opener.addheaders = [ ( 'User-agent', 'Mozilla/5.0' ) ]
+ req = urllib2.Request(url, data)
+ except urllib2.HTTPError, err:
+ print str(err)
+ return {}
+ else:
+ d = feedparser.parse(opener.open(req).read())
+ if d.bozo == 1:
+ return {}
+ return d
if __name__ == "__main__":
- usage = "python twitterbot.py -u username -p password -t search_tag"
- try:
- opts, args = getopt.getopt(sys.argv[1:], "u:p:t:")
- if len(opts) == 3:
- for o, a in opts:
- if o == "-u":
- user = a
- continue
- elif o == "-p":
- passw = a
- continue
- elif o == "-t":
- tag = a
- continue
-
- tb = TwitterBot(user, passw)
- tb.startBot(tag)
- sys.exit(0)
+ if not Config.consumer_key or not Config.consumer_secret or not Config.access_token_key or not Config.access_token_secret:
+ print "Please, fill the confion file"
+ sys.exit(1)
- else:
- print usage
- sys.exit(1)
+ usage = "usage: %prog [options]"
+ parser = OptionParser(usage=usage)
+ parser.add_option('-t', dest='tag', help='Hashtag to search for')
+ options, args = parser.parse_args()
- except getopt.GetoptError, err:
- print str(err)
- print usage
+ if options.tag:
+ bot = TwitterBot()
+ bot.start(options.tag)
+ else:
+ parser.print_help()
sys.exit(1)

0 comments on commit 196fe95

Please sign in to comment.