Python twython 模块，TwythonError() 实例源码

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

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