sdg-ai-twitter/main.py

import tweepy
from tweepy import OAuthHandler
import logging
import csv


API = {
    'consumer_key': 'dYgo8B2sDLruMrVXR0Kd7Tp5N',
    'consumer_secret': '5Yb6v2v4Y4zulq8fXL5iRuuTK4G0df48AoQI36B7XTm0cw1sIT',
    'access_token': '1204735114581827585-OLFpb3m4IHKmRlZ0kQM1j5LMRhNCyB',
    'access_token_secret': 'JXQcqSP15ihwjYDzyub9sWJFRwIjuUolhlSOlj5IlhjkP'
}


class TwitterClient:

    def __init__(self):
        '''API connector'''
        self._consumer_key = API['consumer_key']
        self._consumer_secret = API['consumer_secret']
        self._access_token = API['access_token']
        self._access_token_secret = API['access_token_secret']

        try:
            '''API Authentication'''
            self.auth = OAuthHandler(self._consumer_key, self._consumer_secret)
            self.auth.set_access_token(self._access_token, self._access_token_secret)
            self.api = tweepy.API(self.auth, wait_on_rate_limit=True)
        except Exception:
            logging.error('Authentication Failed')

    def get_tweets(self, hashtag):

        with open('./data/tweets.csv', 'a') as csv_file:
            csv_writer = csv.writer(csv_file)
            csv_writer.writerow(["Date", "Text"])
            for tweet in tweepy.Cursor(self.api.search, q=hashtag, count=1, lang='es', since="2021-01-01").items(1):
                print(tweet)
                csv_writer.writerow([tweet.created_at, tweet.text.encode('utf-8')])
        
if __name__ == '__main__': 
    twitter_client = TwitterClient()
    twitter_client.get_tweets("%23TurismoCity")