bss/dr.botzo
1
0
Fork 0
Code Issues 3 Pull Requests Releases 2 Wiki Activity

Twitter: some code cleanups, use self.twit.VerifyCredentials rather than an authed variable

Browse Source
This commit is contained in:
Brian S. Stephan 2012-07-14 08:17:21 -05:00
parent d50b0867e8
commit 82765c7404
1 changed files with 47 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
modules/Twitter.py
View File

@ -39,12 +39,12 @@ class Twitter(Module):
Module.__init__(self, irc, config, server) Module.__init__(self, irc, config, server)
# setup regexes # setup regexes
getstatuspattern = '^!twitter\s+getstatus\s+(\S+)$' getstatuspattern = "^!twitter\s+getstatus\s+(\S+)$"
getuserstatuspattern = '^!twitter\s+getuserstatus\s+(\S+)(\s+.*|$)' getuserstatuspattern = "^!twitter\s+getuserstatus\s+(\S+)(\s+.*|$)"
tweetpattern = '^!twitter\s+tweet\s+(.*)' tweetpattern = "^!twitter\s+tweet\s+(.*)"
gettokenpattern = '^!twitter\s+gettoken$' gettokenpattern = "^!twitter\s+gettoken$"
authpattern = '^!twitter\s+auth\s+(\S+)$' authpattern = "^!twitter\s+auth\s+(\S+)$"
replytopattern = '^!twitter\s+replyto\s+(\S+)\s+(.*)' replytopattern = "^!twitter\s+replyto\s+(\S+)\s+(.*)"
self.getstatusre = re.compile(getstatuspattern) self.getstatusre = re.compile(getstatuspattern)
self.getuserstatusre = re.compile(getuserstatuspattern) self.getuserstatusre = re.compile(getuserstatuspattern)
@ -57,16 +57,15 @@ class Twitter(Module):
self.consumer_key = 'N2aSGxBP8t3cCgWyF1B2Aw' self.consumer_key = 'N2aSGxBP8t3cCgWyF1B2Aw'
self.consumer_secret = '0aQPEV4K3MMpicfi2lDtCP5pvjsKaqIpfuWtsPzx8' self.consumer_secret = '0aQPEV4K3MMpicfi2lDtCP5pvjsKaqIpfuWtsPzx8'
self.request_token_url = 'https://api.twitter.com/oauth/request_token' self.request_token_url = "https://api.twitter.com/oauth/request_token"
self.access_token_url = 'https://api.twitter.com/oauth/access_token' self.access_token_url = "https://api.twitter.com/oauth/access_token"
self.authorize_url = 'https://api.twitter.com/oauth/authorize' self.authorize_url = "https://api.twitter.com/oauth/authorize"
self.consumer = oauth.Consumer(self.consumer_key, self.consumer_secret) self.consumer = oauth.Consumer(self.consumer_key, self.consumer_secret)
self.client = oauth.Client(self.consumer) self.client = oauth.Client(self.consumer)
# create a default twitter API account, in case we never auth # create a default twitter API account, in case we never auth
self.twit = twitter.Api() self.twit = twitter.Api()
self.authed = False
self.next_timeline_check = 0 self.next_timeline_check = 0
thread.start_new_thread(self.thread_do, ()) thread.start_new_thread(self.thread_do, ())
@ -80,13 +79,13 @@ class Twitter(Module):
# create tables # create tables
db = self.get_db() db = self.get_db()
try: try:
db.execute(''' db.execute("""
CREATE TABLE twitter_settings ( CREATE TABLE twitter_settings (
since_id INTEGER NOT NULL, since_id INTEGER NOT NULL,
output_channel TEXT NOT NULL output_channel TEXT NOT NULL
)''') )""")
db.execute('''INSERT INTO twitter_settings (since_id, output_channel) VALUES (0, '#drbotzo')''') db.execute("""INSERT INTO twitter_settings (since_id, output_channel) VALUES (0, '#drbotzo')""")
db.execute('INSERT INTO drbotzo_modules VALUES (?,?)', (self.__class__.__name__, 1)) db.execute("INSERT INTO drbotzo_modules VALUES (?,?)", (self.__class__.__name__, 1))
db.commit() db.commit()
db.close() db.close()
except sqlite3.Error as e: except sqlite3.Error as e:
@ -100,7 +99,6 @@ class Twitter(Module):
Module.shutdown(self) Module.shutdown(self)
self.twit.ClearCredentials() self.twit.ClearCredentials()
self.authed = False
def do(self, connection, event, nick, userhost, what, admin_unlocked): def do(self, connection, event, nick, userhost, what, admin_unlocked):
"""Attempt to do twitter things.""" """Attempt to do twitter things."""
@ -128,7 +126,7 @@ class Twitter(Module):
tweet = self.twit.GetStatus(status) tweet = self.twit.GetStatus(status)
return self._return_tweet_or_retweet_text(tweet=tweet, print_source=True) return self._return_tweet_or_retweet_text(tweet=tweet, print_source=True)
except twitter.TwitterError as e: except twitter.TwitterError as e:
return 'Couldn\'t obtain status: ' + str(e) return "Couldn't obtain status: " + str(e)
def twitter_getuserstatus(self, connection, event, nick, userhost, what, admin_unlocked): def twitter_getuserstatus(self, connection, event, nick, userhost, what, admin_unlocked):
"""Get a status for a user. Allows for getting one other than the most recent.""" """Get a status for a user. Allows for getting one other than the most recent."""
@ -146,7 +144,7 @@ class Twitter(Module):
else: else:
index = 0 index = 0
except ValueError as e: except ValueError as e:
print('Couldn\'t convert index: ' + str(e)) print("Couldn't convert index: " + str(e))
index = 0 index = 0
count = (-1*index) + 1 count = (-1*index) + 1
@ -157,9 +155,9 @@ class Twitter(Module):
tweet = tweets[-1*index] tweet = tweets[-1*index]
return self._return_tweet_or_retweet_text(tweet=tweet) return self._return_tweet_or_retweet_text(tweet=tweet)
except twitter.TwitterError as e: except twitter.TwitterError as e:
return 'Couldn\'t obtain status: ' + str(e) return "Couldn't obtain status: " + str(e)
except ValueError as e: except ValueError as e:
return 'Couldn\'t obtain status: ' + str(e) return "Couldn't obtain status: " + str(e)
def twitter_tweet(self, connection, event, nick, userhost, what, admin_unlocked): def twitter_tweet(self, connection, event, nick, userhost, what, admin_unlocked):
"""Tweet. Needs authentication.""" """Tweet. Needs authentication."""
@ -167,18 +165,18 @@ class Twitter(Module):
match = self.tweetre.search(what) match = self.tweetre.search(what)
if match: if match:
tweet = match.group(1) tweet = match.group(1)
if self.authed is False: if self.twit.VerifyCredentials() is None:
return 'You must be authenticated to tweet.' return "You must be authenticated to tweet."
if admin_unlocked is False: if admin_unlocked is False:
return 'Only admins can tweet.' return "Only admins can tweet."
try: try:
if self.twit.PostUpdates(tweet, continuation='\xe2\x80\xa6') is not None: if self.twit.PostUpdates(tweet, continuation='\xe2\x80\xa6') is not None:
return 'Tweet(s) sent.' return "'{0:s}' tweeted.".format(tweet)
else: else:
return 'Unknown error sending tweet(s).' return "Unknown error sending tweet(s)."
except twitter.TwitterError as e: except twitter.TwitterError as e:
return 'Couldn\'t tweet: ' + str(e) return "Couldn't tweet: " + str(e)
def twitter_replyto(self, connection, event, nick, userhost, what, admin_unlocked): def twitter_replyto(self, connection, event, nick, userhost, what, admin_unlocked):
"""Reply to a tweet, in the twitter in_reply_to_status_id sense. Needs authentication.""" """Reply to a tweet, in the twitter in_reply_to_status_id sense. Needs authentication."""
@ -188,21 +186,21 @@ class Twitter(Module):
status_id = match.group(1) status_id = match.group(1)
tweet = match.group(2) tweet = match.group(2)
if self.authed is False: if self.twit.VerifyCredentials() is None:
return 'You must be authenticated to tweet.' return "You must be authenticated to tweet."
if admin_unlocked is False: if admin_unlocked is False:
return 'Only admins can tweet.' return "Only admins can tweet."
replyee_tweet = self.twit.GetStatus(status_id) replyee_tweet = self.twit.GetStatus(status_id)
target = replyee_tweet.user.screen_name.encode('utf-8', 'ignore') target = replyee_tweet.user.screen_name.encode('utf-8', 'ignore')
try: try:
if self.twit.PostUpdate('@'+target+': '+tweet, in_reply_to_status_id=status_id) is not None: if self.twit.PostUpdate("@"+target+": "+tweet, in_reply_to_status_id=status_id) is not None:
return 'Tweet sent.' return "'{0:s}' tweeted.".format(tweet)
else: else:
return 'Unknown error sending tweet.' return "Unknown error sending tweet."
except twitter.TwitterError as e: except twitter.TwitterError as e:
return 'Couldn\'t tweet: ' + str(e) return "Couldn't tweet: " + str(e)
def twitter_gettoken(self, connection, event, nick, userhost, what, admin_unlocked): def twitter_gettoken(self, connection, event, nick, userhost, what, admin_unlocked):
"""Get an oauth token, so that the user may authenticate the bot.""" """Get an oauth token, so that the user may authenticate the bot."""
@ -217,7 +215,7 @@ class Twitter(Module):
self.request_token = dict(urlparse.parse_qsl(content)) self.request_token = dict(urlparse.parse_qsl(content))
# have the user auth # have the user auth
return 'Go to the following link in your browser: %s?oauth_token=%s and then send me the pin.' % (self.authorize_url, self.request_token['oauth_token']) return "Go to the following link in your browser: %s?oauth_token=%s and then send me the pin." % (self.authorize_url, self.request_token['oauth_token'])
def twitter_auth(self, connection, event, nick, userhost, what, admin_unlocked): def twitter_auth(self, connection, event, nick, userhost, what, admin_unlocked):
"""Authenticate, given a PIN (following gettoken).""" """Authenticate, given a PIN (following gettoken)."""
@ -237,7 +235,6 @@ class Twitter(Module):
# finally, create the twitter API object # finally, create the twitter API object
self.twit = twitter.Api(self.consumer_key, self.consumer_secret, self.access_token['oauth_token'], self.access_token['oauth_token_secret']) self.twit = twitter.Api(self.consumer_key, self.consumer_secret, self.access_token['oauth_token'], self.access_token['oauth_token_secret'])
self.authed = True
# ugly, but the best place to track it. store the connection for later use # ugly, but the best place to track it. store the connection for later use
self.connection = connection self.connection = connection
@ -245,7 +242,7 @@ class Twitter(Module):
# print timeline stuff. this will set up the appropriate timer # print timeline stuff. this will set up the appropriate timer
self._check_self_timeline() self._check_self_timeline()
return 'The bot is now logged in.' return "The bot is now logged in."
def thread_do(self): def thread_do(self):
"""Check the timeline.""" """Check the timeline."""
@ -260,7 +257,7 @@ class Twitter(Module):
if self.next_timeline_check < time.time(): if self.next_timeline_check < time.time():
self.next_timeline_check = time.time() + 300 self.next_timeline_check = time.time() + 300
if self.authed: if self.twit.VerifyCredentials() is not None:
# get the id of the last check we made # get the id of the last check we made
since_id = self._get_last_since_id() since_id = self._get_last_since_id()
output_channel = self._get_output_channel() output_channel = self._get_output_channel()
@ -297,21 +294,28 @@ class Twitter(Module):
if tweet.retweeted_status: if tweet.retweeted_status:
retweet = tweet.retweeted_status retweet = tweet.retweeted_status
if print_source: if print_source:
return '@%s (RT @%s): %s [%s]' % (tweet.user.screen_name.encode('utf-8', 'ignore'), retweet.user.screen_name.encode('utf-8', 'ignore'), super(Twitter, self)._unencode_xml(retweet.text.encode('utf-8', 'ignore')), tweet.id) return "@%s (RT @%s): %s [%s]" % (tweet.user.screen_name.encode('utf-8', 'ignore'),
retweet.user.screen_name.encode('utf-8', 'ignore'),
super(Twitter, self)._unencode_xml(retweet.text.encode('utf-8', 'ignore')),
tweet.id)
else: else:
return '(RT @%s): %s [%s]' % (retweet.user.screen_name.encode('utf-8', 'ignore'), super(Twitter, self)._unencode_xml(retweet.text.encode('utf-8', 'ignore')), tweet.id) return "(RT @%s): %s [%s]" % (retweet.user.screen_name.encode('utf-8', 'ignore'),
super(Twitter, self)._unencode_xml(retweet.text.encode('utf-8', 'ignore')),
tweet.id)
else: else:
if print_source: if print_source:
return '@%s: %s [%s]' % (tweet.user.screen_name.encode('utf-8', 'ignore'), super(Twitter, self)._unencode_xml(tweet.text.encode('utf-8', 'ignore')), tweet.id) return "@%s: %s [%s]" % (tweet.user.screen_name.encode('utf-8', 'ignore'),
super(Twitter, self)._unencode_xml(tweet.text.encode('utf-8', 'ignore')),
tweet.id)
else: else:
return '%s [%s]' % (super(Twitter, self)._unencode_xml(tweet.text.encode('utf-8', 'ignore')), tweet.id) return "%s [%s]" % (super(Twitter, self)._unencode_xml(tweet.text.encode('utf-8', 'ignore')), tweet.id)
def _get_last_since_id(self): def _get_last_since_id(self):
"""Get the since_id out of the database.""" """Get the since_id out of the database."""
try: try:
db = self.get_db() db = self.get_db()
query = 'SELECT since_id FROM twitter_settings' query = "SELECT since_id FROM twitter_settings"
cursor = db.execute(query) cursor = db.execute(query)
result = cursor.fetchone() result = cursor.fetchone()
db.close() db.close()
@ -327,7 +331,7 @@ class Twitter(Module):
try: try:
db = self.get_db() db = self.get_db()
query = 'SELECT output_channel FROM twitter_settings' query = "SELECT output_channel FROM twitter_settings"
cursor = db.execute(query) cursor = db.execute(query)
result = cursor.fetchone() result = cursor.fetchone()
db.close() db.close()
@ -344,7 +348,7 @@ class Twitter(Module):
try: try:
db = self.get_db() db = self.get_db()
cur = db.cursor() cur = db.cursor()
statement = 'UPDATE twitter_settings SET since_id = ?' statement = "UPDATE twitter_settings SET since_id = ?"
cur.execute(statement, (since_id,)) cur.execute(statement, (since_id,))
db.commit() db.commit()
db.close() db.close()