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
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.0
・ Notes for using Python Twitter Tools
・ Twitter User Stream substitute memo for Python
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)
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.
[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.
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
I'm just reading the file and executing the function.
bot.py
import bot
if __name__ == '__main__':
bot.start()
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
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).
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']
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