app.py

import os
import string
import random

# import tweepy
import time
import json
import requests
from requests_oauthlib import OAuth1

import pandas as pd
from datasets import load_dataset

BEARER_TOKEN = os.environ.get("BEARER_TOKEN")
CONSUMER_KEY = os.environ.get("CONSUMER_KEY")
CONSUMER_SECRET = os.environ.get("CONSUMER_SECRET")
ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN")
ACCESS_TOKEN_SECRET = os.environ.get("ACCESS_TOKEN_SECRET")

# client = tweepy.Client(
#         bearer_token=BEARER_TOKEN,
#         consumer_key=CONSUMER_KEY,
#         consumer_secret=CONSUMER_SECRET,
#         access_token=ACCESS_TOKEN,
#         access_token_secret=ACCESS_TOKEN_SECRET,
# )


# Real tweets

real_dataset = load_dataset("dawood/elon-tweets", split="train", cache_dir="dataset_cache")
real_df = real_dataset.to_pandas()
real_tweets = real_df["Tweets"][:500].values.tolist()


def load_real_tweet():
    index = random.randint(0, len(real_tweets))
    return real_tweets[index]


# Fake tweets
fake_df = pd.read_csv("elon_generated_tweets.csv")
fake_tweets = fake_df.values.tolist()

def load_fake_tweet():
    index = random.randint(0, len(fake_tweets))
    return fake_tweets[index]

# response = client.create_tweet(
#     text=load_real_tweet()
# )


def connect_to_endpoint(url, params=None, data=None, type='GET'):
    resp = None
    if type == 'GET':
        resp = requests.get(url, json=data, params=params, auth=bearer_oauth)
    elif type == 'POST':
        resp = requests.post(url, json=data, params=params, auth=OAuth1(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET))
    elif type == 'DELETE':
        resp = requests.delete(url, json=data, params=params, auth= OAuth1(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET))

    if not resp.status_code in (200, 201):
        raise Exception(resp.status_code, resp.text)
    return resp.json()



def create_poll():
    ''' Creates a Twitter poll using the given question and options '''
    url = 'https://api.twitter.com/2/tweets'
    real_or_fake = random.randint(0, 2)
    function = load_real_tweet if real_or_fake > 0 else load_fake_tweet
    tweet = function()
    data = {
        'text': tweet,
        'poll': {
            'options': [
                "Elon",
                "Not",
            ],
            'duration_minutes': 24 * 60
        }
    }
    connect_to_endpoint(url, data=data, type='POST')



if __name__ == "__main__":
    while True:
        create_poll()
        time.sleep(24 * 60 * 60)