Create a Twitter bot in python - part 2
Tags: python, bot, twitter, scheduling
Published: 2022-12-27 12:20:19

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](https://dev.to/le_woudar/create-a-twitter-bot-in-python-part-1-2o2n).

## 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](https://apscheduler.readthedocs.io/en/3.x/userguide.html): to schedule repetitive tasks.
- [emails](https://python-emails.readthedocs.io/en/latest/): to send emails.

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

{% embed https://gist.github.com/lewoudar/2033b1bc943457fd8430582054be2317 %}

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](https://github.com/mailhog/MailHog) service.
- Lines 34 to 47, to store the tweets in the file, I use the [JSON lines](https://jsonlines.org/) 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](https://docs.python.org/3/howto/logging.html#logging-basic-tutorial) 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](https://cht.sh/crontab) syntax. If you want to check the syntax of your crontab before applying it, you can use this [website](https://crontab.guru/).
- Line 74, our scheduler will run forever.

You can host your application for free on platforms like [fly.io](https://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](https://oauth.net/2/) vocabulary can be found here.
- `code`: a security value returned by the Twitter API.

Here is an example of a [FastAPI](https://fastapi.tiangolo.com/) server to set up quickly a redirect url.

```python
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](https://oauth.net/1/) 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](https://docs.tweepy.org/en/stable/authentication.html#tweepy.OAuth2UserHandler) class by filling in all the necessary information.

```python
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.

```python
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.

```python
client = tweepy.Client('access token')
```

Now you can create and retweet tweets as you wish:

```python
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:

```python
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](https://developer.twitter.com/en/docs/authentication/oauth-2-0/user-access-token) 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:

```python
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](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/interval.html#module-apscheduler.triggers.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`.

```python
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.

{% embed https://gist.github.com/lewoudar/08738c7029b9b9a2236af47142e97035 %}

Notes:

- We use the [background](https://apscheduler.readthedocs.io/en/3.x/modules/schedulers/background.html#apscheduler.schedulers.background.BackgroundScheduler) scheduler this time which is appropriate for a web server.
- We use some [pydantic validators](https://pydantic-docs.helpmanual.io/usage/validation_decorator/) 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](https://github.com/swen128/twitter-text-python) 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](https://click.palletsprojects.com/en/8.1.x/) library. I have a [tutorial](https://lewoudar.medium.com/click-a-beautiful-python-library-to-write-cli-applications-9c8154847066) if you want to learn more about it.

```python
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! 😁

---
This article was originally published on [Medium](https://lewoudar.medium.com/create-a-twitter-bot-in-python-part-2-501c1a3b2950).

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](https://lewoudar.substack.com/) 😉