Create a Twitter bot in python — part 2

Photo by Rock'n Roll Monkey on Unsplash

In this second part, we will see how to send an email with collected tweets and retweet relevant tweets instantly. If you haven’t read the first part, you can find it here.

Weekly sending of search results by email

We will first see how to send an email with an attachment of the different tweets resulting from the search for positive messages about the movie Wakanda Forever every Monday at 9 am, resulting from past week.

We will use the following python libraries:

• apscheduler: to schedule repetitive tasks.
• emails: to send emails.

Here is the resulting code that we will comment on next:

Periodic emailing with tweets in attachment

Notes:

• There are some environment variables to configure before sending the email. Email login and password are only necessary if using TLS, which will be the case most of the time.
• To test email sending locally, you can use the mailhog service.
• Lines 34 to 47, to store the tweets in the file, I use the JSON lines format. It is convenient to save a lot of data without eating up all the memory.
• In lines 49 to 60, we define email information (you can modify them as you wish), the attachment, and the recipient. I print an error message when the email is not sent but you can configure logging instead. By the way, to try to debug the error, you should configure logging to display messages from the emails library at debug level.
• Line 66, we define the scheduler, and we use the Blocking version which is appropriate to our use case. But depending on the type of project you are building, you could use the Threading or Asyncio version. I let you look at the documentation for more details.
• Line 70, we schedule our job using the crontab syntax. If you want to check the syntax of your crontab before applying it, you can use this website.
• Line 74, our scheduler will run forever.

You can host your application for free on platforms like fly.io.

Retweet Wakanda tribute

For the second application, we want to retweet instantly tweets paying tribute to our favorite movie. Unfortunately, we can’t use the bearer token here anymore. We need an access token with specific rights. If we look at the endpoint documentation, we see that we need the permissions tweet:read, tweet:write and users:read. The workflow to gain an access token is defined here. Nevertheless, I will show you how to do it with tweepy. You will need a redirect url. This url will be used to send a code necessary to gain the access token. On this url, two pieces of information will be passed as query parameters:

• state: a random security string. More information about oauth2 vocabulary can be found here.
• code: a security value returned by the Twitter API.

Here is an example of a FastAPI server to set up quickly a redirect url.

import logging
from fastapi import FastAPI, Response

app = FastAPI()
logger = logging.getLogger(__name__)

@app.get('/')
def get_code(code: str, state: str):
    logger.info('code: %s, state: %s', code, state)
    return Response(status_code=200)

Once you have a redirect url set up, you need to go back to your developer portal, on the detail page of your project, you have a section User authentication settings, click here. You can pass the first section App permissions which concerns oauth1 that we don’t use. In the second section Type of App, choose Web app, Automated App or bot. In the section App info, fill in the redirection url corresponding to your server. You will also have to fill in a personal url (you can put anything as long as it is linked to you) and other optional information if you want. Once this is done, you will get a client_id and a client_secret needed for oauth2 authentication. Keep them in a safe place. We will use in the next scripts the CLIENT_ID and CLIENT_SECRET environment variables which must contain these values. As for the workflow to obtain an access token with tweepy, it is as follows:

1 — Use the tweepy.OAuth2UserHandler class by filling in all the necessary information.

import os
import tweepy

oauth2_user_handler = tweepy.OAuth2UserHandler(
    client_id=os.getenv('CLIENT_ID'),
    client_secret=os.getenv('CLIENT_SECRET'),
    redirect_uri='votre url de redirection ici',
    # we list the permissions we want here
    scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'],
)
# it will generate the authorization url that we must launch in our web browser.
print(oauth2_user_handler.get_authorization_url())

You will notice for the scope argument that I added the offline.access permission. It will help to refresh the access token without manual intervention, I will explain it later. Once you have the authorization url, copy it into the address bar of your browser and follow it (click on Enter). You will have to authorize your bot to have access to your account, once this is done you will be redirected to the url that you have defined as your redirection url.

2 — Copy the redirection url in the browser that should contain the code assigned to you, and use a method of OAuth2UserHandler that will retrieve the access token.

access_token = oauth2_user_handler.fetch_token(
    'redirection url here'
)

3 — Once this is done you can use the tweepy client as usual, except that instead of the bearer token, the access token will be used.

client = tweepy.Client('access token')

Now you can create and retweet tweets as you wish:

import os
import tweepy

client = tweepy.Client(os.getenv('ACCESS_TOKEN'))
tweet_response = client.create_tweet(text='Hello from bot!', user_auth=False)
response = client.retweet(tweet_response.data['id'], user_auth=False)
print(response)

The argument user_auth=False is important otherwise tweepy will try an oauth1 authentication and the request will fail. It's a bit strange as an API, but it's the legacy of the old API. For your information, an access token is valid for two hours. And now some of you must be wondering if they will have to repeat the operation manually with the browser... That doesn't sound very automated. Remember the offline.access permission we used at the beginning to get the authorization url? It will be used to refresh our token before it expires. In fact, when we retrieved the access token, our permission also allowed us to retrieve a refresh token. It is this last token that is used for the refresh. It is kept internally by tweepy. To refresh the access token with tweepy, here is how we proceed:

import tweepy

token_info = oauth2_user_handler.refresh_token(
    'https://api.twitter.com/2/oauth2/token'
)
client = tweepy.Client(token['access_token'])

We have to pass the url to refresh a token. If you wonder where I found this url, it is available on this page under the Step 5… section. If you have a refresh_token that you got in another way than by tweepy, you can pass it with the refresh_token argument. In token_info we have a set of information including the access token and the new refresh token (again saved by tweepy for later use). So we can instantiate a new client with the new access token without having to make a manual intervention. And to automate everything, we can use apscheduler that we know well now. An example of code that you can write:

import os

import tweepy
from apscheduler.schedulers.blocking import BlockingScheduler


scheduler = BlockingScheduler()
oauth2_user_handler = tweepy.OAuth2UserHandler(
    client_id=os.getenv('CLIENT_ID'),
    client_secret=os.getenv('CLIENT_SECRET'),
    redirect_uri='your redirection url here',
    scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'],
)

def refresh_token():
    token_data = oauth2_user_handler.refresh_token(
        'https://api.twitter.com/2/oauth2/token'
    )
    # save the token where you want
    os.environ['ACCESS_TOKEN'] = token_data['access_token']

# think to add this job the first time you get an access token
scheduler.add_job(refresh_token, 'interval', hours=1, minutes=55)
scheduler.start()

Here we use the interval trigger to refresh our token every one hour and fifty-five minutes since the token only lasts two hours. 😁

Finally, here is the code to automatically retweet every tweet paying tribute to the movie Wakanda Forever.

import os

import tweepy


client = tweepy.Client(os.getenv('ACCESS_TOKEN'))


class WakandaTributeRetweet(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        print(tweet.id, tweet.text)
        client.retweet(tweet.id, user_auth=False)
    def on_errors(self, errors):
        print(errors)
    def on_connection_error(self):
        self.disconnect()

bot = WakandaTributeRetweet(os.getenv('BEARER_TOKEN'))
bot.filter()

Note: You need to still have the rules we created in the first part for the above script to work.

Bonus: schedule the creation of a tweet at a later date

Surprisingly enough, there is no way to schedule tweet creation as we can do on the web application. But we can do it ourselves with all the tools we already know. 😉

We will need a server to get tweet creation requests and schedule the operations, and a client to send information about the tweet. Let’s start with the server. Again I will use FastAPI but you can choose whatever you want.

FastAPI server to schedule tweet creation

Notes:

• We use the background scheduler this time which is appropriate for a web server.
• We use some pydantic validators to check our payload. We check that the creation_date is correct, the tweet text is correct according to Twitter rules, and for that, we use the twitter-text-parser library.
• We also add a custom method to check that the payload contains a text or media information.

Now for the client, we will write a command line interface using the click library. I have a tutorial if you want to learn more about it.

from datetime import datetime
from typing import Tuple

import click
import requests


@click.command('tweet')
@click.option('-t', '--text', help='optional text of the tweet')
@click.option(
    '-m',
    '--media-ids',
    multiple=True,
    help='optional media ids for tweet, can be multiple'
)
@click.option(
    '-T',
    '--tagged-user-ids',
    multiple=True,
    help='optional user ids tagged in the tweet, can be multiple'
)
@click.option(
    '-c', '--creation-date',
    type=click.DateTime(),
    required=True,
    help='creation date of the tweet, it must be in UTC timezone'
)
def create_tweet(text: str, media_ids: Tuple[str], tagged_user_ids: Tuple[str], creation_date: datetime):
    """A simple command to create tweet at a later date"""
    payload = {'text': text, 'creation_date': creation_date.isoformat()}
    if media_ids or tagged_user_ids:
        payload['media'] = {'media_ids': media_ids, 'tagged_user_ids': tagged_user_ids}
    # change the url with the url of your server
    print(payload)
    response = requests.post('http://localhost:8000/tweet', json=payload)
    if response.status_code == 200:
        click.secho('tweet creation was scheduled correctly', fg='green')
    else:
        click.secho('Error: ', nl=False, err=True, fg='red')
        click.secho(response.text, err=True, fg='red')

if __name__ == '__main__':
    create_tweet()

You can save this script in a file and run it with the python command. An example usage can look like this: python my_script.py --text "hello from bot" -c "2022-11-20 16:48:00".

This is the end of the tutorial, hope you enjoy reading it as I enjoy writing it. Take care of yourself and see you next time! 😁

If you like my article and want to continue learning with me, don’t hesitate to follow me here and subscribe to my newsletter on substack 😉