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)