#1

Hi,

Thanks for your support in advance! I am having issues using the code below to search for tweets over a set period of time using a premium account. I am getting a 403 error.

def get_tweets(self, num_tweets, MAX_RESULTS_PER_CALL):
        print("get_tweets_1")
        try:
            config = dict(
                search_tweets_api=dict(
                    account_type='premium',
                    endpoint='https://api.twitter.com/1.1/tweets/search/fullarchive/SearchFullArchive.json',
                    consumer_key=API_KEY,
                    consumer_secret=API_KEY_SECRET
                )
            )
            print("get_tweets_2")
            with open('twitter_keys_fullarchive.yaml', 'w') as config_file:
                yaml.dump(config, config_file, default_flow_style=False)

            premium_search_args = load_credentials("twitter_keys_fullarchive.yaml",
                                                   yaml_key="search_tweets_api",
                                                   env_overwrite=False)
            rule = gen_rule_payload(self.query,
                                    results_per_call=MAX_RESULTS_PER_CALL,
                                    from_date=self.start_date,
                                    to_date=self.until
                                    )
            rs = ResultStream(rule_payload=rule,
                              max_results=num_tweets,
                              **premium_search_args)

I am thinking it is likely an argument in the payload or the endpoint but I am unsure. Any help would be appreciated!

#2

What’s the exact error message from the API? Are you sure you have the App and environment correct in the Dashboard?

Also since ResultStream is a generator you need to iterate over it to make requests, like:

for result in rs:
    print(result)
    ... Etc

But maybe the full code you have has that already, it’s hard to say from the snippet alone.

#3

On the App and environment set up, I have:

  1. Purchased a Full Archive/Premium environment
  2. Set up the Dev environment and connected it to an App
  3. Created a project and connected the App to the project

image
image761×630 18.3 KB

#4

I think there is some confusion here, because purchasing a fullarchive subscription is for the v1.1 API, and setting up a project is for the v2 API, they are not the same and are not compatible. The v1.1 Premium API fullarchive endpoint you purchased is configured on: https://developer.twitter.com/en/account/environments

Any mention of “Project” is referring to the v2 API which is not relevant.

Also make sure you are using the v1.1 search tweets package, searchtweets · PyPI and not the v2 one searchtweets-v2 · PyPI

#5

Hi Igor, thanks for the response.

The error message was
HTTP Error code: 403: {“error”:{“message”:“Forbidden: Authentication succeeded but account is not authorized to access this resource.”,“sent”:“2021-12-23T11:43:01+00:00”,“transactionId”:“344bf251b15e18f9”}}
Request payload: {‘query’: ‘#OvertakeTheFuture’, ‘maxResults’: 500, ‘toDate’: ‘202112070000’, ‘fromDate’: ‘202112050000’}

Here is more of the code which may be more useful for this error message:

class TweetCollector:
    """
    New class to collect tweets for a given query for a given timeframe
    """

    def __init__(self, query, geocode, result_type, until, start_date):
        self.query = query
        self.geocode = geocode
        self.result_type = result_type
        self.until = until
        self.start_date = start_date

        self.auth = TwitterAuthenticator().authenticate_twitter_app()
        self.twitter_client = API(self.auth)

    def find_since_id(self):
        # Not needed with new get_tweets function
        try:
            twitter_client = TwitterClient() 
            api = twitter_client.get_twitter_client_api() 
            first = api.search("a", count=1, until=self.start_date)
            self.since_id = first[0].id

        except Exception as err:
            return print(err)

    def get_tweets(self, num_tweets):
        print("get_tweets_1")
        try:
            config = dict(
                search_tweets_api=dict(
                    account_type='premium',
                    endpoint='https://api.twitter.com/1.1/tweets/search/fullarchive/SearchFullArchive.json',
                    consumer_key=API_KEY,
                    consumer_secret=API_KEY_SECRET
                )
            )
            print("get_tweets_2")
            with open('twitter_keys_fullarchive.yaml', 'w') as config_file:
                yaml.dump(config, config_file, default_flow_style=False)

            premium_search_args = load_credentials("twitter_keys_fullarchive.yaml",
                                                   yaml_key="search_tweets_api",
                                                   env_overwrite=False)
            rule = gen_rule_payload(self.query,
                                    results_per_call=500,
                                    from_date=self.start_date,
                                    to_date=self.until
                                    )
            rs = ResultStream(rule_payload=rule,
                              max_results=num_tweets,
                              **premium_search_args)
            print("get_tweets_3")
            tweets = []
            n = 0
            for tweet in rs.stream():
                n += 1
                if n % 100 == 0:
                    print('{0}: {1}'.format(str(n), tweet['created_at']))
                tweets.append(tweet)
            print('done, total tweets: ' + str(n))

Appreciate your help!

#6

This error suggests that your premium API environment is not configured correctly, in https://developer.twitter.com/en/account/environments SearchFullArchive has to be the name of the environment created for the fullarchive endpoint