A Python Twitter BOT example: Covid-19 BOT Italia

In the previous post (http://notes.massimilianofigini.com/2021/01/how-to-build-and-schedule-python.html) you can find how to create your Twitter developer account and how to schedule a Twitter bot.

Here there is the complete code of my "Covid-19 BOT Italia", a BOT started in April and still active, that automatically retweets all the Covid-19 related tweets by Italian authorities. 
You can see the result here: https://twitter.com/COVIDbot_ITA


You can also find the always updated code on my Github (https://github.com/massyfigini/CovidBOT).

First I have created a CSV file with two fields: Account contains all the Twitter accounts of the Italian authorities (ex. GiuseppeConteIT is the account name of the Italian Prime Minister), LastTweetID contains the ID of the last tweet (numeric, the last part of the link of the tweet).
Then I have created an empty txt file named "Error.txt".
Both files are in the same directory of the Python script.
This is the Python script:

C_KEY = "Replace this sentence with your API Key"  
C_SECRET = "Replace this sentence with your API Key Secret"  
A_TOKEN = "Replace this sentence with your Access Token"  
A_TOKEN_SECRET = "Replace this sentence with your Access Token Secret"  

#pip install tweepy

# import libraries
import tweepy
import pandas as pd

# Authenticate to Twitter
auth = tweepy.OAuthHandler(C_KEY, C_SECRET)
auth.set_access_token(A_TOKEN, A_TOKEN_SECRET)

# Create API object
api = tweepy.API(auth)

# search for covid
text1 = 'coronavirus'
text2 = 'covid'
text3 = 'lockdown'
text4 = 'pandemia'

text = [text1, text2, text3, text4]

# read file
df = pd.read_csv('account.csv')


# for each row of my csv
for index, row in df.iterrows():
    
    # take data
    account = row['Account']
    last_tweet_id = row['LastTweetID']
    
    # tweet limit
    max_tweets = 1000
    
    # take all user tweets
    searched_tweets = [status.id for status in 
                       tweepy.Cursor(api.user_timeline, id = account, since_id = last_tweet_id,
                                     exclude_replies = True).items(max_tweets)]
    
    try:
        # for each tweet
        for idx, i in enumerate(searched_tweets):
            tweet = api.get_status(i, tweet_mode = 'extended')  # all the text
            tweetText = tweet.full_text   # take text
            tweetText = tweetText.lower()   # lowercase
            if idx == 0: df.iloc[index,3] = tweet.id   # last retweeted id
            # filter only covid tweets and exclude retweets
            if not(tweetText.startswith('rt')) and any(x in tweetText for x in text):
                # tweet!    
                api.retweet(i)
             
    except:
        # open a file to append the error
        Err = open("Error.txt""a"encoding="utf-8")
        Err.write(" ACCOUNT: " + account + ". TWEET IN ERROR: " + tweetText)

# rewrite account.csv with last id tweet updated
df.to_csv('account.csv'index=False)

As you can see, for each account in the csv file, the script retweets every tweets that contain some key words starting with the last tweet ID. After that it updates the csv file with the ID of the last tweet. If there is an error, the tweet that generates the problem is wrote in the txt file, so you can easily investigate the problem.


No comments:

Post a Comment