python类TweepError()的实例源码

def search(writer, query, since_id=0):
    """Get hydrated Tweet-objects using the Search API."""
    LOGGER.info("search() starting")

    # initialize config and Twitter API
    config = read_config()
    api = get_app_auth_api(config)

    # process the query, storing returned Tweets in JSON format
    num_tweets = 0
    search_params = {
        "q": query,
        "count": SEARCH_COUNT,
        "result_type": "recent",
    }
    if since_id > 0:
        search_params.update({"since_id": since_id})
    limit = config.getint("search", "limit")
    try:
        num_tweets = write_objs(writer, api.search, search_params,
                                cursored=True, limit=limit)
        LOGGER.info("downloaded %d Tweet(s)", num_tweets)
    except TweepError as err:
        log_tweep_error(LOGGER, err)

    # finished
    LOGGER.info("search() finished")

def handle(self, *args, **options):
        api_keys = settings.TWITTER_USER_KEYS
        auth = tweepy.OAuthHandler(api_keys['consumer_key'], api_keys['consumer_secret'])
        auth.set_access_token(api_keys['access_token_key'], api_keys['access_token_secret'])

        new_portraits = Portrait.objects.filter(active=True, demo_portrait=False)

        print('scheduled portraits', new_portraits.count())

        for portrait in new_portraits:
            is_new_portrait = portrait.portrait_content is None

            print('user', portrait.auth_screen_name)
            print('new', is_new_portrait)

            try:
                portrait_api(portrait)
                print('OK')
            except TweepError as err:
                print('ERROR', err)
                portrait.active = False
                portrait.save()
                continue
            except Exception as err:
                print('ERROR', err)
                continue

            # to avoid too many connections
            time.sleep(5)

def handle(self, *args, **options):
        api_keys = settings.TWITTER_USER_KEYS
        auth = tweepy.OAuthHandler(api_keys['consumer_key'], api_keys['consumer_secret'])

        try:
            redirect_url = auth.get_authorization_url()
        except tweepy.TweepError as e:
            print('Error! Failed to get request token.')
            raise e

        print(redirect_url)
        verifier = input('Verifier:')

        print(auth.get_access_token(verifier=verifier))

def random_tweet():
    # build a twitter list of five possible tweets
    tweet_list = [
    'Charm was a scheme for making strangers like and trust a person immediately, no matter what the charmer had in mind.',
    "There is no good reason good can't triumph over evil, if only angels will get organized along the lines of the mafia.",
    "Shrapnel was invented by an Englishman of the same name. Don't you wish you could have something named after you?",
    "If your brains were dynamite there wouldn't be enough to blow your hat off.",
    "And so on."
    ]
    x = 0
    # create a loop to tweet twice
    while x < 2:
        try:
            # generate a random variable between 0 and 4
            rand = random.randint(0, 4)
            # use that random variable to grab an item from tweet_list
            api.update_status(tweet_list[rand])
            # take a 30-second break so we don't overwhelm our followers
            sleep(30)
            #increment our counter
            x += 1
        # if there's an error, catch it instead of crashing and print out the error message
        except tweepy.TweepError, e:
            print 'Error sending tweet:', e[0][0]['message']

# a function to tweet a gif

def gif_tweet():
    try:
        # takes a file location and a text string to update status
        api.update_with_media('../img/pulp-fiction-search.gif','BEHOLD, a gif')
    # if there's an error, catch it instead of crashing and print out the error message
    except tweepy.TweepError, e:
            print 'Error sending tweet:', e[0][0]['message']

# utility function for retrieving secret keys
# from our secret key file

def on_follow(self, f_id):
        """
        Follow back on being followed
        """
        try:
            self.api.create_friendship(f_id, follow=True)
            self.friends.append(f_id)
            logging.info('Followed user id {}'.format(f_id))
        except tweepy.TweepError as e:
            self._log_tweepy_error('Unable to follow user', e)

def _follow_all(self):
        """
        follows all followers on initialization
        """
        logging.info("Following back all followers....")
        try:
            self.followers['new'] = list(set(self.followers['existing']) - set(self.friends))
            self._handle_followers()
        except tweepy.TweepError as e:
            self._log_tweepy_error('Can\'t follow back existing followers', e)

def _check_followers(self):
        """
        Checks followers.
        """
        logging.info("Checking for new followers...")

        try:
            self.followers['new'] = [f_id for f_id in self.api.followers_ids(self.id) if f_id not in self.followers['existing']]
            self.state['last_follow_check'] = time.time()

        except tweepy.TweepError as e:
            self._log_tweepy_error('Can\'t update followers', e)

def twitter_bot():
    tf_config = tf.ConfigProto(device_count = {"GPU":0}, log_device_placement = True)

    auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
    auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
    api = tweepy.API(auth)

    with tf.Session(config = tf_config) as sess:
        predictor = predict.EasyPredictor(sess)

        for tweet in tweets():
            status_id, status, bot_flag = tweet
            print("Processing {0}...".format(status.text))
            screen_name = status.author.screen_name
            replies = predictor.predict(status.text)
            if not replies:
                print("no reply")
                continue

            reply_body = replies[0]
            if reply_body is None:
                print("No reply predicted")
            else:
                try:
                    if is_contain(status.text, '??????'):
                        special_reply(api, bot_flag, screen_name, status_id, code = 1)
                    elif is_contain(status.text, '????'):
                        special_reply(api, bot_flag, screen_name, status_id, code = 2)

                    elif is_contain(status.text, '?????'):
                        special_reply(api, bot_flag, screen_name, status_id, code = 3)

                    else:
                        post_reply(api, bot_flag, reply_body, screen_name, status_id)
                except tweepy.TweepError as e:
                    if e.api_code == 187:
                        pass
                    else:
                        raise
            mark_tweet_processed(status_id)

def get_tweets(user_id, api):
    cursor = tweepy.Cursor(api.user_timeline, user_id).pages()
    while True:
        try:
            tweets = [page for page in cursor]

        except tweepy.TweepError as e:
            tweets = []
            api_codes = [401, 404, 500]
            if not str(e): break
            if(int(filter(str.isdigit, str(e))) in api_codes): break
            print('get_tweets: ' + str(e))

    return tweets

def user_status_count(user_id, api):
    try: 
        user = api.get_user(user_id=user_id)
        if(user.statuses_count):
            count = user.statuses_count

    except tweepy.TweepError as e:
        count = 0
        #print(e.message[0]['message'])

    finally:
        return count

def verify_working_credentials(api):
    verified = True
    try:
        api.verify_credentials()
    except tweepy.TweepError as e:
        verified = False
    finally:
        return verified

def send_token():
    global auth
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret, callback_url)

    try:
        redirect_url = auth.get_authorization_url()
        session['request_token'] = auth.request_token
    except tweepy.TweepError:
        print ('Error! Failed to get request token.')

    return redirect(redirect_url)

# callback. once twitter authorizes user, it sends user back to this page

def get_verification():
    global auth
    #get the verifier key from the request url
    verifier = request.args['oauth_verifier']

    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    token = session['request_token']
    del session['request_token']

    auth.request_token = token

    try:
        auth.get_access_token(verifier)
    except tweepy.TweepError:
        print 'Error! Failed to get access token.'

    api = tweepy.API(auth)
    # cache info to avoid rate limit
    access_info['api'] = api
    access_info['following'] = api.friends_ids(api.me().screen_name)
    access_info['followers'] = api.followers_ids(api.me().screen_name)
    access_info["access_token"] = auth.access_token
    access_info["access_token_secret"] = auth.access_token_secret


    return redirect(url_for('twitter_DL'))

def troll_bot_analyzer(user, api):
    try:
        user_data = data_user(user, api)
    except tweepy.TweepError:
        logging.error("This user is protected or does not exist. His  information cannot be accessed")
    else:
        if len(user_data["tweets"]) == 0:
            logging.error("There is not enough information to classify this user")
            return False
        hashtags_per_tweet = float(user_data["number_hashtags"]) / len(user_data["tweets"])
        mentions_per_tweet = float(user_data["number_mentions"]) / len(user_data["tweets"])
        percentage_tweet_with_mention = float(user_data["tweet_with_mentions"]) / len(user_data["tweets"])
        percentage_tweet_with_hashtag = float(user_data["tweets_with_hashtags"]) / len(user_data["tweets"])
        signs_per_char, capitals_per_char, activity, percentage_tweet_with_omg = drama_queen(user_data)
        periodicity, answer = periodicity_answer(user_data)
        diversity_hashtags = tweet_iteration_hashtags(user_data)
        diversity_tweets = tweet_iteration_stemming(user_data)
        urls_percentage = tweet_iteration_urls(user_data)
        num_stalker, who_stalker = stalker(user_data)
        per_drama_queen = percentage_drama_queen(activity, percentage_tweet_with_omg, capitals_per_char, signs_per_char,
                                                 percentage_tweet_with_hashtag, percentage_tweet_with_mention,
                                                 mentions_per_tweet, hashtags_per_tweet)
        per_bot = percentage_bot(periodicity, answer, diversity_tweets)
        per_stalker, famous, non_famous = percentage_stalker(num_stalker, who_stalker, mentions_per_tweet, percentage_tweet_with_mention, api)
        if per_stalker == 0:
            per_stalker = num_stalker
        per_spammer = percentage_spammer(diversity_tweets, diversity_hashtags, urls_percentage)
        per_hater = (1 - sentiment(user_data)) * 100

        max_value = [per_bot, per_drama_queen, per_stalker, per_hater, per_spammer]
        index = max_value.index(max(max_value))
        labels = ["bot", "drama_queen", "stalker", "hater", "spammer"]
        final = labels[index]

        return {"user_id": user, "bot": per_bot, "drama_queen": per_drama_queen, "stalker": per_stalker, "hater": per_hater, "spammer": per_spammer, "famous": famous, "non_famous": non_famous, "stalked": who_stalker, "final": final}

def get_user(self, uid=None, name=None):
        if not self._api:
            raise tweepy.TweepError('Api NOT inited!')
        try:
            user = self._api.get_user(user_id=uid, screen_name=name)
            self._user = user
        except tweepy.TweepError as e:
            logging.error('Uid ({0}) and name ({1}) has error: {2}'.format(uid, name, e))
            if e.api_code in self._IGNORE_ERROR_CODES:
                return None
            raise e
        return self

def authentication(method):
        def judge(self, *args, **kwargs):
            if not self._api:
                raise tweepy.TweepError('Api NOT inited!')
            if not self._user:
                raise tweepy.TweepError('User NOT inited!')
            method(self, *args, **kwargs)
            return self
        return judge

def get_user_init():
    global status
    global path
    global rev
    try:
        path = 'D:/Twitter/Depth-%d/%s' % (depth_num,str(user_id))
        try:
            os.makedirs(path)
        except FileExistsError:
            print("Folder already exist")
            status="skip"
            return

        data = api.get_user(user_id)
        if data['protected']==False:
            get_user(data)
            rev=1
        else:
            status="skip"
            print("Protected")

    except tweepy.RateLimitError:
        countdown(960)
        get_user_init()

    except tweepy.TweepError as e:
            if tweepy.TweepError is "[{'message': 'Over capacity', 'code': 130}]" or e[0]['code'] is 130:
                countdown_te(600,e)
                get_user_init()
            elif tweepy.TweepError is "[{'message': 'User not found.', 'code':50}]" or e[0]['code'] is 50:
                status="skip"
                return
            else:
                print(e)

def get_id(sn):
    try:
        return(api.get_user(screen_name=sn)['id'])
    except tweepy.RateLimitError:
        countdown(960)
        get_id(sn)

    except tweepy.TweepError as e:
            if tweepy.TweepError is "[{u'message': u'Over capacity', u'code': 130}]" or e is "[{u'message': u'Over capacity', u'code': 130}]":
                countdown_te(600,e)
                get_id(sn)
            else:
                print(e)

def limit_handler(cursor):
    while True:
        try:
            yield cursor.next()
        except tweepy.RateLimitError:
            countdown(960)
        except tweepy.TweepError as e:
            if tweepy.TweepError is "[{u'message': u'Over capacity', u'code': 130}]" or e is "[{u'message': u'Over capacity', u'code': 130}]":
                countdown_te(600,e)
            else:
                print(e)