# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Functionality to read tweets

# +
# project_dir = os.path.join(os.path.dirname(__file__), os.pardir)
# dotenv_path = os.path.join(project_dir, '.env')
# print(dotenv_path)
# print(os.path.pardir)
# print(os.pardir)
# print(os.path.dirname)

# +
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
from pykafka import KafkaClient
import dotenv
import json
import os

def get_kafka_client():
    return KafkaClient(hosts='localhost:9092')


class TwitterAuthenticator():
    """
    Class for authenticating twitter client api.
    """
    def authenticate_twitter_app(self):
        cred = dotenv.dotenv_values()
        auth = OAuthHandler(cred.get('TWI_CONSUMER_KEY'), cred.get('TWI_CONSUMER_SECRET'))
        auth.set_access_token(cred.get('TWI_ACCESS_TOKEN'), cred.get('TWI_ACCESS_TOKEN_SECRET'))
        return auth


class TwitterStreamer():
    """
    Class for streaming and processing live tweets.
    """
    def __init__(self):
        self.twitter_autenticator = TwitterAuthenticator()    

    def stream_tweets(self, hash_tag_list):
        # This handles Twitter authetification and the connection to Twitter Streaming API
        listener = TwitterListener()
        auth = self.twitter_autenticator.authenticate_twitter_app() 
        stream = Stream(auth, listener)

        # This line filter Twitter Streams to capture data by the keywords: 
        stream.filter(track=hash_tag_list, languages=["en"])


class TwitterListener(StreamListener):
    """
    This is a basic listener that just prints received tweets to stdout.
    """
    def on_data(self, raw_data):
        client = get_kafka_client()
        topic = client.topics['twitterdata']
        producer = topic.get_sync_producer()
        producer.produce(raw_data.encode('utf-8'))
        message = json.loads(raw_data)
        print(message)
        return True

    def on_error(self, status_code):
        print(status_code)


if __name__ == '__main__':
    hash_tag_list = ['covid']
    stream = TwitterStreamer()
    tweets = stream.stream_tweets(hash_tag_list)

StarcoderdataJupyterScriptsDedupFiltered