[PYTHON] Create a real-time auto-reply bot using the Twitter Streaming API

Introduction

I will write in detail about the Twitter API part of what I did in this article.

While developing the Twitter automatic reply bot, I could not find a reference article with the content such as "Getting tweets (replies) in real time" for the 2020 version, so I would like to summarize what I researched this time. I think.

Here are the specifications of the bot I made this time.

--Monitor the timeline in real time and automatically reply when you find a tweet for yourself that contains a specific word

--Automatic reply up to once a day per account

--Does not respond to tweets other than followers

--Does not respond to retweets

--Does not (cannot) respond to tweets from private accounts

--Be able to respond to your own tweets

Twitter API

Get Twitter API Key

As of August 2020, the method of obtaining the API key is slightly different, but I referred to this article such as how to apply for use. ・ Detailed explanation from the example sentence of the 2020 version Twitter API usage application to the acquisition of the API key

Once your application was approved, you did the following on the Developer Portal:

--Changed App permissions from Read to Read and Write (to tweet from the program)

--Generate Access Token & Secret (required to use API)

Twitter Streaming API

Apparently, the User Streams API was used to get tweets on the timeline in real time. However, it seems that this API was abolished in August 2018, and as a substitute, Statuses / filter of Filter realtime Tweets / tweets / filter-realtime / api-reference / post-statuses-filter) can be used.

For details, I referred to this article. ・ About Twitter Streaming API --kivantium activity diary

In the above article, it seems that tweepy is used, but I personally seemed to use the twitter library, so I used the twitter library. Install with the following command. (Run in virtual environment)

(venv)$ pip install twitter

For the twitter library, the following sites were helpful. ・ Twitter 1.18.0Notes for using Python Twitter ToolsTwitter User Stream substitute memo for Python

How to use twitter library

Preparing interface objects for Twitter API

import twitter

#Creating an OAuth authentication object
oauth = twitter.OAuth('Your Access token',
                      'Your Access token secret',
                      'Your API key',
                      'Your API secret key')

#Object for using Twitter REST API
twitter_api = twitter.Twitter(auth=oauth)
#Object for using Twitter Streaming API
twitter_stream = twitter.TwitterStream(auth=oauth)

Send a reply

API for sending tweets POST statuses / update Is a REST API, so use the twitter_api object prepared above.

text = 'test'

#For regular tweets
twitter_api.statuses.update(status=text)

#When replying to a specific tweet
reply_text = '@username' + '\n'
reply_text += text
twitter_api.statuses.update(status=reply_text, in_reply_to_status_id='ID of the tweet to reply')

When sending a reply, it seems that you must not only specify in_reply_to_status_id but also include @ username (@ + user name of the tweet to reply to) in the text.

Get the timeline in real time

[As above](# twitter-streaming-api), Streaming API is used, so use the twitter_stream object prepared above.

#Character string to be monitored
tracking_text = '@my_username test' #AND search with a half-width space.','OR search by delimiter

for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
    print(tweet)

If you do this, you'll get a tweet object tweet that contains both the" @my_username"and" test "strings. (For all tweets in the public account) Please check the contents. Use the information contained in thistweet` object to reply.

In addition, there is also a follow argument in the filter, and if you pass a"',' delimited user ID "to this, you can limit the users to be monitored. The reason why I didn't specify follow this time is that it will be an OR search instead of an AND search with the condition of track. In other words, the tweet of the user specified by follow will be acquired regardless of the content. In this specification, specifying track can narrow down the tweets to be acquired rather than specifying follow, so only track is specified.

Source code

Directory structure

This time, I made a management screen with flask to register and edit the reply text, so the structure is similar to flask. There is no problem even if you put them together in one file.

app/
├── .env             #Prepare Twitter API Key etc.
├── bot.py           #Executable file
└── bot/             #DB processing / service class omitted for reply text
    ├── __init__.py
    ├── config.py
    └── tweet.py

Executable file

I'm just reading the file and executing the function.

bot.py


import bot

if __name__ == '__main__':
    bot.start()

setting file

We are preparing an object to use the Twitter API by reading the environment variables.

config.py


import os
from os.path import join, dirname
from dotenv import load_dotenv

import twitter

dotenv_path = join(dirname(__file__), '../.env')
load_dotenv(dotenv_path)


oauth = twitter.OAuth(os.environ.get('ACCESS_TOKEN_KEY'),
                      os.environ.get('ACCESS_TOKEN_SECRET'),
                      os.environ.get('CONSUMER_KEY'),
                      os.environ.get('CONSUMER_SECRET'))

twitter_api = twitter.Twitter(auth=oauth)
twitter_stream = twitter.TwitterStream(auth=oauth)
SCREEN_NAME = os.environ.get('TWITTER_SCREEN_NAME') # @username your own username

Main processing

It monitors the timeline in real time and automatically replies to replies containing specific words. It would have been nice to put them together in this file, but I prepared another class to handle the retrieved Tweet object and separated the processing.

__init__.py


import logging
from bot.config import twitter_stream, SCREEN_NAME
from bot.tweet import Tweet


def start():

    #logging settings
    formatter = '%(levelname)s : %(asctime)s : %(message)s'
    logging.basicConfig(level=logging.INFO, format=formatter)

    REPLY_TEXT = 'Hello'

    replied_user_list = [] #Manage replying users in a list

    tracking_text = '@'+ SCREEN_NAME + 'Hello'
    for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
        tweet_obj = Tweet(tweet)

        #If the required key is not included
        required_keys_list = [
            'id_str',
            'text',
            'user'
        ]
        if not tweet_obj.has_required_keys(required_keys_list):
            logging.warning('FALSE->required key is empty')
            print(tweet_obj.tweet)
            continue

        #For retweets (tracking_If it matches the text, it will react)
        if tweet_obj.is_retweet():
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->is retweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #For users who replied in the past
        user_id = tweet_obj.get_user_id()
        if user_id in replied_user_list:
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->has already replied', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #For your own tweets
        if tweet_obj.is_reply_from_me():
            tweet_obj.reply(REPLY_TEXT)
            replied_user_list.append(user_id)
            logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS->self tweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #If not a follower's tweet
        if not tweet_obj.is_reply_from_follower():
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->not follwer', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #Normal system
        tweet_obj.reply(REPLY_TEXT)
        replied_user_list.append(user_id)
        logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())

This time, the explanation of the processing of the service class for reply text is omitted, so it is changed to the fixed reply text REPLY_TEXT. For one reply per day, click here (https://qiita.com/YiwaiY/items/ce790838725aba54c035#1%E3%82%A2%E3%82%AB%E3%82%A6%E3%83% B3% E3% 83% 88% E3% 81% AB% E3% 81% A4% E3% 81% 8D1% E6% 97% A51% E3% 83% AA% E3% 83% 97% E3% 83% A9% E3% 82% A4% E3% 82% 92% E3% 81% A9% E3% 81% 86% E5% AE% 9F% E8% A3% 85% E3% 81% 99% E3% 82% 8B% E3% Please refer to 81% 8B).

Class that handles tweet objects

The processing that uses the Twitter REST API and the condition judgment method are summarized here.

tweet.py


from bot.config import twitter_api, SCREEN_NAME


class Tweet():

    def __init__(self, tweet):
        self.tweet = tweet

    def has_required_keys(self, required_keys_list):
        for required_key in required_keys_list:
            if required_key not in self.tweet:
                return False
        return True

    def reply(self, text):
        status = '@' + self.get_user_screenname() + '\n'
        status += text
        return twitter_api.statuses.update(status=status, in_reply_to_status_id=self.tweet['id_str'])

    def is_retweet(self):
        return 'retweeted_status' in self.tweet

    def is_reply_from_me(self):
        return self.get_user_screenname() == SCREEN_NAME

    def is_reply_from_follower(self):
        followers = twitter_api.followers.ids(screen_name=SCREEN_NAME, count=5000) #ID array of int type followers
        return self.get_user_id() in followers['ids']

    def get_user_id(self):
        return self.tweet['user']['id']

    def get_user_name(self):
        return self.tweet['user']['name']

    def get_user_screenname(self):
        return self.tweet['user']['screen_name']

    def get_tweet_text(self):
        return self.tweet['text']

Summary

At first, I wanted to make an automatic reply bot on Twitter, and then I was researching various things, but I couldn't quite reach those articles, and I wondered if I couldn't make them anymore. It was my first experience to make something by myself, so I stumbled on various things, but I'm glad I completed it for the time being.

I'm still in my second year of programming, and I think the Python code of the guy who usually uses PHP isn't quite there, but I hope it helps someone who has troubles like me. ..

I introduced it at the beginning, but I would appreciate your favor in this article as well. ・ [First personal development] Deploying Flask app and Twitter auto-reply bot on Heroku

Recommended Posts

Create a real-time auto-reply bot using the Twitter Streaming API
Try using the Twitter API
Try using the Twitter API
Create a Twitter BOT with the GoogleAppEngine SDK for Python
Create a CRUD API using FastAPI
Create a graph using the Sympy module
Create an application using the Spotify API
Create a bot that boosts Twitter trends
Create a Twitter BOT service with GAE / P + Tweepy + RIOT API! (Part 1)
Create a Twitter BOT service with GAE / P + Tweepy + RIOT API! (Part 2)
Let's create a REST API using SpringBoot + MongoDB
Steps to create a Twitter bot with python
Create a GUI on the terminal using curses
I made a LINE BOT that returns a terrorist image using the Flickr API
Create a REST API using the model learned in Lobe and TensorFlow Serving.
Create a slack bot
Create a data collection bot in Python using Selenium
[CRUD] [Django] Create a CRUD site using the Python framework Django ~ 1 ~
Create a tweet heatmap with the Google Maps API
Post to your account using the API on Twitter
Create a pseudo REST API server using GitHub Pages
[CRUD] [Django] Create a CRUD site using the Python framework Django ~ 2 ~
[CRUD] [Django] Create a CRUD site using the Python framework Django ~ 3 ~
[CRUD] [Django] Create a CRUD site using the Python framework Django ~ 4 ~
[CRUD] [Django] Create a CRUD site using the Python framework Django ~ 5 ~
A little bit from Python using the Jenkins API
Create a dictionary by searching the table using sqlalchemy
Use Twitter API to reduce the time taken by Twitter (create a highlighting (like) timeline)
Try using the PeeringDB 2.0 API
Create a correlation diagram from the conversation history of twitter
Tweet Now Playing to Twitter using the Spotify API. [Python]
The story of creating a database using the Google Analytics API
Create an API that returns data from a model using turicreate
Export the contents acquired by Twitter Streaming API in JSON format
Create a pandas Dataflame by searching the DB table using sqlalchemy
I made a Twitter Bot with Go x Qiita API x Lambda
[Ev3dev] Create a program that captures the LCD (screen) using python
Create a TalkBot easily using Discord.py and A3RT's Talk API (pya3rt).
[LINE Messaging API] Create a BOT that connects with someone with Python
Create a REST API to operate dynamodb with the Django REST Framework
Create a python GUI using tkinter
Specifying the date with the Twitter API
Create a nested dictionary using defaultdict
Create API using hug with mod_wsgi
Access the Twitter API in Python
I tried using the checkio API
Create a C wrapper using Boost.Python
Create a LINE Bot in Django
[Python] Created a Twitter bot that generates friend-like tweets using Markov chains
Create a record with attachments in KINTONE using the Python requests module
Hanashi who became a read through BOT when trying to make a morphological analysis BOT using the popular LINE BOT API
Tweet the weather forecast with a bot
Create a bot to retweet coronavirus information
Try using the Wunderlist API in Python
Knowledge when making a bot using discord.py
I tried using Twitter api and Line api
A class that hits the DMM API
Try using the Kraken API in Python
[Python] Create a Batch environment using AWS-CDK
[Python] [LINE Bot] Create a parrot return LINE Bot
Create a dataframe from excel using pandas