Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def update_status(self, **kwargs):
self._updates.append(kwargs)
r = Status(api=self)
setattr(r, 'id', len(self._updates))
setattr(r, 'author', self.me())
# Status.user is "DEPRECIATED" so we omit it
return r
def on_data(self, data):
if 'in_reply_to_status_id' in data:
status = Status.parse(self.api, json.loads(data))
if self.on_status(status, data) is False:
return False
elif 'delete' in data:
delete = json.loads(data)['delete']['status']
if self.on_delete(delete['id'], delete['user_id']) is False:
return False
elif 'limit' in data:
if self.on_limit(json.loads(data)['limit']['track']) is False:
return False
data = json.loads(raw_data)
if 'in_reply_to_status_id' in data:
status = Status.parse(self.api, data)
if self.on_status(status) is False:
return False
elif 'delete' in data:
delete = data['delete']['status']
if self.on_delete(delete['id'], delete['user_id']) is False:
return False
elif 'event' in data:
status = Status.parse(self.api, data)
if self.on_event(status) is False:
return False
elif 'direct_message' in data:
status = Status.parse(self.api, data)
if self.on_direct_message(status) is False:
return False
elif 'friends' in data:
if self.on_friends(data['friends']) is False:
return False
elif 'limit' in data:
if self.on_limit(data['limit']['track']) is False:
return False
elif 'disconnect' in data:
if self.on_disconnect(data['disconnect']) is False:
return False
elif 'warning' in data:
if self.on_warning(data['warning']) is False:
return False
else:
logging.error("Unknown message type: " + str(raw_data))
def on_data(self, data):
"""Called when raw data is received from connection.
Override this method if you wish to manually handle
the stream data. Return False to stop stream and close connection.
"""
if 'in_reply_to_status_id' in data:
status = Status.parse(self.api, json.loads(data))
if self.on_status(status) is False:
return False
elif 'delete' in data:
delete = json.loads(data)['delete']['status']
if self.on_delete(delete['id'], delete['user_id']) is False:
return False
elif 'limit' in data:
if self.on_limit(json.loads(data)['limit']['track']) is False:
return False
def gen_tuple(jsontweet):
tweet = Status.parse(api, json.loads(jsontweet))
retweeted = (getattr(tweet, 'retweeted_status', None) != None)
return (tweet.author.id, tweet.created_at, convert_to_utf8_str(tweet.text), retweeted)
def any_tweet_to_str(tweet):
""" Status / unicode / str """
if isinstance(tweet, tweepy.models.Status):
tweet_text = tweet_to_text(tweet)
else:
tweet_text = tweet
if isinstance(tweet_text, unicode):
return tweet_text.encode('utf-8')
elif isinstance(tweet, str):
return tweet_text
else:
print 'Unknown tweet type:', type(tweet)
return None