def follow_user(screen_name):
follows = get_list('follows.txt')
add = True
for stored_screen_name in follows:
if screen_name == stored_screen_name:
add == False
if add == True:
unfollow_add = unfollow_fifo()
try:
twitter.create_friendship(screen_name=screen_name)
add_line(screen_name, 'follows.txt')
print 'followed ' + screen_name.encode('ascii', 'ignore')
except TwythonError as e:
print 'FAILED to follow ' + screen_name.encode('ascii', 'ignore')
return (unfollow_add + 1)
else:
return unfollow_add
python类TwythonError()的实例源码
def get_tweets(counter):
blacklist = get_blacklist()
keywords = "rt2win OR retweet to win OR Gewinnspiel OR Verlosung -filter:retweets AND -filter:replies"
search_results = twitter.search(q=keywords, result_type= "recent", count=counter)
tweet_ids = []
try:
for tweet in search_results["statuses"]:
add_tweet = True
for name in blacklist:
tweeter = tweet["user"]["screen_name"]
if (name.rstrip('\n') == tweeter):
add_tweet = False
if (add_tweet == True):
tweet_ids.append(str(tweet["id_str"]))
except TwythonError as e:
print e
return tweet_ids
#retweet tweets from task-list, follow users and log actions
def tweetMessage(tweet, tweettype, media):
retry = 10
while retry:
try:
retry -= 1
time.sleep(1)
if tweettype == 'stats':
api.update_status(status=tweet, media_ids=[media['media_id']])
else:
api.update_status(status=tweet)
logPrint("Successfully tweeted:\n" + tweet)
return
except TwythonError as te:
logPrint("Twitter error: \n" + str(te) + "\n")
break
except IOError as e:
logPrint("Unable to tweet: \n" + str(e) + "\nRetrying...")
continue
except OpenSSL.SSL.SysCallError as e:
logPrint("Unable to tweet: \n" + str(e) + "\nRetrying...")
continue
logPrint("ERROR: Tweet failed!\n")
#Coffee Statistics
def get_trump(self, ctx):
"""Get Trump's latest Yuge success!"""
if common.twitter is None:
await ctx.send("Twitter not activated.")
return
twitter_id = 'realdonaldtrump'
tweet_text = \
':pen_ballpoint::monkey: Trump has been saying things, as ' \
'usual...'
rt_text = \
':pen_ballpoint::monkey: Trump has been repeating things, as ' \
'usual... (RT ALERT)'
try:
await get_last_tweet(twitter_id, tweet_text, rt_text, ctx)
except TwythonError:
await self.bot.say("Twitter is acting up, try again later.")
if common.trump_chance_roll_rdy:
await item_chance_roll(ctx.bot, ctx.message.author.display_name,
ctx.message.channel)
common.trump_chance_roll_rdy = False
def get_news(self, ctx):
"""Grab a news story"""
if common.twitter is None:
return
shuffle(common.news_handles)
found_art = False
while not found_art:
source = common.news_handles.pop(0)
common.news_handles.append(source)
tweet_text = "It looks like @" + source + " is reporting:"
rt_text = "It looks like @" + source + " is retweeting:"
try:
await get_last_tweet(source, tweet_text, rt_text, ctx)
except TwythonError:
print("Error in get_news, trying another source")
else:
found_art = True
return
def process_one_waiting_status(self, result_log, result_summary, media_ios, out_cn,
is_dry_run):
result_summary["result"] = "Start"
result_summary.update(result_log.get_processed_at_dict())
[result_summary_eid] = result_log.update_result_summary_in_db(
result_summary=result_summary, eids=[result_summary.eid])
out_status = None
try:
out_cn.connect()
if not is_dry_run:
media_ids = out_cn.upload_medias(media_ios)\
if self.config.outbound.getboolean("attach_media", "yes")\
else []
out_status = out_cn.update_status(
result_summary["status_string"], media_ids=media_ids)
result_summary["result"] = "Succeed"
else:
result_summary["result"] = "Test"
except TwythonError as twython_ex:
result_summary["result"] = "Failed"
result_summary.result_string = "{0}".format(str(twython_ex))
finally:
pass
if not is_dry_run:
if out_status:
out_summary = result_log.make_status_summary(
"outbound", out_status)
result_summary.update(out_summary)
[updated_eid] = result_log.update_result_summary_in_db(
result_summary=result_summary, eids=[result_summary_eid])
if result_summary_eid is not updated_eid:
self.logger.debug("Failed to update log database.")
return True
self.logger.info(self.message_to_logger(result_summary=result_summary))
return False
def twitter_follow(screen_name):
twitter = get_twitter_instance()
try:
twitter.create_friendship(screen_name = screen_name)
except TwythonError as err:
print("Had this error, bud: " + str(err))
def retweet(tweet):
try:
twitter.retweet(id=tweet['id'])
print 'retweeted: [tweet_id:' + str(tweet['id']) + ']'
date_time = (time.strftime("%d-%m-%Y") + " | " + time.strftime("%H:%M:%S"))
log_entry = str(date_time) + " | RT | " + str(tweet['id']) + " | " + tweet['user']['screen_name']
add_line(log_entry, 'log.txt')
except TwythonError as e:
print 'FAILED to retweet: [tweet_id:' + str(tweet['id']) + ']'
def like_this(tweet):
try:
twitter.create_favorite(id=tweet['id'])
print 'liked tweet: [tweet_id:' + str(tweet['id']) + ']'
except TwythonError as e:
print 'FAILED to like tweet: [tweet_id:' + str(tweet['id']) + ']'
def do_retweets_to(retweet_list, counter, iter):
for tweet_ids in retweet_list:
try:
tweet = twitter.show_status(id = tweet_ids)
user_id = tweet["user"]["id_str"]
screenname = tweet["user"]["screen_name"]
twitter.retweet(id = tweet_ids)
try:
twitter.create_friendship(user_id = user_id)
add_line(screenname, "follows.txt")
except TwythonError as e:
print("Follow-Error: ")
print e
datum_zeit = (time.strftime("%d-%m-%Y") + " | " + time.strftime("%H:%M:%S"))
log_entry = str(datum_zeit) + " | RT | " + tweet_ids + " | " + screenname
add_line(log_entry, "log.txt")
wait_for_it()
except TwythonError as e:
print ("Retweet-Error: " + tweet["user"]["screen_name"])
print e
wait_for_it()
retweet_action(counter, iter, seconds())
#get list of users we don't like (mainly botspotters)
def retweet(tweet):
try:
twitter.retweet(id=tweet['id'])
date_time = (time.strftime("%d-%m-%Y") + " | " + time.strftime("%H:%M:%S"))
log_entry = str(date_time) + " | RT | " + str(tweet['id']) + " | " + tweet['user']['screen_name']
add_line(log_entry, 'log.txt')
except TwythonError as e:
a = '';
def get_user_timeline(screen_name, count=200):
"""Return list of most recent tweets posted by screen_name."""
# ensure count is valid
if count < 1 or count > 200:
raise RuntimeError("invalid count")
# ensure environment variables are set
if not os.environ.get("API_KEY"):
raise RuntimeError("API_KEY not set")
if not os.environ.get("API_SECRET"):
raise RuntimeError("API_SECRET not set")
# get screen_name's most recent tweets
# https://dev.twitter.com/rest/reference/get/users/lookup
# https://dev.twitter.com/rest/reference/get/statuses/user_timeline
# https://github.com/ryanmcgrath/twython/blob/master/twython/endpoints.py
try:
twitter = Twython(os.environ.get("API_KEY"), os.environ.get("API_SECRET"))
user = twitter.lookup_user(screen_name=screen_name)
if user[0]["protected"]:
return None
tweets = twitter.get_user_timeline(screen_name=screen_name, count=count)
return [html.unescape(tweet["text"].replace("\n", " ")) for tweet in tweets]
except TwythonAuthError as e:
raise RuntimeError("invalid API_KEY and/or API_SECRET") from None
except TwythonRateLimitError:
raise RuntimeError("you've hit a rate limit") from None
except TwythonError:
return None
def addFollowers():
try:
followers = api.get_followers_ids(screen_name=username)
for i in followers['ids']:
follower = '@' + api.show_user(user_id=i)["screen_name"]
if follower not in handles:
handles.append(follower)
logPrint("Added:" + follower + " to handles.")
logPrint("addfollowers() finished")
except TwythonError as e:
logPrint("Twy: Unable to add followers:\n" + str(e))
return
except OpenSSL.SSL.SysCallError as e:
logPrint("SSL: Unable to add followers:\n" + str(e))
return
def get_last_tweet(_id, tweet_text, rt_text, ctx):
"""
Gets the last tweet for id.
:param _id: Twitter id
:param tweet_text: flavor text for tweets
:param rt_text: flavor text for retweets
:param ctx: Context
:return:
"""
if common.twitter is None:
await ctx.bot.say("Twitter not activated.")
return
try:
last_tweet = common.twitter.get_user_timeline(screen_name=_id, count=1,
include_retweets=True)
except TwythonError as e:
raise e
else:
# if it's a retweet, send the original tweet
if 'retweeted_status' in last_tweet[0]:
if _id == 'realdonaldtrump'and \
common.last_id != last_tweet[0]['id']:
await item_chance_roll(ctx.bot, ctx.message.author.display_name,
ctx.message.channel)
common.last_id = last_tweet[0]['id']
rt_id = last_tweet[0]['retweeted_status']['id']
rt_screen_name = last_tweet[0]['retweeted_status']['user'][
'screen_name']
await ctx.bot.say('{}\n\nhttps://twitter.com/{}/status/{}'
.format(rt_text, rt_screen_name, str(rt_id)))
# otherwise, send the tweet
else:
if _id == 'realdonaldtrump':
common.last_id = last_tweet[0]['id']
await ctx.bot.say('{}\n\nhttps://twitter.com/{}/status/{}'
.format(tweet_text, last_tweet[0]['user']
['screen_name'],
str(last_tweet[0]['id'])))
def twitter(query):
ckey = os.environ['consumer_key']
csecret = os.environ['consumer_secret']
atoken = os.environ['access_token']
asecret = os.environ['access_secret']
twitter = Twython(ckey, csecret, atoken, asecret)
try:
search_results = twitter.search(q=query, languages = ['pt'] ,count=100)
except TwythonError as e:
print (e)
reviews = []
tweets = []
for tweet in search_results['statuses']:
if tweet['lang'].encode("utf-8") == 'pt':
tweets.append(tweet['text'])
if tweets == []:
return [], [], []
sents = sentiment(tweets)
both = zip(tweets,sents)
overall_sentiment = []
count_pos = 0
count_neutral = 0
count_neg = 0
for i in range(len(both)):
sent_dict = {}
sent_dict['tweet'] = both[i][0]
sent_dict['sentimento'] = both[i][1]
if sent_dict['sentimento'] == 0:
sent_dict['sentimento'] = "negative"
overall_sentiment.append(-1.0)
count_neg = count_neg + 1
elif sent_dict['sentimento'] == 1:
sent_dict['sentimento'] = "neutral"
overall_sentiment.append(0.0)
count_neutral = count_neutral + 1
elif sent_dict['sentimento'] == 2:
sent_dict['sentimento'] = "positive"
overall_sentiment.append(1.0)
count_pos = count_pos + 1
reviews.append(sent_dict)
overall_sentiment = sum(overall_sentiment)/len(overall_sentiment)
data = [count_neg, count_neutral, count_pos]
return reviews, overall_sentiment, data
